# Copyright (c) ETH Zurich, SIS ID and HVL D-ITET
#
import threading
from contextlib import contextmanager
from time import sleep
from typing import (
Any,
cast,
)
from hvl_ccb.comm.base import SyncCommunicationProtocol
from hvl_ccb.comm.serial import (
SerialCommunication,
SerialCommunicationBytesize,
SerialCommunicationConfig,
SerialCommunicationParity,
SerialCommunicationStopbits,
)
from hvl_ccb.configuration import configdataclass
from hvl_ccb.utils.poller import Poller
from hvl_ccb.utils.typing import Number
from hvl_ccb.utils.validation import validate_number
from .utils import (
LuminoxMeasurementType,
LuminoxMeasurementTypeDict,
LuminoxMeasurementTypeValue,
LuminoxOutputMode,
logger,
)
[docs]
@configdataclass
class LuminoxSerialCommunicationConfig(SerialCommunicationConfig):
#: Baudrate for SST Luminox is 9600 baud
baudrate: int = 9600
#: SST Luminox does not use parity
parity: str | SerialCommunicationParity = SerialCommunicationParity.NONE
#: SST Luminox does use one stop bit
stopbits: int | SerialCommunicationStopbits = SerialCommunicationStopbits.ONE
#: One byte is eight bits long
bytesize: int | SerialCommunicationBytesize = SerialCommunicationBytesize.EIGHTBITS
#: The terminator is CR LF
terminator: bytes = b"\r\n"
#: use 3 seconds timeout as default
timeout: Number = 3
wait_sec_mode_switching: Number = 0.5
wait_sec_update_mode_retry: Number = 0.1
trials_update_mode_retry: int = 5
[docs]
def clean_values(self) -> None:
super().clean_values()
validate_number(
"Wait time (sec)",
self.wait_sec_mode_switching,
(0, None),
closed_interval=False,
logger=logger,
)
validate_number(
"Re-try wait time (sec)",
self.wait_sec_update_mode_retry,
(0, None),
closed_interval=False,
logger=logger,
)
validate_number(
"Trials for mode activation",
self.trials_update_mode_retry,
(0, None),
int,
closed_interval=False,
logger=logger,
)
[docs]
class LuminoxSerialCommunication(SerialCommunication, SyncCommunicationProtocol):
"""
Specific communication protocol implementation for the SST Luminox oxygen sensor.
Already predefines device-specific protocol parameters in config.
"""
def __init__(self, configuration) -> None:
super().__init__(configuration)
# Stores a reference to the device class this is a communication protocol of
# TODO: This should be moved higher up the inheritance chain with
# https://gitlab.com/ethz_hvl/hvl_ccb/-/merge_requests/379
self._device: Any | None = None
# Internal buffer for latest streaming measurements
self._latest_measurements: LuminoxMeasurementTypeDict | None = None
# Lock to protect buffer access
self._buffer_lock = threading.Lock()
# Poller for continuous reading in streaming mode
self._streaming_poller: Poller = Poller(
spoll_handler=self._streaming_read_handler,
polling_interval_sec=1.0, # Luminox provides one update per second
)
[docs]
@staticmethod
def config_cls():
return LuminoxSerialCommunicationConfig
[docs]
def open(self) -> None:
super().open()
# Read out buffer to get rid of any garbage from previous session
logger.debug(self.read())
def _streaming_read_handler(self):
"""Poller handler for continuously reading streaming data."""
try:
raw = self.read().rstrip(self.config.terminator_str())
parsed = cast(
"LuminoxMeasurementTypeDict",
cast(
"LuminoxMeasurementType", LuminoxMeasurementType.ALL_MEASUREMENTS
).parse_read_measurement_value(raw),
)
with self._buffer_lock:
self._latest_measurements = parsed
except (ValueError, TypeError) as e:
logger.warning(f"Streaming read failed: {e}")
def _stop_streaming_poller(self):
"""Stops the streaming poller if it's running."""
if self._streaming_poller.is_polling():
self._streaming_poller.stop_polling()
logger.info("Stopped Luminox streaming poller")
def _resume_streaming_poller(self):
"""Resumes the streaming poller if it's not running."""
if not self._streaming_poller.is_polling():
self._streaming_poller.start_polling()
logger.info("Resumed Luminox streaming poller")
@contextmanager
def _prepare_for_query(self):
"""
Context manager that acquires lock and prepares for sending a query,
subsequently resumes previous state.
"""
with self.access_lock:
self._stop_streaming_poller()
if hasattr(self, "_serial_port") and self._serial_port.is_open:
self._serial_port.reset_input_buffer()
yield
if self._device._mode == LuminoxOutputMode.STREAMING:
# Start continuous background reading if not already running
self._resume_streaming_poller()
else:
# Stop streaming if active and clear cached data
self._stop_streaming_poller()
with self._buffer_lock:
self._latest_measurements = None
def _get_mode(self) -> LuminoxOutputMode | bool:
"""
Tries to read the mode from the device output, for the configured number of
trials, and returns it if successful.
:return: mode if reading of mode possible else false.
"""
# Tries to get the configuration for number of trials else default
trials = getattr(self.config, "trials_update_mode_retry", 1)
# Tries to get the configuration for seconds to wait else default
wait = getattr(self.config, "wait_sec_update_mode_retry", 0)
for trial in range(trials):
msg = self.read().strip("\x00\n\r\t ")
logger.debug(f"Trial {trial + 1}: Received message: '{msg}'")
if msg in ("M 00", "M 01", "M 02"):
mode = LuminoxOutputMode(int(msg.removeprefix("M 0")))
logger.info(
f"Reading of mode possible in: trial {trial + 1} out of {trials}"
)
return mode
if trial + 1 == trials:
logger.info(f"Reading of mode not possible after: {trials} trials")
else:
sleep(wait)
return False
def _get_measurement(
self, measurement_type: LuminoxMeasurementType
) -> LuminoxMeasurementTypeValue | None:
"""
:return: Requested data depending on the data type and current mode.
- If it is a measurement and streaming mode → read from cached buffer.
- If it is metadata or in polling mode → query the device directly.
"""
mode = getattr(self._device, "_mode", LuminoxOutputMode.STREAMING)
polling = bool(mode == LuminoxOutputMode.POLLING)
metadata = bool(
measurement_type
in (
LuminoxMeasurementType.DATE_OF_MANUFACTURE,
LuminoxMeasurementType.SERIAL_NUMBER,
LuminoxMeasurementType.SOFTWARE_REVISION,
)
)
raw = None
if polling:
raw = self.query(measurement_type.value)
elif metadata:
with self._prepare_for_query():
raw = self.query(measurement_type.value)
if polling or metadata:
if raw is None:
return None
parsed = measurement_type.parse_read_measurement_value(raw)
return parsed.get(measurement_type) if isinstance(parsed, dict) else parsed
with self._buffer_lock:
if self._latest_measurements is None:
logger.debug("No streaming data available yet.")
return None
return self._latest_measurements.get(measurement_type)