From 3360935751a2d361f71d73ad28cf8333795affd1 Mon Sep 17 00:00:00 2001 From: Richard Bowman Date: Mon, 14 Sep 2026 23:32:08 +0100 Subject: [PATCH 1/4] Functional properties that track attribute access This is the start of computed properties. Computed properties are like functional properties, but they keep track of attribute access. This should then allow us to update them whenever their inputs change. Currently, an error is thrown if anything other than an observable property is accessed. --- src/labthings_fastapi/computed_properties.py | 101 +++++++++++++++++++ tests/test_computed_properties.py | 60 +++++++++++ 2 files changed, 161 insertions(+) create mode 100644 src/labthings_fastapi/computed_properties.py create mode 100644 tests/test_computed_properties.py diff --git a/src/labthings_fastapi/computed_properties.py b/src/labthings_fastapi/computed_properties.py new file mode 100644 index 00000000..0ecffc20 --- /dev/null +++ b/src/labthings_fastapi/computed_properties.py @@ -0,0 +1,101 @@ +"""Computed properties. + +A computed property is like a functional property, but it only depends on +properties which are observable. +This means LabThings is able to recompute +it whenever its dependencies change, allowing it to be observed. +""" + +from collections.abc import Callable +from typing import TYPE_CHECKING, Any, Generic +from weakref import WeakKeyDictionary + +from labthings_fastapi.exceptions import PropertyNotObservableError +from labthings_fastapi.properties import FunctionalProperty, Owner, Value + +if TYPE_CHECKING: + from labthings_fastapi.thing import Thing + + +class AccessWrapper: + """Wrap access to the properties of an object.""" + + def __init__(self, obj: "Thing", dependencies: set[str]) -> None: + """Initialise the AccessWrapper. + + :param obj: the object being wrapped. + :param dependencies: a set to use for tracking dependencies. + """ + self._obj = obj + self._dependencies = dependencies + + def __getattr__(self, name: str) -> Any: + """Proxy attribute access to the underlying object. + + :param name: The name of the attribute being accessed. + :return: The value of the object. + :raises PropertyNotObservableError: if a non-observable property + is accessed. + :raises TypeError: if something other than a property is accessed. + """ + if name in self._obj.properties: + if self._obj.properties[name].is_observable: + self._dependencies.add(name) + print(f"Found an access to {name}") + return getattr(self._obj, name) + else: + raise PropertyNotObservableError( + f"{self._obj.name}.{name} is not observable. " + "Only observable properties may be accessed from a " + "computed property." + ) + else: + raise TypeError( + f"{self._obj.name}.{name} is not a property. " + "Only observable properties may be accessed from a " + "computed property." + ) + + +class ComputedProperty(FunctionalProperty[Owner, Value], Generic[Owner, Value]): + """A property that recomputes its value on demand. + + This is a way to make functional properties observable. + """ + + def __init__( + self, + fget: Callable[[Owner], Value], + **kwargs: Any, + ) -> None: + r"""Initialise a computed property. + + :param fget: The getter function. + :param \**kwargs: Additional keyword arguments are passed to + `BaseProperty`. + """ + super().__init__(fget=fget, **kwargs) + self._dependencies: "WeakKeyDictionary[Thing, set[str]]" = WeakKeyDictionary() + + def instance_get(self, obj: Owner) -> Value: + """Get the value of this functional property. + + :param obj: the object on which this property is being accessed. + :return: the value of the property. + """ + dependencies: set[str] = set() + access_wrapper = AccessWrapper(obj, dependencies=dependencies) + # access_wrapper is wrapping `self` but has its own type. We therefore + # ignore type checking on this line. + val = self._fget(access_wrapper) # type: ignore[arg-type] + self._dependencies[obj] = dependencies + return val + + +def computed_property(fget: Callable[[Owner], Value]) -> ComputedProperty[Owner, Value]: + """Decorate a method as a computed property. + + :param fget: the getter function, which must depend only on observable properties. + :return: a computed property descriptor. + """ + return ComputedProperty(fget=fget) diff --git a/tests/test_computed_properties.py b/tests/test_computed_properties.py new file mode 100644 index 00000000..bdaa62f7 --- /dev/null +++ b/tests/test_computed_properties.py @@ -0,0 +1,60 @@ +"""Test computed properties in isolation.""" + +from typing import Literal + +import pytest + +import labthings_fastapi as lt +from labthings_fastapi.computed_properties import computed_property +from labthings_fastapi.exceptions import PropertyNotObservableError +from labthings_fastapi.testing import create_thing_without_server + + +class MyThing(lt.Thing): + quantity: int = lt.property(default=0) + quantity2: int = lt.property(default=0) + selector: Literal["q1", "q2"] = lt.property(default="q1") + + @computed_property + def double(self) -> int: + return 2 * self.quantity + + @computed_property + def selected(self) -> int: + if self.selector == "q1": + return self.quantity + elif self.selector == "q2": + return self.quantity2 + else: + raise ValueError("invalid selector") + + @lt.property + def unobservable(self) -> str: + return "unobservable" + + @computed_property + def broken(self) -> str: + return self.unobservable + + +def test_dependencies(): + thing = create_thing_without_server(MyThing) + + thing.quantity = 1 + assert thing.double == 2 + assert MyThing.double._dependencies[thing] == {"quantity"} + + assert thing.selected == 1 + assert MyThing.selected._dependencies[thing] == {"selector", "quantity"} + + thing.selector = "q2" + thing.quantity2 = 10 + assert thing.selected == 10 + assert MyThing.selected._dependencies[thing] == {"selector", "quantity2"} + + +def test_unobservable(): + """Check computed properties error if they depend on unobservable quantities.""" + thing = create_thing_without_server(MyThing) + with pytest.raises(PropertyNotObservableError): + _ = thing.broken From 939873d234db52852e04d42495356e324d584c67 Mon Sep 17 00:00:00 2001 From: Richard Bowman Date: Tue, 15 Sep 2026 09:10:26 +0100 Subject: [PATCH 2/4] Add a helper function to eliminate type: ignore --- src/labthings_fastapi/computed_properties.py | 32 ++++++++++++++++++-- 1 file changed, 29 insertions(+), 3 deletions(-) diff --git a/src/labthings_fastapi/computed_properties.py b/src/labthings_fastapi/computed_properties.py index 0ecffc20..94d45d50 100644 --- a/src/labthings_fastapi/computed_properties.py +++ b/src/labthings_fastapi/computed_properties.py @@ -7,7 +7,7 @@ """ from collections.abc import Callable -from typing import TYPE_CHECKING, Any, Generic +from typing import TYPE_CHECKING, Any, Generic, cast from weakref import WeakKeyDictionary from labthings_fastapi.exceptions import PropertyNotObservableError @@ -56,6 +56,32 @@ def __getattr__(self, name: str) -> Any: "computed property." ) + def __setattr__(self, name: str, value: Any) -> None: + """Don't allow attributes to be set, there should be no side-effects. + + :param name: the name of the attribute. + :param value: the value to set. + :raises AttributeError: because the wrapper is read-only. + """ + raise AttributeError("Computed properties may not set values.") + + +def access_wrapper(obj: Owner, dependencies: set[str]) -> Owner: + """Wrap a Thing to record attribute access. + + This function is preferred to instantiating AccessWrapper directly, + as it ensures the wrapper is type hinted as the original object. + + :param obj: the Thing to wrap. + :param dependencies: a set to store dependencies. + :return: `obj` with an attribute access wrapper. + """ + # Typing note: AccessWrapper proxies attribute access back to the + # wrapped object, so its signature should be identical to `obj` + # and thus the `cast` below is justified. + wrapper = AccessWrapper(obj, dependencies=dependencies) + return cast(Owner, wrapper) + class ComputedProperty(FunctionalProperty[Owner, Value], Generic[Owner, Value]): """A property that recomputes its value on demand. @@ -84,10 +110,10 @@ def instance_get(self, obj: Owner) -> Value: :return: the value of the property. """ dependencies: set[str] = set() - access_wrapper = AccessWrapper(obj, dependencies=dependencies) + wrapper = access_wrapper(obj, dependencies=dependencies) # access_wrapper is wrapping `self` but has its own type. We therefore # ignore type checking on this line. - val = self._fget(access_wrapper) # type: ignore[arg-type] + val = self._fget(wrapper) # type: ignore[arg-type] self._dependencies[obj] = dependencies return val From 222722d0b6a5781e5e342e7a78ae20847a13e835 Mon Sep 17 00:00:00 2001 From: Richard Bowman Date: Wed, 16 Sep 2026 10:56:47 +0100 Subject: [PATCH 3/4] A first implementation that dynamically starts listening on first read. This will create streams and subscribe to updates on dependencies whenever the property is read. This should mean that we need only read the property once at the start, and we'll then publish updates thereafter. Currently there isn't any global trigger to start listening, and there's no way to shut down the listening coroutines. --- src/labthings_fastapi/computed_properties.py | 95 ++++++++++++++++++- src/labthings_fastapi/properties.py | 29 ++++++ .../thing_server_interface.py | 36 +++++++ 3 files changed, 156 insertions(+), 4 deletions(-) diff --git a/src/labthings_fastapi/computed_properties.py b/src/labthings_fastapi/computed_properties.py index 94d45d50..f60e372b 100644 --- a/src/labthings_fastapi/computed_properties.py +++ b/src/labthings_fastapi/computed_properties.py @@ -10,7 +10,11 @@ from typing import TYPE_CHECKING, Any, Generic, cast from weakref import WeakKeyDictionary +from anyio import create_memory_object_stream +from anyio.streams.memory import MemoryObjectReceiveStream, MemoryObjectSendStream + from labthings_fastapi.exceptions import PropertyNotObservableError +from labthings_fastapi.message_broker import Message from labthings_fastapi.properties import FunctionalProperty, Owner, Value if TYPE_CHECKING: @@ -102,6 +106,15 @@ def __init__( """ super().__init__(fget=fget, **kwargs) self._dependencies: "WeakKeyDictionary[Thing, set[str]]" = WeakKeyDictionary() + self._streams: WeakKeyDictionary[ + "Thing", + tuple[MemoryObjectSendStream[Message], MemoryObjectReceiveStream[Message]], + ] = WeakKeyDictionary() + + @property + def is_computed(self) -> bool: + """Whether the property is a computed property.""" + return True def instance_get(self, obj: Owner) -> Value: """Get the value of this functional property. @@ -111,12 +124,76 @@ def instance_get(self, obj: Owner) -> Value: """ dependencies: set[str] = set() wrapper = access_wrapper(obj, dependencies=dependencies) - # access_wrapper is wrapping `self` but has its own type. We therefore - # ignore type checking on this line. - val = self._fget(wrapper) # type: ignore[arg-type] - self._dependencies[obj] = dependencies + val = self._fget(wrapper) + if not dependencies: + self._stop_watching(obj) + else: + self._watch_for_changes(obj, dependencies) return val + def _watch_for_changes(self, obj: Owner, dependencies: set[str]) -> None: + """Ensure a coroutine is watching for changes. + + This method will check whether we are currently processing messages + from the properties we depend on, so that this property will be + recomputed when they change. If that's not happening, we will start + a new coroutine to do this. + + :param obj: the object on which the property is defined. + :param dependencies: the properties on which the object depends. + :raises ValueError: if the dependencies set is empty. + """ + if not dependencies: + # This shouldn't ever happen - the calling function checks that + # dependencies is non-empty. + raise ValueError( + "_watch_for_changes must have at least one dependency to watch." + ) + streams = self._streams.get(obj) + if streams is None or streams[1].statistics().open_receive_streams == 0: + # If the streams are missing or closed, create them and start the coroutine. + send, recv = create_memory_object_stream[Message]() + obj._thing_server_interface.start_async_task_soon( + _recompute_on_changes, self.descriptor_info().publish, recv + ) + self._streams[obj] = send, recv + else: + send, recv = self._streams[obj] + + # The streams exist and a coroutine is monitoring them. Now, subscribe + # to changes in our dependencies + for affordance in dependencies: + obj._thing_server_interface.subscribe(obj.name, affordance, send) + # Unsubscribe from any dependencies no longer needed + for affordance in self._dependencies[obj].difference(dependencies): + obj._thing_server_interface.unsubscribe(obj.name, affordance, send) + self._dependencies[obj] = dependencies + + def _stop_watching(self, obj: Owner) -> None: + """Stop watching for changes, as we have no dependencies. + + :param obj: the Thing on which we are defined. + """ + send, _recv = self._streams[obj] + # Unsubscribe from any dependencies no longer needed + for affordance in self._dependencies[obj]: + obj._thing_server_interface.unsubscribe(obj.name, affordance, send) + # Close the stream, so that the coroutine terminates + send.close() + + +async def _recompute_on_changes( + recompute: Callable[[], None], stream: MemoryObjectReceiveStream[Message] +) -> None: + """Recompute and publish a property if its dependencies change. + + :param recompute: a function to call when a dependency changes. + :param stream: the stream on which we'll get notified. + """ + async for _msg in stream: + # Currently, we don't check what changed - we just trigger a recompute. + recompute() + def computed_property(fget: Callable[[Owner], Value]) -> ComputedProperty[Owner, Value]: """Decorate a method as a computed property. @@ -125,3 +202,13 @@ def computed_property(fget: Callable[[Owner], Value]) -> ComputedProperty[Owner, :return: a computed property descriptor. """ return ComputedProperty(fget=fget) + + +def initialise_computed_properties(thing: "Thing") -> None: + """Initialise the computed properties on a Thing. + + :param thing: the Thing on which to initialise computed properties. + """ + for prop in thing.properties.values(): + if prop.is_computed: + prop.publish() diff --git a/src/labthings_fastapi/properties.py b/src/labthings_fastapi/properties.py index 2de6fc68..de715f15 100644 --- a/src/labthings_fastapi/properties.py +++ b/src/labthings_fastapi/properties.py @@ -414,6 +414,11 @@ def __init__( `False` for functional properties. """ + @builtins.property + def is_computed(self) -> bool: # noqa: DOC201 + """Whether the property is a computed property.""" + return False + @staticmethod def _validate_constraints(constraints: Mapping[str, Any]) -> FieldConstraints: """Validate an untyped dictionary of constraints. @@ -1214,6 +1219,30 @@ def reset(self) -> None: """ return self.get_descriptor().reset(self.owning_object_or_error()) + @builtins.property + def is_computed(self) -> bool: # noqa: DOC201 + """Whether the property is a computed property.""" + # This is done by inspecting the name rather than an isinstance check + # to avoid circular dependencies. + return self.get_descriptor().is_computed + + def publish(self) -> None: + """Get the property's value, and publish it to any observers. + + This reduces boilerplate when you are managing property notifications manually, + and is also used to initialise computed properties. + """ + value = self.get() + obj = self.owning_object_or_error() + obj._thing_server_interface.publish( + Message( + thing=obj.name, + affordance=self.name, + message_type="property", + payload=value, + ) + ) + def validate(self, value: Any) -> Value: """Use the validation logic in `self.model`. diff --git a/src/labthings_fastapi/thing_server_interface.py b/src/labthings_fastapi/thing_server_interface.py index 3acbf011..29418e0f 100644 --- a/src/labthings_fastapi/thing_server_interface.py +++ b/src/labthings_fastapi/thing_server_interface.py @@ -18,6 +18,8 @@ ) from weakref import ReferenceType, ref +from anyio.streams.memory import MemoryObjectSendStream + from labthings_fastapi.exceptions import FeatureNotEnabledError, ServerNotRunningError from labthings_fastapi.global_lock import GlobalLock from labthings_fastapi.message_broker import Message @@ -156,6 +158,40 @@ def publish(self, message: Message) -> None: except ServerNotRunningError: pass # If the server isn't running yet, we can't publish events. + def subscribe( + self, thing: str, affordance: str, stream: MemoryObjectSendStream[Message] + ) -> None: + """Subscribe to messages from an affordance. + + :param thing: the name of the Thing being subscribed to. + :param affordance: the name of the affordance being subscribed to. + :param stream: the stream messages should be sent to. + :raises ServerNotRunningError: if the server hasn't started yet. It doesn't + make sense to subscribe to events if the event loop isn't yet running. + """ + try: + broker = self._get_server().message_broker + self.start_async_task_soon(broker.subscribe, thing, affordance, stream) + except ServerNotRunningError: + raise + + def unsubscribe( + self, thing: str, affordance: str, stream: MemoryObjectSendStream[Message] + ) -> None: + """Unsubscribe to messages from an affordance. + + :param thing: the name of the Thing being subscribed to. + :param affordance: the name of the affordance being subscribed to. + :param stream: the stream messages should be sent to. + :raises ServerNotRunningError: if the server hasn't started yet. It doesn't + make sense to unsubscribe from events if the event loop isn't yet running. + """ + try: + broker = self._get_server().message_broker + self.start_async_task_soon(broker.unsubscribe, thing, affordance, stream) + except ServerNotRunningError: + raise + @property def settings_folder(self) -> str: """The path to a folder where persistent files may be saved.""" From 6e6dee1e9c7aa48f50e2a9617e22b352e140d10f Mon Sep 17 00:00:00 2001 From: Richard Bowman Date: Wed, 16 Sep 2026 12:13:56 +0100 Subject: [PATCH 4/4] Streamlined code for watching The only missing feature is a way of manually triggering a recomputation, which might reintroduce much of the faff I eliminated. --- src/labthings_fastapi/computed_properties.py | 140 ++++++++----------- 1 file changed, 59 insertions(+), 81 deletions(-) diff --git a/src/labthings_fastapi/computed_properties.py b/src/labthings_fastapi/computed_properties.py index f60e372b..c7cb7a6c 100644 --- a/src/labthings_fastapi/computed_properties.py +++ b/src/labthings_fastapi/computed_properties.py @@ -8,10 +8,9 @@ from collections.abc import Callable from typing import TYPE_CHECKING, Any, Generic, cast -from weakref import WeakKeyDictionary from anyio import create_memory_object_stream -from anyio.streams.memory import MemoryObjectReceiveStream, MemoryObjectSendStream +from anyio.streams.memory import MemoryObjectSendStream from labthings_fastapi.exceptions import PropertyNotObservableError from labthings_fastapi.message_broker import Message @@ -93,106 +92,80 @@ class ComputedProperty(FunctionalProperty[Owner, Value], Generic[Owner, Value]): This is a way to make functional properties observable. """ - def __init__( - self, - fget: Callable[[Owner], Value], - **kwargs: Any, - ) -> None: - r"""Initialise a computed property. - - :param fget: The getter function. - :param \**kwargs: Additional keyword arguments are passed to - `BaseProperty`. - """ - super().__init__(fget=fget, **kwargs) - self._dependencies: "WeakKeyDictionary[Thing, set[str]]" = WeakKeyDictionary() - self._streams: WeakKeyDictionary[ - "Thing", - tuple[MemoryObjectSendStream[Message], MemoryObjectReceiveStream[Message]], - ] = WeakKeyDictionary() - @property def is_computed(self) -> bool: """Whether the property is a computed property.""" return True - def instance_get(self, obj: Owner) -> Value: + def instance_get(self, obj: Owner, dependencies: set[str] | None = None) -> Value: """Get the value of this functional property. :param obj: the object on which this property is being accessed. + :param dependencies: an optional set to be populated with properties accessed + during recomputation. :return: the value of the property. """ - dependencies: set[str] = set() - wrapper = access_wrapper(obj, dependencies=dependencies) - val = self._fget(wrapper) - if not dependencies: - self._stop_watching(obj) + if dependencies is not None: + wrapper = access_wrapper(obj, dependencies=dependencies) + return self._fget(wrapper) else: - self._watch_for_changes(obj, dependencies) - return val + return self._fget(obj) - def _watch_for_changes(self, obj: Owner, dependencies: set[str]) -> None: - """Ensure a coroutine is watching for changes. + async def _watch_for_changes(self, obj: Owner) -> None: + """Watch for changes in our dependencies, and recompute as needed. - This method will check whether we are currently processing messages - from the properties we depend on, so that this property will be - recomputed when they change. If that's not happening, we will start - a new coroutine to do this. + This method will evaluate the computed property, tracking which + properties are accessed. It then subscribes to these properties, + and will recompute the property as required. :param obj: the object on which the property is defined. - :param dependencies: the properties on which the object depends. - :raises ValueError: if the dependencies set is empty. """ - if not dependencies: - # This shouldn't ever happen - the calling function checks that - # dependencies is non-empty. - raise ValueError( - "_watch_for_changes must have at least one dependency to watch." - ) - streams = self._streams.get(obj) - if streams is None or streams[1].statistics().open_receive_streams == 0: - # If the streams are missing or closed, create them and start the coroutine. - send, recv = create_memory_object_stream[Message]() - obj._thing_server_interface.start_async_task_soon( - _recompute_on_changes, self.descriptor_info().publish, recv - ) - self._streams[obj] = send, recv - else: - send, recv = self._streams[obj] + send, recv = create_memory_object_stream[Message](max_buffer_size=1) + # Subscribe to a non-existent affordance, to ensure that the stream + # is closed by the message broker even if there are no other + # subscriptions. + obj._thing_server_interface.subscribe(obj.name, "#dummy", send) - # The streams exist and a coroutine is monitoring them. Now, subscribe - # to changes in our dependencies - for affordance in dependencies: - obj._thing_server_interface.subscribe(obj.name, affordance, send) - # Unsubscribe from any dependencies no longer needed - for affordance in self._dependencies[obj].difference(dependencies): - obj._thing_server_interface.unsubscribe(obj.name, affordance, send) - self._dependencies[obj] = dependencies + dependencies: set[str] = set() + # Add a message to the stream, so we initialise everything immediately + # in the `async for` loop. + initial_message = Message(obj.name, self.name, "property", None) + await send.send(initial_message) + + # Whenever a dependency changes, we'll recompute and publish an update + async for message in recv: + old_dependencies = dependencies + dependencies = set() + value = self.instance_get(obj, dependencies=dependencies) + await self._update_subscriptions(obj, dependencies, old_dependencies, send) + if message is initial_message: + continue + obj._thing_server_interface.publish( + Message(obj.name, self.name, "property", value) + ) - def _stop_watching(self, obj: Owner) -> None: - """Stop watching for changes, as we have no dependencies. + @staticmethod + async def _update_subscriptions( + obj: Owner, + dependencies: set[str], + old_dependencies: set[str], + send_stream: MemoryObjectSendStream[Message], + ) -> None: + """Subscribe and unsubscribe to other properties. :param obj: the Thing on which we are defined. + :param dependencies: a set of the names of current dependencies. + :param old_dependencies: previous dependencies - any not in ``dependencies`` + will be unsubscribed. + :param send_stream: the stream to use for subscriptions. """ - send, _recv = self._streams[obj] + # The streams exist and a coroutine is monitoring them. Now, subscribe + # to changes in our dependencies + for affordance in dependencies: + obj._thing_server_interface.subscribe(obj.name, affordance, send_stream) # Unsubscribe from any dependencies no longer needed - for affordance in self._dependencies[obj]: - obj._thing_server_interface.unsubscribe(obj.name, affordance, send) - # Close the stream, so that the coroutine terminates - send.close() - - -async def _recompute_on_changes( - recompute: Callable[[], None], stream: MemoryObjectReceiveStream[Message] -) -> None: - """Recompute and publish a property if its dependencies change. - - :param recompute: a function to call when a dependency changes. - :param stream: the stream on which we'll get notified. - """ - async for _msg in stream: - # Currently, we don't check what changed - we just trigger a recompute. - recompute() + for affordance in old_dependencies.difference(dependencies): + obj._thing_server_interface.unsubscribe(obj.name, affordance, send_stream) def computed_property(fget: Callable[[Owner], Value]) -> ComputedProperty[Owner, Value]: @@ -211,4 +184,9 @@ def initialise_computed_properties(thing: "Thing") -> None: """ for prop in thing.properties.values(): if prop.is_computed: - prop.publish() + computed_property = prop.get_descriptor() + if not isinstance(computed_property, ComputedProperty): + continue + thing._thing_server_interface.start_async_task_soon( + computed_property._watch_for_changes, thing + )