From ece851f5a01fcbc64ab19781a2d464fc2950e64a Mon Sep 17 00:00:00 2001 From: Pierre Gironde Date: Sun, 4 Oct 2026 11:49:58 +0200 Subject: [PATCH] add support for expression in sensor_attributes --- ...st_virtual_sensor_attribute_expressions.py | 103 ++++++++++++++++++ virtualsensors.py | 48 ++++++-- 2 files changed, 143 insertions(+), 8 deletions(-) create mode 100644 unit_tests/test_virtual_sensor_attribute_expressions.py diff --git a/unit_tests/test_virtual_sensor_attribute_expressions.py b/unit_tests/test_virtual_sensor_attribute_expressions.py new file mode 100644 index 0000000..074a755 --- /dev/null +++ b/unit_tests/test_virtual_sensor_attribute_expressions.py @@ -0,0 +1,103 @@ +"""Check reactive expressions alongside legacy entity-backed attributes.""" +import ast +import re +import sys +import unittest +from pathlib import Path + + +MODULE_DIR = Path(__file__).resolve().parents[1] +sys.path.insert(0, str(MODULE_DIR)) +import smartvalue as SmartValue +from expressionparser import ParsingException + + +source = MODULE_DIR / "virtualsensors.py" +node = next(node for node in ast.parse(source.read_text()).body + if isinstance(node, ast.ClassDef) and node.name == "VirtualSensorBase") +namespace = {"SmartValue": SmartValue, "ParsingException": ParsingException, "re": re} +exec(compile(ast.Module(body=[node], type_ignores=[]), str(source), "exec"), namespace) +VirtualSensorBase = namespace["VirtualSensorBase"] + + +class FakeAPI: + def __init__(self): + self.states = { + "sensor.icon_source": {"state": "mdi:bike", "attributes": {}}, + "sensor.next_rain": {"state": "unknown", "attributes": {"minutes_until_rain": "12"}}, + } + self.listeners = [] + + def entity_exists(self, entity): + return entity in self.states + + def get_state(self, entity, attribute=None): + data = self.states.get(entity) + if data is None: + return None + return data["state"] if attribute is None else data["attributes"].get(attribute) + + def set_state(self, entity, state=None, attributes=None): + data = self.states.setdefault(entity, {"state": None, "attributes": {}}) + if state is not None: + data["state"] = state + if attributes is not None: + data["attributes"].update(attributes) + + def listen_state(self, callback, entity, attribute=None): + self.listeners.append((callback, entity, attribute)) + + def change(self, entity, value, attribute=None): + previous = self.get_state(entity, attribute) + if attribute is None: + self.states[entity]["state"] = value + else: + self.states[entity]["attributes"][attribute] = value + for callback, watched_entity, watched_attribute in list(self.listeners): + if (entity, attribute) == (watched_entity, watched_attribute): + callback(entity, attribute, previous, value, {}) + + +class FakeLogger: + def log_info(self, message): + pass + + def log_warning(self, message): + pass + + def log_error(self, message): + raise AssertionError(message) + + +class ExampleSensor(VirtualSensorBase): + def initialize(self): + self.ad_api.set_state(self.sensor_name, state="off") + + +class SensorAttributeExpressionsTests(unittest.TestCase): + def test_legacy_state_and_expressions_track_their_sources(self): + api = FakeAPI() + sensor = ExampleSensor( + api, FakeLogger(), virtual_sensor_name="example", + yaml_block={"sensor_attributes": { + "icon": "sensor.icon_source", + "minutes": "sensor.next_rain.minutes_until_rain", + "minutes_plus_five": "sensor.next_rain.minutes_until_rain + 5", + }}, + ) + + self.assertEqual(set(sensor.generate_attributes_dependencies_list()), + {"sensor.icon_source", "sensor.next_rain"}) + sensor._initialize() + attrs = api.states["sensor.example"]["attributes"] + self.assertEqual((attrs["icon"], attrs["minutes"], attrs["minutes_plus_five"]), + ("mdi:bike", 12, 17)) + + api.change("sensor.next_rain", "7", "minutes_until_rain") + api.change("sensor.icon_source", "mdi:weather-rainy") + self.assertEqual((attrs["icon"], attrs["minutes"], attrs["minutes_plus_five"]), + ("mdi:weather-rainy", 7, 12)) + + +if __name__ == "__main__": + unittest.main() diff --git a/virtualsensors.py b/virtualsensors.py index d0cd076..5bf9511 100644 --- a/virtualsensors.py +++ b/virtualsensors.py @@ -4,6 +4,7 @@ import smartvalue as SmartValue from kwargsparser import kwargsParser from expressionparser import ParsingException import time +import re from logger_interface import LoggerInterface from ad_toolbox.eventhandler import EventHandler @@ -65,21 +66,48 @@ class VirtualSensorBase: def generate_dependencies_list(self): return [] + @staticmethod + def is_entity_reference(value): + # Keep plain entity IDs on the legacy path so their raw state type is preserved. + return isinstance(value, str) and re.fullmatch(r"[a-z_][a-z0-9_]*\.[a-z0-9_]+", value) is not None + def generate_attributes_dependencies_list(self): dependencies = list() if 'sensor_attributes' in self.yaml_block: - for key,entity in self.yaml_block['sensor_attributes'].items(): - dependencies.append(entity) + for expression in self.yaml_block['sensor_attributes'].values(): + if self.is_entity_reference(expression): + dependencies.append(expression) + else: + evaluator = SmartValue.Evaluator( + self.ad_api, expression, constants=self.constants, + sensor_existence_validation=False, activated=False, log_init=False) + self.extract_dependencies_from_entities_to_listen( + evaluator.get_entities_to_listen_as_list(), dependencies) return dependencies def _initialize(self): + self.attribute_expressions = dict() if 'sensor_attributes' in self.yaml_block: - for attribute_name,sensor_name in self.yaml_block['sensor_attributes'].items(): - if not self.ad_api.entity_exists(sensor_name): - self.log_error(f"Error in sensor_attribute. {attribute_name}:{sensor_name}. {sensor_name} doesn't exist") + for attribute_name,expression in self.yaml_block['sensor_attributes'].items(): + if self.is_entity_reference(expression): + if not self.ad_api.entity_exists(expression): + self.log_error(f"Error in sensor_attribute. {attribute_name}:{expression}. {expression} doesn't exist") + else: + self.attributes[attribute_name] = self.ad_api.get_state(expression) + self.ad_api.listen_state(self.on_attribute_sensor_state_change,expression) else: - self.attributes[attribute_name] = self.ad_api.get_state(sensor_name) - self.ad_api.listen_state(self.on_attribute_sensor_state_change,sensor_name) + try: + evaluator = SmartValue.Evaluator( + self.ad_api, expression, logger_interface=self.logger_interface, + expression_name=f"{self.sensor_name}.{attribute_name}", + constants=self.constants, sensor_existence_validation=False, + trigger_callback_on_activation=False, + on_change_cb=self.on_attribute_expression_change, + pass_expression_name_to_cb=attribute_name) + self.attribute_expressions[attribute_name] = evaluator + self.attributes[attribute_name] = evaluator.evaluate(False) + except ParsingException as error: + self.log_error(error) try: self.initialize() except ParsingException as e: @@ -96,6 +124,10 @@ class VirtualSensorBase: if attributes_update: self.ad_api.set_state(self.sensor_name, attributes = self.attributes) + def on_attribute_expression_change(self, attribute_name, previous_value, value): + self.attributes[attribute_name] = value + self.ad_api.set_state(self.sensor_name, attributes=self.attributes) + def initialize(self): assert False def extract_dependencies_from_entities_to_listen(self,entities_to_listen,dependencies): @@ -712,4 +744,4 @@ class VirtualSensors(): class VirtualSensorsApp(hass.Hass): def initialize(self): self.logger_interface = LoggerInterface(self.get_ad_api(),default_log = "virtualsensors_log") - self.virtual_sensors = VirtualSensors(ad_api = self.get_ad_api(),logger_interface = self.logger_interface,yaml_block = self.args,args_to_ignore_in_validation = ['module','class','global_dependencies','priority','name','config_path'],app_name = self.name) \ No newline at end of file + self.virtual_sensors = VirtualSensors(ad_api = self.get_ad_api(),logger_interface = self.logger_interface,yaml_block = self.args,args_to_ignore_in_validation = ['module','class','global_dependencies','priority','name','config_path'],app_name = self.name)