Skip to content
Open
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
3 changes: 3 additions & 0 deletions src/lib/sensors/__init__.py
Original file line number Diff line number Diff line change
Expand Up @@ -9,6 +9,7 @@
from lib.sensors.PMSA003I import PMSA003I
from lib.sensors.sensor_module import SensorModule
from lib.sensors.file_logging import FileLogger
from lib.sensors.internal_metrics import InternalMetrics
from lib.sensors.alarm import Alarm
from lib.displays.display import Display
from lib.networking import WirelessNetwork
Expand All @@ -32,6 +33,8 @@ def __init__(self, i2c: I2C, display: Display, wifi: WirelessNetwork, space_stat
self.file_logger = FileLogger(init_files=True)
self.load_modules(self.SENSOR_MODULES)
self._configure_modules()
self.internal_metrics = InternalMetrics(self.space_state)
self.configured_modules['InternalMetrics'] = self.internal_metrics
self.alarm = Alarm(self.display, self.space_state)
if CO2_ALARM_THRESHOLD_PPM > 0 and 'SCD30' in self.configured_modules:
self.alarm.enable()
Expand Down
39 changes: 39 additions & 0 deletions src/lib/sensors/internal_metrics.py
Original file line number Diff line number Diff line change
@@ -0,0 +1,39 @@
from lib.relay_history import RTCUnreliableError
from lib.sensors.sensor_module import SensorModule
from lib.space_state import SpaceState
from lib.ulogging import uLogger

class InternalMetrics(SensorModule):
"""
Reports internal SMIBHID metrics as sensor readings, rather than
readings from an external, physically attached, I2C sensor.
"""

def __init__(self, space_state: SpaceState) -> None:
super().__init__([{"name": "relay_on_time", "unit": "s"}])
self.log = uLogger("InternalMetrics")
self.space_state = space_state

def _get_relay_on_time(self) -> float | None:
"""
Return cumulative relay active seconds from relay history, or None
if relay history is not available or not enabled.
"""
relay_history = getattr(self.space_state, "relay_history", None)
if relay_history is None:
self.log.info("Relay history not available - no relay on time to report")
return None

try:
return relay_history.get_total_active_seconds(bool(self.space_state._last_relay_state))
except RTCUnreliableError:
self.log.info("System clock not yet reliable - no relay on time to report")
return None

def get_reading(self) -> dict[str, float | None]:
"""
Get internal metrics reading in SMIBHID format.
"""
return {
"relay_on_time": self._get_relay_on_time()
}