Skip to content

Handle sensor state and thermoBox safety off #202

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
bbx-a merged 1 commit into from
Aug 12, 2026
Merged

Handle sensor state and thermoBox safety off #202

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
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
30 changes: 29 additions & 1 deletion blebox_uniapi/box_types.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -191,6 +191,7 @@ def get_latest_api_level(product_type: str) -> Union[dict, int]:
"minimum": "thermo.minimumTemp",
"maximum": "thermo.maximumTemp",
"temperature": lambda x: f"sensors[?id == `{x}`]|[0]|value",
"temperature.state": lambda x: f"sensors[?id == `{x}`]|[0]|state",
"state": "thermo.state",
"mode": "thermo.mode",
"safetySensorId": "thermo.safetyTempSensor.sensorId",
Expand All @@ -203,6 +204,7 @@ def get_latest_api_level(product_type: str) -> Union[dict, int]:
"temperature",
{
"temperature": lambda x: f"sensors[?id == `{x}`]|[0]|value",
"temperature.state": lambda x: f"sensors[?id == `{x}`]|[0]|state",
},
]
],
Expand Down Expand Up @@ -530,7 +532,7 @@ def get_latest_api_level(product_type: str) -> Union[dict, int]:
{
"temperature": "tempSensor.sensors[?id == `0`]|[0]|value",
"trend": "tempSensor.sensors[?id == `0`]|[0]|trend",
"state": "tempSensor.sensors[?id == `0`]|[0]|state",
"temperature.state": "tempSensor.sensors[?id == `0`]|[0]|state",
"elapsed": "tempSensor.sensors[?id == `0`]|[0]|elapsedTimeS",
},
]
Expand Down Expand Up @@ -682,7 +684,9 @@ def get_latest_api_level(product_type: str) -> Union[dict, int]:
"multiSensor",
{
"temperature": lambda x: f"multiSensor.sensors[?id == `{x}`]|[0]|value",
"temperature.state": lambda x: f"multiSensor.sensors[?id == `{x}`]|[0]|state",
"wind": lambda x: f"multiSensor.sensors[?id == `{x}`]|[0]|value",
"wind.state": lambda x: f"multiSensor.sensors[?id == `{x}`]|[0]|state",
},
]
],
Expand All @@ -704,8 +708,11 @@ def get_latest_api_level(product_type: str) -> Union[dict, int]:
"multiSensor",
{
"temperature": lambda x: f"multiSensor.sensors[?id == `{x}`]|[0]|value",
"temperature.state": lambda x: f"multiSensor.sensors[?id == `{x}`]|[0]|state",
"wind": lambda x: f"multiSensor.sensors[?id == `{x}`]|[0]|value",
"wind.state": lambda x: f"multiSensor.sensors[?id == `{x}`]|[0]|state",
"humidity": lambda x: f"multiSensor.sensors[?id == `{x}`]|[0]|value",
"humidity.state": lambda x: f"multiSensor.sensors[?id == `{x}`]|[0]|state",
},
]
],
Expand All @@ -727,9 +734,13 @@ def get_latest_api_level(product_type: str) -> Union[dict, int]:
"multiSensor",
{
"illuminance": lambda x: f"multiSensor.sensors[?id == `{x}`]|[0]|value",
"illuminance.state": lambda x: f"multiSensor.sensors[?id == `{x}`]|[0]|state",
"temperature": lambda x: f"multiSensor.sensors[?id == `{x}`]|[0]|value",
"temperature.state": lambda x: f"multiSensor.sensors[?id == `{x}`]|[0]|state",
"wind": lambda x: f"multiSensor.sensors[?id == `{x}`]|[?type == 'wind']|[0]|value",
"wind.state": lambda x: f"multiSensor.sensors[?id == `{x}`]|[?type == 'wind']|[0]|state",
"humidity": lambda x: f"multiSensor.sensors[?id == `{x}`]|[0]|value",
"humidity.state": lambda x: f"multiSensor.sensors[?id == `{x}`]|[0]|state",
},
]
],
Expand All @@ -751,22 +762,39 @@ def get_latest_api_level(product_type: str) -> Union[dict, int]:
"multiSensor",
{
"frequency": lambda x: f"multiSensor.sensors[?id == `{x}`]|[?type == 'frequency']|[0]|value",
"frequency.state": lambda x: f"multiSensor.sensors[?id == `{x}`]|[?type == 'frequency']|[0]|state",
"current": lambda x: f"multiSensor.sensors[?id == `{x}`]|[?type == 'current']|[0]|value",
"current.state": lambda x: f"multiSensor.sensors[?id == `{x}`]|[?type == 'current']|[0]|state",
"voltage": lambda x: f"multiSensor.sensors[?id == `{x}`]|[?type == 'voltage']|[0]|value",
"voltage.state": lambda x: f"multiSensor.sensors[?id == `{x}`]|[?type == 'voltage']|[0]|state",
"apparentPower": lambda x: f"multiSensor.sensors[?id == `{x}`]|[?type == 'apparentPower']|[0]|value",
"apparentPower.state": lambda x: f"multiSensor.sensors[?id == `{x}`]|[?type == 'apparentPower']|[0]|state",
"activePower": lambda x: f"multiSensor.sensors[?id == `{x}`]|[?type == 'activePower']|[0]|value",
"activePower.state": lambda x: f"multiSensor.sensors[?id == `{x}`]|[?type == 'activePower']|[0]|state",
"reactivePower": lambda x: f"multiSensor.sensors[?id == `{x}`]|[?type == 'reactivePower']|[0]|value",
"reactivePower.state": lambda x: f"multiSensor.sensors[?id == `{x}`]|[?type == 'reactivePower']|[0]|state",
"reverseActiveEnergy": lambda x: f"multiSensor.sensors[?id == `{x}`]|[?type == 'reverseActiveEnergy']|[0]|value",
"reverseActiveEnergy.state": lambda x: f"multiSensor.sensors[?id == `{x}`]|[?type == 'reverseActiveEnergy']|[0]|state",
"forwardActiveEnergy": lambda x: f"multiSensor.sensors[?id == `{x}`]|[?type == 'forwardActiveEnergy']|[0]|value",
"forwardActiveEnergy.state": lambda x: f"multiSensor.sensors[?id == `{x}`]|[?type == 'forwardActiveEnergy']|[0]|state",
"forwardReactiveEnergy": lambda x: f"multiSensor.sensors[?id == `{x}`]|[?type == 'forwardReactiveEnergy']|[0]|value",
"forwardReactiveEnergy.state": lambda x: f"multiSensor.sensors[?id == `{x}`]|[?type == 'forwardReactiveEnergy']|[0]|state",
"reverseReactiveEnergy": lambda x: f"multiSensor.sensors[?id == `{x}`]|[?type == 'reverseReactiveEnergy']|[0]|value",
"reverseReactiveEnergy.state": lambda x: f"multiSensor.sensors[?id == `{x}`]|[?type == 'reverseReactiveEnergy']|[0]|state",
"illuminance": lambda x: f"multiSensor.sensors[?id == `{x}`]|[?type == 'illuminance']|[0]|value",
"illuminance.state": lambda x: f"multiSensor.sensors[?id == `{x}`]|[?type == 'illuminance']|[0]|state",
"temperature": lambda x: f"multiSensor.sensors[?id == `{x}`]|[?type == 'temperature']|[0]|value",
"temperature.state": lambda x: f"multiSensor.sensors[?id == `{x}`]|[?type == 'temperature']|[0]|state",
"wind": lambda x: f"multiSensor.sensors[?id == `{x}`]|[?type == 'wind']|[0]|value",
"wind.state": lambda x: f"multiSensor.sensors[?id == `{x}`]|[?type == 'wind']|[0]|state",
"humidity": lambda x: f"multiSensor.sensors[?id == `{x}`]|[?type == 'humidity']|[0]|value",
"humidity.state": lambda x: f"multiSensor.sensors[?id == `{x}`]|[?type == 'humidity']|[0]|state",
"openStatus": lambda x: f"multiSensor.sensors[?id == `{x}`]|[?type == 'openStatus']|[0]|value",
"openStatus.state": lambda x: f"multiSensor.sensors[?id == `{x}`]|[?type == 'openStatus']|[0]|state",
"co2": lambda x: f"multiSensor.sensors[?id == `{x}`]|[?type == 'co2']|[0]|value",
"co2.state": lambda x: f"multiSensor.sensors[?id == `{x}`]|[?type == 'co2']|[0]|state",
"co2Definition": lambda x: f"multiSensor.sensors[?id == `{x}`]|[?type == 'co2Definition']|[0]|value",
"co2Definition.state": lambda x: f"multiSensor.sensors[?id == `{x}`]|[?type == 'co2Definition']|[0]|state",
},
]
],
Expand Down
45 changes: 38 additions & 7 deletions blebox_uniapi/climate.py
View file
Open in desktop
Original file line number Diff line number Diff line change
@@ -1,9 +1,19 @@
from enum import IntEnum
from .sensor import Temperature
from typing import Optional, Any, Union
from .feature import Feature
from blebox_uniapi.jfollow import follow


class BleboxThermoState(IntEnum):
"""Possible states of a thermo regulator, as reported by the device."""

OFF = 0
ON = 1
BOOST = 3
SAFETY_OFF = 4


class Climate(Temperature):
_is_on: Optional[bool]
_desired: Union[float, int, None]
Expand Down Expand Up @@ -90,12 +100,18 @@ def _read_is_on(self) -> Optional[bool]:
raw = self.raw_value("state")
if raw is not None: # no reading
alias = self._alias
return product.expect_int(alias, raw, 3, 0) in (
1,
3,
) # 1: On, 3: Boost(thermoBox max temp of mode)
return product.expect_int(alias, raw, max(BleboxThermoState), 0) in (
BleboxThermoState.ON,
BleboxThermoState.BOOST,
)
return None

def _read_safety_off(self) -> bool:
"""Return True when the safety system forced the output off."""
if self._product.last_data is None:
return False
return self.raw_value("state") == BleboxThermoState.SAFETY_OFF

def _read_operating_state(self) -> Optional[int]:
"""Return current operating state"""
if self._product.last_data is not None:
Expand All @@ -107,13 +123,13 @@ def _read_operating_state(self) -> Optional[int]:
return None

def _read_is_heating(self) -> Optional[bool]:
if not self._product.last_data:
if not self._product.last_data or self.current is None or self.desired is None:
return None

return self.is_on and (self.current < self.desired)

def _read_is_cooling(self) -> Optional[bool]:
if not self._product.last_data:
if not self._product.last_data or self.current is None or self.desired is None:
return None

return self.is_on and (self.current > self.desired)
Expand All @@ -131,7 +147,22 @@ def _read_mode(self) -> Optional[int]:
def after_update(self) -> None:
self._is_on = self._read_is_on()
self._desired = self._read_temperature("desired")
self._current = self._read_temperature("temperature")

state = self._read_state("temperature")
if self._state_is_error(state):
self._error = True
self._current = None
else:
self._error = False
if self._state_is_initializing(state):
self._current = None
else:
self._current = self._read_temperature("temperature")

if self._read_safety_off():
self._error = True

self._native_value = self._current
self._is_heating = self._read_is_heating()
self._is_cooling = self._read_is_cooling()
self._havc_action = self._read_operating_state()
Expand Down
109 changes: 89 additions & 20 deletions blebox_uniapi/sensor.py
View file
Open in desktop
Original file line number Diff line number Diff line change
@@ -1,5 +1,5 @@
import datetime
import numbers
from enum import IntEnum
from functools import partial

from .feature import Feature
Expand Down Expand Up @@ -67,10 +67,9 @@ def many_from_config(cls, product, box_type_config, extended_state):
# note: methods for sensor readings are provided as template
# functions (lambdas) in the box config. We need to "materialize"
# them to make sure they are properly indexed by sensor ID
materialized_methods = {
**methods,
device_class: methods[device_class](sensor_id),
}
materialized_methods = Feature.resolve_access_method_paths(
methods, sensor_id
)

feature = constructor(
product=product,
Expand All @@ -97,12 +96,24 @@ def many_from_config(cls, product, box_type_config, extended_state):
return []


class BleboxSensorState(IntEnum):
"""Possible states of a sensor reading, not all valid for every device type."""

IDLE = 0
INITIALIZING = 1
ACTIVE = 2
ERROR = 3
ABOVE_RANGE = 4
BELOW_RANGE = 5


class BaseSensor(Feature):
_unit: str
_device_class: str
_native_value: Union[float, int, str]
_native_value: Optional[Union[float, int, str]] = None
_sensor_type: Optional[str]
_sensor_id: Optional[int]
_error: bool = False

def __init__(
self,
Expand Down Expand Up @@ -130,6 +141,26 @@ def device_class(self) -> str:
def native_value(self):
return self._native_value

@property
def is_error(self) -> bool:
return self._error

def _read_state(self, name: str) -> Optional[int]:
if self._product.last_data is None:
return None
raw = self.raw_value(f"{name}.state")
if not isinstance(raw, (int, float)):
return None
return int(raw)

@staticmethod
def _state_is_error(state: Optional[int]) -> bool:
return state == BleboxSensorState.ERROR

@staticmethod
def _state_is_initializing(state: Optional[int]) -> bool:
return state == BleboxSensorState.INITIALIZING

@property
def sensor_id(self):
return self._sensor_id
Expand Down Expand Up @@ -201,9 +232,22 @@ def after_update(self):
if product.last_data is None:
return

state = self._read_state(self._device_class)
if self._state_is_error(state):
self._error = True
self._native_value = None
return

self._error = False

if self._state_is_initializing(state):
self._native_value = None
return

raw = self.raw_value(self._device_class)
if not isinstance(raw, numbers.Number):
raw = float("nan")
if not isinstance(raw, (int, float)):
self._native_value = None
return

native = raw / self._scale
if self._precision:
Expand Down Expand Up @@ -234,7 +278,7 @@ def _read_period_of_measurement(self) -> int:

@SensorFactory.register("temperature")
class Temperature(BaseSensor):
_current: Union[float, int, None]
_current: Optional[Union[float, int]]

def __init__(
self,
Expand All @@ -249,21 +293,35 @@ def __init__(
self._device_class = "temperature"

@property
def current(self) -> Union[float, int, None]:
def current(self) -> Optional[Union[float, int]]:
return self._current

def _read_temperature(self, field: str) -> Union[float, int, None]:
def _read_temperature(self, field: str) -> Optional[Union[float, int]]:
product = self._product
if product.last_data is not None:
raw = self.raw_value(field)
if raw is not None:
alias = self._alias
return round(product.expect_int(alias, raw, 12500, -5500) / 100.0, 1)
if isinstance(raw, (int, float)):
return round(raw / 100.0, 1)
return None

def after_update(self) -> None:
self._current = self._read_temperature("temperature")
self._native_value = self._read_temperature("temperature")
state = self._read_state("temperature")
if self._state_is_error(state):
self._error = True
self._current = None
self._native_value = None
return

self._error = False

if self._state_is_initializing(state):
self._current = None
self._native_value = None
return

current = self._read_temperature("temperature")
self._current = current
self._native_value = current


@SensorFactory.register("airSensor")
Expand All @@ -282,14 +340,25 @@ def __init__(
self._unit = "concentration_of_mp"
self._device_class = alias

def _pm_value(self, name: str) -> Optional[int]:
def _pm_value(self, name: str) -> Optional[Union[int, float]]:
product = self._product
if product.last_data is not None:
raw = self.raw_value(name)
if raw is not None:
alias = self._alias
return product.expect_int(alias, raw, 3000, 0)
if isinstance(raw, (int, float)):
return raw
return None

def after_update(self) -> None:
state = self._read_state(self.device_class)
if self._state_is_error(state):
self._error = True
self._native_value = None
return

self._error = False

if self._state_is_initializing(state):
self._native_value = None
return

self._native_value = self._pm_value(f"{self.device_class}.value")
Loading
Loading