feature: add pyasic miner adapter

.vscode/settings.json

@@ -54,6 +54,7 @@
         "sheduler",
         "satoshi",
         "hashrate",
-        "homeassistant"
+        "homeassistant",
+        "pyasic"
     ]
 }

edge_mining/adapters/domain/miner/cli/commands.py

@@ -18,6 +18,7 @@
 from edge_mining.shared.adapter_configs.miner import (
     MinerControllerDummyConfig,
     MinerControllerGenericSocketHomeAssistantAPIConfig,
+    MinerControllerPyASICConfig,
 )
 from edge_mining.shared.adapter_maps.miner import MINER_CONTROLLER_TYPE_EXTERNAL_SERVICE_MAP
 from edge_mining.shared.external_services.entities import ExternalService
@@ -540,6 +541,20 @@ def handle_miner_controller_generic_socket_home_assistant_api_config(miner: Opti
     )
 
 
+def handle_miner_controller_pyasic_config(miner: Optional[Miner]) -> MinerControllerConfig:
+    """Handle configuration for the PyASIC Miner Controller."""
+    click.echo(click.style("\n--- PyASIC Miner Controller Configuration ---", fg="yellow"))
+
+    ip: str = click.prompt(
+        "IP address of the PyASIC miner (eg. 192.168.1.100)",
+        type=str,
+        default="192.168.1.100",
+    )
+    return MinerControllerPyASICConfig(
+        ip=ip,
+    )
+
+
 def handle_miner_controller_configuration(
     adapter_type: MinerControllerAdapter, miner: Optional[Miner]
 ) -> Optional[MinerControllerConfig]:
@@ -549,6 +564,8 @@ def handle_miner_controller_configuration(
         config = handle_miner_controller_dummy_config(miner)
     elif adapter_type.value == MinerControllerAdapter.GENERIC_SOCKET_HOME_ASSISTANT_API.value:
         config = handle_miner_controller_generic_socket_home_assistant_api_config(miner)
+    elif adapter_type.value == MinerControllerAdapter.PYASIC.value:
+        config = handle_miner_controller_pyasic_config(miner)
     else:
         click.echo(click.style("Unsupported controller type selected. Aborting.", fg="red"))
     return config

edge_mining/adapters/domain/miner/controllers/pyasic.py

@@ -0,0 +1,214 @@
+"""
+pyasic adapter (Implementation of Port)
+that controls a miner via pyasic.
+"""
+
+import asyncio
+from typing import Dict, Optional
+
+import pyasic
+from pyasic import AnyMiner
+from pyasic.device.algorithm.hashrate import AlgoHashRate
+
+from edge_mining.domain.common import Watts
+from edge_mining.domain.miner.common import MinerStatus
+from edge_mining.domain.miner.entities import Miner
+from edge_mining.domain.miner.exceptions import MinerControllerConfigurationError
+from edge_mining.domain.miner.ports import MinerControlPort
+from edge_mining.domain.miner.value_objects import HashRate
+from edge_mining.shared.adapter_configs.miner import MinerControllerPyASICConfig
+from edge_mining.shared.external_services.ports import ExternalServicePort
+from edge_mining.shared.interfaces.config import Configuration
+from edge_mining.shared.interfaces.factories import MinerControllerAdapterFactory
+from edge_mining.shared.logging.port import LoggerPort
+
+
+class PyASICMinerControllerAdapterFactory(MinerControllerAdapterFactory):
+    """
+    Create a factory for pyasic Miner Controller Adapter.
+    This factory is used to create instances of the adapter.
+    """
+
+    def __init__(self):
+        self._miner: Optional[Miner] = None
+
+    def from_miner(self, miner: Miner):
+        """Set the miner for this controller."""
+        self._miner = miner
+
+    def create(
+        self,
+        config: Optional[Configuration] = None,
+        logger: Optional[LoggerPort] = None,
+        external_service: Optional[ExternalServicePort] = None,
+    ) -> MinerControlPort:
+        """Create a miner controller adapter instance."""
+
+        if not isinstance(config, MinerControllerPyASICConfig):
+            raise MinerControllerConfigurationError("Invalid configuration for pyasic Miner Controller.")
+
+        # Get the config from the provided configuration
+        miner_controller_configuration: MinerControllerPyASICConfig = config
+
+        return PyASICMinerController(
+            ip=miner_controller_configuration.ip,
+            password=miner_controller_configuration.password,
+            logger=logger,
+        )
+
+
+class PyASICMinerController(MinerControlPort):
+    """Controls a miner via pyasic."""
+
+    def __init__(
+        self,
+        ip: str,
+        password: str | None = None,
+        logger: Optional[LoggerPort] = None,
+    ):
+        self.logger = logger
+
+        self.ip = ip
+        self.password = password
+
+        self._miner: Optional[AnyMiner] = None
+
+        self._log_configuration()
+
+        # Retrieve the pyasic miner instance
+        self._get_miner()
+
+    def _log_configuration(self):
+        if self.logger:
+            self.logger.debug(f"Entities Configured: IP={self.ip}")
+
+    def _get_miner(self) -> None:
+        """Retrieve the pyasic miner instance."""
+        if self._miner is None:
+            self._miner: AnyMiner = asyncio.run(pyasic.get_miner(self.ip))
+        if self._miner is not None and self.password is not None:
+            if self._miner.rpc is not None:
+                self._miner.rpc.pwd = self.password
+            if self._miner.web is not None:
+                self._miner.web.pwd = self.password
+
+    def get_miner_hashrate(self) -> Optional[HashRate]:
+        """
+        Gets the current hash rate, if available.
+        """
+
+        if self.logger:
+            self.logger.debug(f"Fetching hashrate from from {self.ip}...")
+
+        # Get pyasic miner instance
+        self._get_miner()
+
+        if not self._miner:
+            if self.logger:
+                self.logger.debug(f"Failed to retrieve miner instance from {self.ip}...")
+            return None
+
+        hashrate: Optional[AlgoHashRate] = asyncio.run(self._miner.get_hashrate())
+        if hashrate is None:
+            if self.logger:
+                self.logger.debug(f"Failed to fetch hashrate from {self.ip}...")
+            return None
+        real_hashrate = HashRate(value=float(hashrate), unit=str(hashrate.unit))
+
+        if self.logger:
+            self.logger.debug(f"Hashrate fetched: {real_hashrate}")
+
+        return real_hashrate
+
+    def get_miner_power(self) -> Optional[Watts]:
+        """Gets the current power consumption, if available."""
+        if self.logger:
+            self.logger.debug(f"Fetching power consumption from from {self.ip}...")
+
+        # Get pyasic miner instance
+        self._get_miner()
+
+        if not self._miner:
+            if self.logger:
+                self.logger.debug(f"Failed to retrieve miner instance from {self.ip}...")
+            return None
+
+        wattage = asyncio.run(self._miner.get_wattage())
+        if wattage is None:
+            if self.logger:
+                self.logger.debug(f"Failed to fetch power consumption from {self.ip}...")
+            return None
+        power_watts = Watts(wattage)
+
+        if self.logger:
+            self.logger.debug(f"Power consumption fetched: {power_watts}")
+
+        return power_watts
+
+    def get_miner_status(self) -> MinerStatus:
+        """Gets the current operational status of the miner."""
+        if self.logger:
+            self.logger.debug(f"Fetching miner status from {self.ip}...")
+
+        # Get pyasic miner instance
+        self._get_miner()
+
+        if not self._miner:
+            if self.logger:
+                self.logger.debug(f"Failed to retrieve miner instance from {self.ip}...")
+            return MinerStatus.UNKNOWN
+
+        mining_state = asyncio.run(self._miner.is_mining())
+
+        state_map: Dict[Optional[bool], MinerStatus] = {
+            True: MinerStatus.ON,
+            False: MinerStatus.OFF,
+            None: MinerStatus.UNKNOWN,
+        }
+
+        miner_status = state_map.get(mining_state, MinerStatus.UNKNOWN)
+
+        if self.logger:
+            self.logger.debug(f"Miner status fetched: {miner_status}")
+
+        return miner_status
+
+    def stop_miner(self) -> bool:
+        """Attempts to stop the specified miner. Returns True on success request."""
+        if self.logger:
+            self.logger.debug(f"Sending stop command to miner at {self.ip}...")
+
+        # Get pyasic miner instance
+        self._get_miner()
+
+        if not self._miner:
+            if self.logger:
+                self.logger.debug(f"Failed to retrieve miner instance from {self.ip}...")
+            return False
+
+        success = asyncio.run(self._miner.stop_mining())
+
+        if self.logger:
+            self.logger.debug(f"Stop command sent. Success: {success}")
+
+        return success or False
+
+    def start_miner(self) -> bool:
+        """Attempts to start the miner. Returns True on success request."""
+        if self.logger:
+            self.logger.debug(f"Sending start command to miner at {self.ip}...")
+
+        # Get pyasic miner instance
+        self._get_miner()
+
+        if not self._miner:
+            if self.logger:
+                self.logger.debug(f"Failed to retrieve miner instance from {self.ip}...")
+            return False
+
+        success = asyncio.run(self._miner.resume_mining())
+
+        if self.logger:
+            self.logger.debug(f"Start command sent. Success: {success}")
+
+        return success or False

edge_mining/adapters/domain/miner/schemas.py

@@ -1,5 +1,6 @@
 """Validation schemas for miner domain."""
 
+import ipaddress
 import uuid
 from typing import Dict, Optional, Union, cast
 
@@ -12,6 +13,7 @@
 from edge_mining.shared.adapter_configs.miner import (
     MinerControllerDummyConfig,
     MinerControllerGenericSocketHomeAssistantAPIConfig,
+    MinerControllerPyASICConfig,
 )
 from edge_mining.shared.adapter_maps.miner import MINER_CONTROLLER_CONFIG_TYPE_MAP
 from edge_mining.shared.interfaces.config import MinerControllerConfig
@@ -523,10 +525,43 @@ class Config:
         validate_assignment = True
 
 
+class MinerControllerPyASICConfigSchema(BaseModel):
+    """Schema for MinerControllerPyASICConfig."""
+
+    ip: str = Field(..., description="IP address of the PyASIC miner")
+
+    @field_validator("ip")
+    @classmethod
+    def validate_ip(cls, v: str) -> str:
+        """Validate that the value is a plausible IP address."""
+        v = v.strip()
+        if not v:
+            raise ValueError("IP address must be a non-empty string")
+        try:
+            ipaddress.ip_address(str(v))
+        except ValueError as e:
+            raise ValueError(f"Invalid IP address: {v}") from e
+        return v
+
+    def to_model(self) -> MinerControllerPyASICConfig:
+        """
+        Convert schema to MinerControllerPyASICConfig adapter configuration model instance.
+        """
+
+        return MinerControllerPyASICConfig(ip=self.ip)
+
+    class Config:
+        """Pydantic configuration."""
+
+        use_enum_values = True
+        validate_assignment = True
+
+
 MINER_CONTROLLER_CONFIG_SCHEMA_MAP: Dict[
     type[MinerControllerConfig],
     Union[type[MinerControllerDummyConfigSchema], type[MinerControllerGenericSocketHomeAssistantAPIConfigSchema]],
 ] = {
     MinerControllerDummyConfig: MinerControllerDummyConfigSchema,
     MinerControllerGenericSocketHomeAssistantAPIConfig: MinerControllerGenericSocketHomeAssistantAPIConfigSchema,
+    MinerControllerPyASICConfig: MinerControllerPyASICConfigSchema,
 }

edge_mining/application/services/adapter_service.py

@@ -4,18 +4,19 @@
 
 from typing import Dict, List, Optional, Union
 
-from edge_mining.adapters.domain.energy.monitors.dummy_solar import DummySolarEnergyMonitorFactory
-from edge_mining.adapters.domain.energy.monitors.home_assistant_api import HomeAssistantAPIEnergyMonitorFactory
-from edge_mining.adapters.domain.forecast.providers.dummy_solar import DummyForecastProviderFactory
-from edge_mining.adapters.domain.forecast.providers.home_assistant_api import HomeAssistantForecastProviderFactory
-from edge_mining.adapters.domain.home_load.providers.dummy import DummyHomeForecastProvider
+from edge_mining.adapters.domain.energy.dummy_solar import DummySolarEnergyMonitorFactory
+from edge_mining.adapters.domain.energy.home_assistant_api import HomeAssistantAPIEnergyMonitorFactory
+from edge_mining.adapters.domain.forecast.dummy_solar import DummyForecastProviderFactory
+from edge_mining.adapters.domain.forecast.home_assistant_api import HomeAssistantForecastProviderFactory
+from edge_mining.adapters.domain.home_load.dummy import DummyHomeForecastProvider
 from edge_mining.adapters.domain.miner.controllers.dummy import DummyMinerController
 from edge_mining.adapters.domain.miner.controllers.generic_socket_home_assistant_api import (
     GenericSocketHomeAssistantAPIMinerControllerAdapterFactory,
 )
-from edge_mining.adapters.domain.notification.notifiers.dummy import DummyNotifier
-from edge_mining.adapters.domain.notification.notifiers.telegram import TelegramNotifierFactory
-from edge_mining.adapters.domain.performance.trackers.dummy import DummyMiningPerformanceTracker
+from edge_mining.adapters.domain.miner.controllers.pyasic import PyASICMinerControllerAdapterFactory
+from edge_mining.adapters.domain.notification.dummy import DummyNotifier
+from edge_mining.adapters.domain.notification.telegram import TelegramNotifierFactory
+from edge_mining.adapters.domain.performance.dummy import DummyMiningPerformanceTracker
 from edge_mining.adapters.infrastructure.homeassistant.homeassistant_api import ServiceHomeAssistantAPIFactory
 from edge_mining.adapters.infrastructure.rule_engine.common import RuleEngineType
 from edge_mining.adapters.infrastructure.rule_engine.factory import RuleEngineFactory
@@ -47,6 +48,7 @@
     EnergyMonitorAdapterFactory,
     ExternalServiceFactory,
     ForecastAdapterFactory,
+    MinerControllerAdapterFactory,
 )
 from edge_mining.shared.logging.port import LoggerPort
 
@@ -249,7 +251,6 @@ def _initialize_miner_controller_adapter(
             return cached_instance
 
         # Retrieve the external service associated to the miner controller
-        external_service: Optional[ExternalServicePort] = None
         if miner_controller.external_service_id:
             external_service = self.get_external_service(miner_controller.external_service_id)
             if not external_service:
@@ -259,6 +260,7 @@ def _initialize_miner_controller_adapter(
                 )
 
         try:
+            miner_controller_factory: Optional[MinerControllerAdapterFactory] = None
             instance: Optional[MinerControlPort] = None
 
             if miner_controller.adapter_type == MinerControllerAdapter.DUMMY:
@@ -278,6 +280,17 @@ def _initialize_miner_controller_adapter(
 
                 miner_controller_factory.from_miner(miner)
 
+                instance = miner_controller_factory.create(
+                    config=miner_controller.config,
+                    logger=self.logger,
+                    external_service=external_service,
+                )
+            elif miner_controller.adapter_type == MinerControllerAdapter.PYASIC:
+                # --- PyASIC Controller ---
+                miner_controller_factory = PyASICMinerControllerAdapterFactory()
+
+                miner_controller_factory.from_miner(miner)
+
                 instance = miner_controller_factory.create(
                     config=miner_controller.config,
                     logger=self.logger,
@@ -329,7 +342,6 @@ def _initialize_notifier_adapter(self, notifier: Notifier) -> Optional[Notificat
             return cached_instance
 
         # Retrieve the external service associated to the notifier
-        external_service: Optional[ExternalServicePort] = None
         if notifier.external_service_id:
             external_service = self.get_external_service(notifier.external_service_id)
             if not external_service: