"""Abstract base class defining the interface for all Oizom sensor wrappers.
Every sensor wrapper in :mod:`OzWrapper` must inherit from :class:`GenericSensor`
and implement its four abstract methods. This ensures a consistent lifecycle
across all sensor types:
1. :meth:`initialize` -- receive device configuration from the Gateway
2. :meth:`initializeSensor` -- set up individual hardware sensors
3. :meth:`getSensorReading` -- perform a measurement cycle and return data
4. :meth:`putSensorValue` -- merge sensor readings into the outbound payload
The orchestrator :class:`Sensor.Sensor` calls these methods in order during
its ``setup()`` and ``loop()`` phases.
Typical usage::
from SensorBase import GenericSensor
class OzTemp(GenericSensor):
def initialize(self, config, init_value):
...
def initializeSensor(self, sensor):
...
def getSensorReading(self):
return {"temp": 25.3, "hum": 60.0}
def putSensorValue(self, value):
value.update(self.getSensorReading())
return value
"""
from abc import ABCMeta, abstractmethod
from typing import ClassVar
from Configuration import Config
[docs]
class GenericSensor(metaclass=ABCMeta):
"""Abstract base class for all Oizom sensor wrappers.
Provides a shared :attr:`baseConfig` loaded from ``Configuration/config.json``
and defines the four-method lifecycle contract that every wrapper must fulfill.
Subclasses are expected to set :attr:`partNumber` and :attr:`parameter` to
their respective Gateway-assigned identifiers so the orchestrator can match
incoming configuration entries to the correct wrapper.
Attributes:
baseConfig: Base device configuration dictionary loaded from
``Configuration/config.json``. Shared across all wrapper instances.
partNumber: Gateway part-number identifier used to match this wrapper
to entries in the device configuration. Subclasses must override.
parameter: Primary measurement parameter key (e.g., ``"pm"``, ``"temp"``).
Subclasses must override.
"""
baseConfig: ClassVar[dict] = {}
partNumber = "pn"
parameter = "pm"
[docs]
def __init__(self) -> None:
config = Config()
self.baseConfig = config.config
[docs]
@abstractmethod
def initialize(self, config: dict, init_value: dict) -> None:
"""Receive and store the device configuration from the Gateway.
Called once during :meth:`Sensor.Sensor.setup` after the Gateway
responds with the device configuration. Implementations should parse
the ``config`` dictionary to determine which hardware sensors are
enabled and store any initial calibration values.
Args:
config: Device configuration dictionary from the Gateway containing
sensor enable flags, calibration offsets, and part numbers.
init_value: Initial sensor values dictionary, typically containing
the last known readings before shutdown for continuity.
"""
pass
[docs]
@abstractmethod
def initializeSensor(self, sensor: dict) -> None:
"""Initialize a single hardware sensor based on its configuration entry.
Called once per enabled sensor during :meth:`initialize`. Implementations
should instantiate the appropriate driver, configure I2C/UART/SPI
communication, and verify the sensor is responding.
Args:
sensor: Single sensor configuration entry from the Gateway, containing
at minimum ``"pn"`` (part number) and bus/address information.
"""
pass
[docs]
@abstractmethod
def getSensorReading(self) -> dict:
"""Perform a measurement cycle and return the latest sensor data.
Called once per sensing loop iteration by the orchestrator. Implementations
should read from all initialized hardware sensors, apply any calibration
or averaging, and return a flat dictionary of parameter-value pairs.
Returns:
Dictionary mapping parameter keys (e.g., ``"temp"``, ``"hum"``,
``"pm2_5"``) to their measured float values. Returns zero-filled
values on sensor read failure rather than raising.
"""
pass
[docs]
@abstractmethod
def putSensorValue(self, value: dict) -> dict:
"""Merge this wrapper's sensor readings into the outbound data payload.
Called by the orchestrator to assemble the final data packet that gets
posted to the Gateway. Implementations should call :meth:`getSensorReading`
and merge the result into ``value``.
Args:
value: Accumulated data payload dictionary being built by the
orchestrator across all active wrappers.
Returns:
The ``value`` dictionary with this wrapper's readings merged in.
"""
pass
[docs]
def putInitvalues(self, value: dict) -> dict:
"""Merge initial/default values into the data payload.
Override in subclasses that need to inject startup defaults or
last-known values into the first data payload after boot.
Args:
value: Accumulated data payload dictionary.
Returns:
The ``value`` dictionary, unmodified by default.
"""
return value