-
Notifications
You must be signed in to change notification settings - Fork 8
How to add a new adapter
EdgeMining's architecture is based on two structural concepts, Hexagonal Architecture and Domain Driven Design, in order to clearly separate the business logic (domain and application layer) from infrastructural dependencies (database, external APIs, hardware control, user interfaces). For a better understanding of how the codebase is structured, what principles underlie the layered structure, and the meaning and utility of Ports and Adapters, I recommend reading this article by hgraca DDD, Hexagonal, Onion, Clean, CQRS, … How I put it all together.
There are different types of ports for each domain. EnergyMonitorPort for energy monitors, MinerControllerPort for miner controllers, or NotifierPort for notifiers. Start from the Port corresponding to the domain (the interface). You can find it at the domain path /edge_mining/domain/<domain>/ports.py. This class represents the starting point. The adapter must implement the described methods, respecting the function signatures (names and types of input and output parameters).
For example at path /edge_mining/domain/energy/ports.py we can find the port used to retrieve energy data for the energy domain:
class EnergyMonitorPort(ABC):
"""Port for the Energy Monitor."""
def __init__(self, energy_monitor_type: EnergyMonitorAdapter):
"""Initialize the Energy Monitor."""
self.energy_monitor_type = energy_monitor_type
@abstractmethod
def get_current_energy_state(self) -> Optional[EnergyStateSnapshot]:
"""Fetches the latest energy readings from the plant."""
raise NotImplementedErrorAdapters are grouped within the adapters folder and are divided into domain and infrastructure. If you are implementing a domain adapter (for example, a Provider for the home_load domain, an EnergyMonitor for the energy domain, or a Controller for the miner domain), it will go into the corresponding domain folder /edge_mining/adapters/domain/<domain>/controllers|providers|monitors/name_of_the_adapter.py.
For example, if you want to implement a Provider that supplies energy forecast data using the external HomeAssistant service, it can be placed inside the file named home_assistant_api.py at the path /edge_mining/adapters/domain/forecast/providers/home_assistant_api.py
The specific adapter implements the port (thus it inherits from it). In the adapter's constructor (__init__() function), all elements useful for the adapter's operation are inserted. For example, if the adapter needs a connection with the HomeAssistantAPI service, the constructor must request an instance of ServiceHomeAssistantAPI which will be injected and used later. Furthermore, if the adapter needs other parameters for its operation such as names of entities to retrieve, login credentials, or other values, these must be specified here. It is good practice to also request a logger instance in the constructor so that it can be used within the adapter for debugging operations.
For example, the GenericSocketHomeAssistantAPIMinerController adapter, which allows controlling a miner through the use of a generic smart socket controlled by HomeAssistant, requires in its constructor:
-
home_assistant: an instance of
ServiceHomeAssistantAPI. - entity_switch: a string as the name of the entity to request from HomeAssistant.
- entity_power: a string as the name of the entity that tracks the consumed power.
- unit_power: a string indicating the unit of measurement for the power used.
- logger: a reference to the system logger instance.
This is a snippet of GenericSocketHomeAssistantAPIMinerController class:
class GenericSocketHomeAssistantAPIMinerController(MinerControlPort):
"""Controls a miner via Home Assistant's entities of a smart socket."""
def __init__(
self,
home_assistant: ServiceHomeAssistantAPI,
entity_switch: str,
entity_power: str,
unit_power: str = "W",
logger: Optional[LoggerPort] = None,
):
# Initialize the HomeAssistant API Service
self.home_assistant = home_assistant
self.logger = logger
self.entity_switch = entity_switch
self.entity_power = entity_power
self.unit_power = unit_power.lower()These variables are requested by the constructor and saved as local variables, ready to be used by the adapter's internal methods.
The adapter should only publicly expose the methods provided by the port. Obviously, there may be other methods used to simplify the adapter's operations, but these must be implemented as private methods.
The adapter might also implement other classes, Factory and/or Builder, which facilitate its manipulation in more complex contexts, but this will be explained later in step 6.
Each adapter will have its own dedicated configuration class and some maps that associate the adapter type with a specific configuration class or indicate that the adapter requires an external service. The configuration acts as a container for all parameters that you want to be saved to the database and that are passed to the adapter's constructor (excluding references to external service instances like ServiceHomeAssistantAPI or the logger). The configuration class for the adapter should be added to the other domain classes at the path /edge_mining/shared/adapter_configs/<domain>.py. The maps, on the other hand, are dictionaries that provide a one-to-one association between the adapter type and the specific configuration class and report if the adapter type needs an external service, indicating the service type. The configuration maps for the adapter should be added to those already present in the file /edge_mining/shared/adapter_maps/<domain>.py.
For example, the configuration class for the GenericSocketHomeAssistantAPIMinerController adapter is called MinerControllerGenericSocketHomeAssistantAPIConfig and is located at /edge_mining/shared/adapter_configs/miner.py. It contains exactly the properties required by the adapter's constructor (excluding the HomeAssistantAPI and logger instances) and implements the following internal methods:
-
is_valid(): a validity check related to the adapter type that verifies if this configuration is suitable for the type of adapter it is being used for. NOTE: the next step will explain how to map a new adapter type. -
to_dict(): a method to convert the class into a dictionary object. -
from_dict(): a class method that allows converting a dictionary of values into an instance of the class.
This is a snippet of code from MinerControllerDummyConfig located at /edge_mining/shared/adapter_config/miner.py:
class MinerControllerDummyConfig(MinerControllerConfig):
"""
Miner controller configuration. It encapsulate the configuration parameters
to control a miner with dummy controller.
"""
initial_status: str = field(default="UNKNOWN")
power_max: float = field(default=3200.0)
hashrate_max: HashRate = field(default=HashRate(90, "TH/s"))
def is_valid(self, adapter_type: MinerControllerAdapter) -> bool:
"""
Check if the configuration is valid for the given adapter type.
For Dummy Miner Controller, it is always valid.
"""
return adapter_type == MinerControllerAdapter.DUMMY
def to_dict(self) -> dict:
"""Converts the configuration object into a serializable dictionary"""
return {**asdict(self)}
@classmethod
def from_dict(cls, data: dict):
"""Create a configuration object from a dictionary"""
return cls(**data)If the configuration are you implementing has only default python data types (str, bool, int, etc.) the to_dict() and from_dict() methods remain the same for all configurations (you can copy and leave them as they are), otherwise if you are using custom data types in your adapter configuration (e.g. HashRate, Watts, etc.) you need to implement the from_dict() method of the configuration adapter class in order to instantiate a configuration class with properties of the right type. While the is_valid() method only needs the adapter type to be updated.
For example, this is a snippet of code from EnergyMonitorDummySolarConfig class located at /edge_mining/shared/adapter_config/energy.py. This class has a max_consumption_power parameter which is of custom python type Watts and requires a reimplementation of the class's from_dict method.
class EnergyMonitorDummySolarConfig(EnergyMonitorConfig):
"""Energy monitor configuration"""
max_consumption_power: Watts = field(default=Watts(3200.0)) # Default max consumption power
def is_valid(self, adapter_type: EnergyMonitorAdapter) -> bool:
"""
Check if the configuration is valid for the given adapter type.
For Dummy Solar, it is always valid.
"""
return adapter_type == EnergyMonitorAdapter.DUMMY_SOLAR
def to_dict(self) -> dict:
"""Converts the configuration object into a serializable dictionary"""
return {**asdict(self)}
@classmethod
def from_dict(cls, data: dict):
"""Create a configuration object from a dictionary"""
max_consumption_power = Watts(data.get("max_consumption_power", 3200.0))
return EnergyMonitorDummySolarConfig(max_consumption_power=max_consumption_power)The configuration maps for this domain are located at /edge_mining/shared/adapter_maps/miner.py and are:
-
MINER_CONTROLLER_CONFIG_TYPE_MAP: maps the adapter type to a specific configuration class. For example, aMinerControllerAdapter.DUMMYadapter requires a configuration class of typeMinerControllerDummyConfig, or an adapter of typeMinerControllerAdapter.GENERIC_SOCKET_HOME_ASSISTANT_APIrequires a configuration class of typeMinerControllerGenericSocketHomeAssistantAPIConfig.MINER_CONTROLLER_CONFIG_TYPE_MAP: Dict[MinerControllerAdapter, Optional[type[MinerControllerConfig]]] = { MinerControllerAdapter.DUMMY: MinerControllerDummyConfig, MinerControllerAdapter.GENERIC_SOCKET_HOME_ASSISTANT_API: MinerControllerGenericSocketHomeAssistantAPIConfig, }
-
MINER_CONTROLLER_TYPE_EXTERNAL_SERVICE_MAP: maps the adapter type to the required external service type. For example, theMinerControllerAdapter.DUMMYadapter does not require any external service, while theMinerControllerAdapter.GENERIC_SOCKET_HOME_ASSISTANT_APIadapter requires the external service of typeExternalServiceAdapter.HOME_ASSISTANT_API.MINER_CONTROLLER_TYPE_EXTERNAL_SERVICE_MAP: Dict[MinerControllerAdapter, Optional[ExternalServiceAdapter]] = { MinerControllerAdapter.DUMMY: None, # Dummy does not use an external service MinerControllerAdapter.GENERIC_SOCKET_HOME_ASSISTANT_API: ExternalServiceAdapter.HOME_ASSISTANT_API, }
NOTE: The adapter configurations will have a Pydantic schema counterpart (see step 8).
Each domain maintains a map, in the form of a class, of the known adapter types. This map can be found at the path /edge_mining/domain/<domain>/common.py. Within the same domain, the names of the adapter types must be unique.
For example, the map representing the controllers for managing a miner in the miner domain is MinerControllerAdapter and is located at /edge_mining/domain/miner/common.py. This maps the known MinerController types like DUMMY, GENERIC_SOCKET_HOME_ASSISTANT_API, etc.
If you are creating a new adapter type, remember to create a new unique name that identifies it and update the is_valid() function of the configuration seen in the previous step. The adapter type is defined as an Enum type associated with a string and is present at the path /edge_mining/domain/<domain>/common.py for each domain.
Depending on the complexity of the adapter, it may be necessary to implement some design patterns to facilitate the manipulation of the adapters themselves during construction. Specifically, I am referring to the use of the Builder or Factory patterns. The implementation of a Factory class, in addition to providing a single creation interface shared among all adapters of the same domain (always using the same create() creation method), simplifies the creation of an adapter from domain objects already present in the system (for example, a miner or an energy_source already configured thanks to functions like from_<entity>()) and encapsulates business logic specific to the adapter (for example: verifying that the configuration provided to create the adapter is compatible with the adapter type). The implementation of a Builder class, on the other hand, facilitates the creation of a new instance of the adapter based on other domain objects in contexts where some objects may be absent (for example: the presence of an energy storage battery in the photovoltaic system or the connection to the national grid for energy exchange, elements that may not always be present in all systems).
The adapter creation classes that implement patterns like Factory and/or Builder should be placed within the same file where the main adapter class was implemented.
For example, the HomeAssistantForecastProviderBuilder class creates an instance of HomeAssistantForecastProvider from the energy forecast entities for the current hour set_actual_power_entity(), the next hour set_next_1h_power_entity(), or the next 12 hours set_next_12h_power_entity() only if these entities are provided to the system. Thanks to the builder, if these entities have not been configured, the created adapter will not request them. A similar situation applies to the HomeAssistantAPIEnergyMonitorBuilder.
It is also possible to implement both patterns, as was done for the HomeAssistantAPIEnergyMonitor adapter of the energy domain. This adapter indeed uses a HomeAssistantAPIEnergyMonitorBuilder within a HomeAssistantAPIEnergyMonitorFactory to manage the creation logic of the specific adapter instance and provide a common creation interface for all EnergyMonitor type adapters.
It is strongly recommended to implement the Factory class to encapsulate and manage the validation logic related to the configuration of the adapter you are developing and to add the Builder class only in cases where there are several parameters required by the adapter or particular business logic that requires it.
AdapterService is located at /edge_mining/application/services/adapter_service.py and is responsible for creating concrete instances of adapters (Controllers, EnergyMonitors, Providers, etc.) based on the adapter type, injecting elements like the adapter configuration (created in the previous step 4) or the reference to the external service (e.g., the ServiceHomeAssistantAPI instance) into the instance's constructor. Remember the adapter type map created in step 5? It is used here to initialize specific adapters based on the chosen type. AdapterService implements several private methods, one for each family of domain adapters, which are responsible for initializing the adapters. These initialization methods are, for example, _initialize_energy_monitor_adapter() to initialize energy monitors of the energy domain, _initialize_miner_controller_adapter() to initialize miner controllers of the miner domain, or _initialize_notifier_adapter() to initialize notifiers of the notify domain, and so on.
Thanks to the use of the Factory pattern, for the same domain, different Factory instances can be created (one for each specific adapter type) and then the create() creation method can be called once to create the adapter instance.
For example, the private method _initialize_energy_monitor_adapter() has a conditional block on the energy monitor adapter type energy_monitor.adapter_type, which could be of type EnergyMonitorAdapter.DUMMY_SOLAR or EnergyMonitorAdapter.HOME_ASSISTANT_API. If a DUMMY type energy monitor adapter has been configured, a DummySolarEnergyMonitorFactory will be created, passing a reference to the previously configured energy_source to the Factory using the factory's from_energy_source() method. If, instead, a HOME_ASSISTANT_API type energy monitor adapter has been configured in the system, a HomeAssistantAPIEnergyMonitorFactory will be created, without any further specific parameters passed to the constructor. Finally, the create() method is called, which returns an instance of the specific adapter.
Below is a snippet of how AdapterService uses the Factory pattern to create instances of different types of EnergyMonitor adapters using a single call to the create() function:
energy_monitor_adapter_factory: Optional[EnergyMonitorAdapterFactory] = None
if energy_monitor.adapter_type == EnergyMonitorAdapter.DUMMY_SOLAR:
# --- Dummy Solar ---
if not energy_source:
raise ValueError("EnergySource is required for DummySolar energy monitor.")
energy_monitor_adapter_factory = DummySolarEnergyMonitorFactory()
# Set energy source as reference
energy_monitor_adapter_factory.from_energy_source(energy_source)
elif energy_monitor.adapter_type == EnergyMonitorAdapter.HOME_ASSISTANT_API:
# --- Home Assistant API ---
if not energy_monitor.config:
raise ValueError("EnergyMonitor config is required for HomeAssistantAPI energy monitor.")
energy_monitor_adapter_factory = HomeAssistantAPIEnergyMonitorFactory()
# Actually HomeAssistantAPI Energy Monitor
# does not needs an energy source as reference
else:
raise ValueError(f"Unsupported energy monitor adapter type: {energy_monitor.adapter_type}")
instance = energy_monitor_adapter_factory.create(
config=energy_monitor.config,
logger=self.logger,
external_service=external_service,
)
return instanceTo manage the configuration of adapters from external sources (see step 9 for CLI), Pydantic models are created that contain the same parameters as the configuration classes, and objects that map the created schema one-to-one with the adapter's configuration class. The use of Pydantic offers several advantages such as automatic data validation, serialization, and JSON deserialization. The map, on the other hand, is a dictionary used by the system that reports the association of the configuration class with the schema. Schemas and maps can be found at the path /edge_mining/adapters/domain/<domain>/schemas.py.
For example, at the path /edge_mining/adapters/domain/energy/schemas.py you can find the Pydantic schemas for the energy domain. The EnergyMonitorHomeAssistantConfigSchema schema represents the schema for the EnergyMonitorHomeAssistantConfig adapter configuration and lists the parameters required by its configuration, default values, type, as well as implementing validation methods. The schema-configuration map, on the other hand, is represented by ENERGY_MONITOR_CONFIG_SCHEMA_MAP where it is indicated that the configuration for the EnergyMonitorHomeAssistantConfig energy monitoring adapter should use the EnergyMonitorHomeAssistantConfigSchema schema.
class EnergyMonitorHomeAssistantConfigSchema(BaseModel):
"""Schema for Home Assistant EnergyMonitorConfig."""
entity_production: str = Field(..., description="Home Assistant production entity")
entity_consumption: str = Field(..., description="Home Assistant consumption entity")
entity_grid: str = Field(default="", description="Home Assistant grid entity")
entity_battery_soc: str = Field(default="", description="Home Assistant battery SOC entity")
entity_battery_power: str = Field(default="", description="Home Assistant battery power entity")
entity_battery_remaining_capacity: str = Field(
default="", description="Home Assistant battery remaining capacity entity"
)
unit_production: str = Field(default="W", description="Production unit")
unit_consumption: str = Field(default="W", description="Consumption unit")
unit_grid: str = Field(default="W", description="Grid unit")
unit_battery_power: str = Field(default="W", description="Battery power unit")
unit_battery_remaining_capacity: str = Field(default="Wh", description="Battery remaining capacity unit")
grid_positive_export: bool = Field(default=False, description="Grid positive export direction")
battery_positive_charge: bool = Field(default=True, description="Battery positive charge direction")
@field_validator("entity_production", "entity_consumption")
@classmethod
def validate_required_entities(cls, v: str) -> str:
"""Validate that required entities are not empty."""
v = v.strip()
if not v:
raise ValueError("Required entity must not be empty")
return v
def to_model(self) -> EnergyMonitorHomeAssistantConfig:
"""Convert schema to EnergyMonitorHomeAssistantConfig domain entity."""
return EnergyMonitorHomeAssistantConfig(
entity_production=self.entity_production,
entity_consumption=self.entity_consumption,
entity_grid=self.entity_grid,
entity_battery_soc=self.entity_battery_soc,
entity_battery_power=self.entity_battery_power,
entity_battery_remaining_capacity=self.entity_battery_remaining_capacity,
unit_production=self.unit_production,
unit_consumption=self.unit_consumption,
unit_grid=self.unit_grid,
unit_battery_power=self.unit_battery_power,
unit_battery_remaining_capacity=self.unit_battery_remaining_capacity,
grid_positive_export=self.grid_positive_export,
battery_positive_charge=self.battery_positive_charge,
)
class Config:
"""Pydantic configuration."""
use_enum_values = True
validate_assignment = True
```
```python
ENERGY_MONITOR_CONFIG_SCHEMA_MAP: Dict[
type[EnergyMonitorConfig],
Union[type[EnergyMonitorDummySolarConfigSchema], type[EnergyMonitorHomeAssistantConfigSchema]],
] = {
EnergyMonitorDummySolarConfig: EnergyMonitorDummySolarConfigSchema,
EnergyMonitorHomeAssistantConfig: EnergyMonitorHomeAssistantConfigSchema,
}After implementing a configuration class for your adapter, remember to go to the reference domain and also create the associated Pydantic schemas and update the map.
The Command Line Interface has dedicated functions that guide the user in creating the configuration for each adapter with steps that depend on the adapter type. The functions for each domain dedicated to the CLI are located at the path /edge_mining/adapters/domain/<domain>/cli/commands.py.
For example, the configuration of an energy monitor adapter for the energy domain is handled by the handle_energy_monitor_configuration() function inside the file /edge_mining/adapters/domain/energy/cli/commands.py. If the adapter is of type EnergyMonitorAdapter.DUMMY_SOLAR, then the flow for configuring a dummy energy monitor will be executed (function handle_energy_monitor_dummy_solar_configuration()), whereas if the adapter is of type EnergyMonitorAdapter.HOME_ASSISTANT_API, the code for configuring an energy monitor that uses HomeAssistant will be executed (function handle_energy_monitor_home_assistant_configuration()).
After implementing your adapter, update the CLI configuration handler for the adapter of the relevant domain and add the functions for your specific configuration.