# Copyright (c) ETH Zurich, SIS ID and HVL D-ITET
#
from time import sleep
from typing import cast
from hvl_ccb.configuration import configdataclass
from hvl_ccb.dev.base import (
DeviceError,
SingleCommDevice,
)
from .comm import LuminoxSerialCommunication
from .utils import (
LuminoxMeasurementType,
LuminoxOutputMode,
logger,
)
[docs]
class LuminoxError(DeviceError):
"""General Error for Luminox Device."""
[docs]
class LuminoxOutputModeError(LuminoxError):
"""Wrong output mode for requested data."""
[docs]
class LuminoxMeasurementTypeError(LuminoxError):
"""Wrong measurement type for requested data."""
[docs]
@configdataclass
class LuminoxConfig:
"""Configuration for the SST Luminox oxygen sensor."""
[docs]
class Luminox(SingleCommDevice):
"""Luminox oxygen sensor device class."""
def __init__(self, com, dev_config=None) -> None:
# Call superclass constructor
super().__init__(com, dev_config)
# Passes the device class to the communication protocol
# TODO: This should be moved higher up the inheritance chain with
# https://gitlab.com/ethz_hvl/hvl_ccb/-/merge_requests/379
self.com._device = self
self._mode: LuminoxOutputMode | None = None
[docs]
@staticmethod
def config_cls():
return LuminoxConfig
[docs]
@staticmethod
def default_com_cls():
return LuminoxSerialCommunication
[docs]
def start(self) -> None:
"""Starts the device. Opens the communication protocol."""
logger.info(f"Starting device {self}")
super().start()
[docs]
def stop(self) -> None:
"""Stops the device. Closes also the communication protocol."""
logger.info(f"Stopping device {self}")
self.com._stop_streaming_poller()
super().stop()
@property
def mode(self) -> LuminoxOutputMode | None:
"""
Sends command to sensor to request the current mode then calls _update_mode() to
update private variable _mode.
:return: Current mode read from _mode.
"""
with self.com._prepare_for_query():
self.com.write("M")
mode_update = self.com._get_mode()
self._mode = mode_update if mode_update is not False else self._mode
return self._mode
@mode.setter
def mode(self, mode: LuminoxOutputMode) -> None:
"""
Activates the selected output mode of the Luminox Sensor.
:param mode: Polling or streaming.
"""
# Tries to get the configuration for seconds to wait else default
wait = getattr(self.com.config, "wait_sec_mode_switching", 0.5)
with self.com._prepare_for_query():
self.com.write(f"M {mode.value}")
# needs a little bit of time to activate
sleep(wait)
mode_update = self.com._get_mode()
self._mode = mode_update if mode_update is not False else self._mode
if mode_update and self._mode == mode:
logger.info(
"Stream mode activation possible. "
f"{mode.name} mode activated {self}"
)
else:
err_msg = "Stream mode activation was not possible "
logger.error(err_msg)
raise LuminoxOutputModeError(err_msg)
@property
def percent_o2(self) -> float | None:
""":return: O₂ concentration (%)"""
value = self.com._get_measurement(
cast("LuminoxMeasurementType", LuminoxMeasurementType.PERCENT_O2)
)
return float(value) if value is not None else None
@property
def partial_pressure_o2(self) -> float | None:
""":return: Partial pressure of O₂ (mbar)."""
value = self.com._get_measurement(
cast("LuminoxMeasurementType", LuminoxMeasurementType.PARTIAL_PRESSURE_O2)
)
return float(value) if value is not None else None
@property
def temperature(self) -> float | None:
""":return: Internal sensor temperature (°C)."""
value = self.com._get_measurement(
cast("LuminoxMeasurementType", LuminoxMeasurementType.TEMPERATURE)
)
return float(value) if value is not None else None
@property
def barometric_pressure(self) -> int | None:
""":return: Ambient pressure (mbar)."""
value = self.com._get_measurement(
cast("LuminoxMeasurementType", LuminoxMeasurementType.BAROMETRIC_PRESSURE)
)
return int(value) if value is not None else None
@property
def sensor_status(self) -> int | None:
""":return: Raw sensor status code."""
value = self.com._get_measurement(
cast("LuminoxMeasurementType", LuminoxMeasurementType.SENSOR_STATUS)
)
return int(value) if value is not None else None
@property
def date_of_manufacture(self) -> str | None:
""":return: Date of manufacture."""
value = self.com._get_measurement(
cast("LuminoxMeasurementType", LuminoxMeasurementType.DATE_OF_MANUFACTURE)
)
return str(value) if value is not None else None
@property
def serial_number(self) -> str | None:
""":return: Device serial number."""
value = self.com._get_measurement(
cast("LuminoxMeasurementType", LuminoxMeasurementType.SERIAL_NUMBER)
)
return str(value) if value is not None else None
@property
def software_revision(self) -> str | None:
""":return: Software revision."""
value = self.com._get_measurement(
cast("LuminoxMeasurementType", LuminoxMeasurementType.SOFTWARE_REVISION)
)
return str(value) if value is not None else None