# Copyright (c) ETH Zurich, SIS ID and HVL D-ITET
#
import logging
import re
from enum import Enum
from typing import TypeAlias, cast
from hvl_ccb.utils.enum import ValueEnum
from hvl_ccb.utils.typing import Number
logger = logging.getLogger(__package__)
LuminoxMeasurementTypeValue: TypeAlias = Number | str
"""A typing hint for all possible LuminoxMeasurementType values as read in either
streaming mode or in a polling mode with `LuminoxMeasurementType.ALL_MEASUREMENTS`.
Beware: has to be manually kept in sync with `LuminoxMeasurementType` instances
`cast_type` attribute values.
"""
LuminoxMeasurementTypeDict: TypeAlias = dict[
"str | LuminoxMeasurementType", LuminoxMeasurementTypeValue
]
"""A typing hint for a dictionary holding LuminoxMeasurementType values. Keys are
allowed as strings because `LuminoxMeasurementType` is of a `StrEnumBase` type.
"""
[docs]
class LuminoxOutputMode(Enum):
"""output mode."""
STREAMING = 0
POLLING = 1
OFF = 2
[docs]
class LuminoxMeasurementType(ValueEnum, init="value cast_type value_re"): # type: ignore[call-arg]
"""
Measurement types for `LuminoxOutputMode.POLLING`.
The `ALL_MEASUREMENTS` type will read values for the actual measurement types
as given in `LuminoxOutputMode.all_measurements_types()`; it parses multiple
single values using regexp's for other measurement types, therefore, no regexp is
defined for this measurement type.
"""
PARTIAL_PRESSURE_O2 = "O", float, r"[0-9]{4}.[0-9]"
PERCENT_O2 = "%", float, r"[0-9]{3}.[0-9]{2}"
TEMPERATURE = "T", float, r"[+-][0-9]{2}.[0-9]"
BAROMETRIC_PRESSURE = "P", int, r"[0-9]{4}"
SENSOR_STATUS = "e", int, r"[0-9]{4}"
DATE_OF_MANUFACTURE = "# 0", str, r"[0-9]{5} [0-9]{5}"
SERIAL_NUMBER = "# 1", str, r"[0-9]{5} [0-9]{5}"
SOFTWARE_REVISION = "# 2", str, r"[0-9]{5}"
ALL_MEASUREMENTS = "A", str, None
[docs]
@classmethod
def all_measurements_types(cls) -> tuple["LuminoxMeasurementType", ...]:
"""
A tuple of `LuminoxMeasurementType` enum instances which are actual
measurements, i.e. not date of manufacture or software revision.
"""
return cast(
"tuple[LuminoxMeasurementType, ...]",
(
cls.PARTIAL_PRESSURE_O2,
cls.TEMPERATURE,
cls.BAROMETRIC_PRESSURE,
cls.PERCENT_O2,
cls.SENSOR_STATUS,
),
)
[docs]
def parse_read_measurement_value(
self, read_txt: str
) -> LuminoxMeasurementTypeDict | LuminoxMeasurementTypeValue:
if self is LuminoxMeasurementType.ALL_MEASUREMENTS:
return {
measurement: measurement._parse_single_measurement_value(read_txt)
for measurement in LuminoxMeasurementType.all_measurements_types()
}
return self._parse_single_measurement_value(read_txt)
def _parse_single_measurement_value(
self, read_txt: str
) -> LuminoxMeasurementTypeValue:
prefix = self.value.split(" ")[0]
pattern = f"{prefix} {self.value_re}"
parsed_data: list[str] = re.findall(pattern, read_txt)
if len(parsed_data) != 1:
self._parse_error(parsed_data)
parsed_measurement: str = parsed_data[0]
try:
parsed_value = self.cast_type(
# don't check for empty match - we know already that there is one
re.search(self.value_re, parsed_measurement).group() # type: ignore[union-attr]
)
except ValueError: # pragma: no cover
logger.exception("Wrong value at measurement value matching")
self._parse_error(parsed_data)
return parsed_value
def _parse_error(self, parsed_data: list[str]) -> None:
from .dev import LuminoxMeasurementTypeError
err_msg = (
f"Expected measurement value for {self.name.replace('_', ' ')} of type "
f'{self.cast_type}; instead tyring to parse: "{parsed_data}"'
)
logger.error(err_msg)
raise LuminoxMeasurementTypeError(err_msg)