diff --git a/docs/source/computed_properties.rst b/docs/source/computed_properties.rst new file mode 100644 index 00000000..7b253a83 --- /dev/null +++ b/docs/source/computed_properties.rst @@ -0,0 +1,103 @@ +.. _computed_properties: + +Computed Properties +========================= + +Computed properties are a special class of functional properties that depend only on the value of other properties. LabThings automatically tracks which other properties they depend on. This means the computed property will be automatically recalculated whenever one of those properties changes. An important consequence of this is that computed properties are *observable*. That means it's possible to subscribe to notifications of when the computed property changes, for example using a websocket. + +In order for computed properties to work, some care is needed to write them such that they *only* depend on other properties of the Thing. All of the properties that a computed property depends on must be observable, which usually means they are either data properties or other computed properties. + +Computed properties can be useful in a variety of places - for example, providing a composite property that includes the values of several individual properties or settings, or checking that a validation criterion is met. For example, consider a Thing that has a few settings: + +.. code-block:: python + + import labthings_fastapi as lt + + + class Scanner(lt.Thing): + scan_points: int = lt.setting(default=10) + vector_a: tuple[float, float] = lt.setting(default=(1, 0)) + vector_b: tuple[float, float] = lt.setting(default=(0, 1)) + + @lt.computed_property + def all_settings(self) -> dict: + """A quick way to get all the scan settings.""" + return { + "scan_points": self.scan_points, + "vector_a": self.vector_a, + "vector_b": self.vector_b, + } + + @lt.computed_property + def settings_valid(self) -> bool: + """Whether the current settings would work. + + This can't be a validator on any one setting, because + it depends on all of them. + """ + a = self.vector_a + b = self.vector_b + if a[0]*b[1] + a[1]*b[0] == 0 or self.scan_points == 1: + return True + else: + return False + + +A structure like this would allow a client to set each of the settings individually, and receive websocket notifications that tell it when the settings are OK, and when they are not valid. That's more efficient and easier to code than polling each time something is changed. A client that observes ``all_settings`` will get an update any time a setting changes, and that update will include all three values in a dictionary. + + +Setting computed properties +--------------------------- + +Computed properties may have setters. LabThings doesn't place any restrictions on what they can do, though it's almost always the case that they will only change the values of properties that are used in the getter. A trivial example would be two properties that are required to add up to the same total: + +.. code-block:: python + + import labthings_fastapi as lt + + class FractionThing(lt.Thing): + """The two properties of this Thing must always add up to 1.""" + + p: float = lt.property(default=0) + """This must always satisfy p+q==1.""" + + @lt.computed_property + def q(self) -> float: + """One minus p.""" + return 1-self.p + + @q.setter + def _set_q(self, value: float) -> None: + """Change p such that q takes this value.""" + self.p = 1 - value + +If ``p`` is changed, ``q`` will be automatically recomputed. If ``q`` is changed, it will change ``p`` such that ``q`` takes the intended value. Note that we can't set ``q`` directly - the only way to change its value is to change the properties it depends on. + +Manually recomputing and publishing +----------------------------------- + +It is possible to trigger an update message about any property in LabThings, by calling ``thing.properties["name"].publish()``. On data or functional properties, this simply gets the property's value and broadcasts it to any observers. If called on a computed property, it will trigger a recomputation of that property. It shouldn't be necessary to do this most of the time, as it ought to recompute automatically. However, it's sometimes possible to get into a situation (e.g. using mutable values) where the value has changed without LabThings triggering a notification. In these situations, calling ``publish()`` will get everything back in sync. + +Concurrency considerations +-------------------------- + +.. info:: + + This section is intended for developers. + +Computed properties rely on the `MessageBroker` mechanism that is used by websockets. As such, the signals that cause them to recompute their value come via the event loop. The computation is performed in a worker thread to minimise the risk that user code ends up blocking the event loop, and to avoid having to think about coroutines when writing what should be a very simple function. This does, however, lead to a slightly awkward structure. The following sequence illustrates how a computed property is implemented. + +* LabThings starts up, and calls ``__enter__`` on each of the `~lt.Thing`\ s. +* LabThings starts a coroutine for each computed property (`ComputedProperty._watch_for_changes`). + - The coroutine recomputes the property by running the getter in a worker thread. + - Each time the getter accesses a property, it calls back to the event loop and subscribes to that property. + - The coroutine recomputes in a worker thread every time the property changes. + +This slightly complicated structure ensures that we are always subscribed to changes before we access the value of a property, avoiding the risk that we miss updates to properties that happen while the recomputation is running. + +Infinite loops and recursion +---------------------------- + +If two computed properties depend on each other, if either one is accessed we will end up with an infinite loop and a recursion error. This is no different from what would happen with two regular properties outside of LabThings, and will result in exactly the same `RecursionError`. + +If the getter for a computed property has side effects, there's the potential for strange behaviour. Access to `self` from the getter function is therefore restricted only to reading other observable properties, which minimises the likelihood of strange things happening. diff --git a/docs/source/index.rst b/docs/source/index.rst index 9622f2c4..f31854b8 100644 --- a/docs/source/index.rst +++ b/docs/source/index.rst @@ -11,6 +11,7 @@ Documentation for LabThings-FastAPI structure.rst actions.rst properties.rst + computed_properties.rst documentation.rst thing_slots.rst blobs.rst diff --git a/docs/source/public_api.rst b/docs/source/public_api.rst index 9f276fe0..6f06183b 100644 --- a/docs/source/public_api.rst +++ b/docs/source/public_api.rst @@ -142,6 +142,11 @@ This page summarises the parts of the LabThings API that should be most frequent For a full listing of attributes that may be modified, see `DataProperty`\ . +.. py:decorator:: computed_property + + Decorate a method as a computed property. See :ref:`computed_properties` for more details. + + .. py:function:: setting(getter: Callable[[Owner], Value]) -> FunctionalSetting[Owner, Value] setting(*, default: Value, readonly: bool = False, use_global_lock: bool | None = None, **constraints: Any) -> Value setting(*, default_factory: Callable[[], Value], readonly: bool = False, use_global_lock: bool | None = None, **constraints: Any) -> Value diff --git a/src/labthings_fastapi/__init__.py b/src/labthings_fastapi/__init__.py index 3dc99adc..78a19b10 100644 --- a/src/labthings_fastapi/__init__.py +++ b/src/labthings_fastapi/__init__.py @@ -27,6 +27,7 @@ from labthings_fastapi import outputs from labthings_fastapi.actions import action from labthings_fastapi.client import ThingClient +from labthings_fastapi.computed_properties import computed_property from labthings_fastapi.endpoints import endpoint from labthings_fastapi.invocation_contexts import ( ThreadWithInvocationID, @@ -64,6 +65,7 @@ "blob", "cancellable_sleep", "cli", + "computed_property", "endpoint", "get_thing_logger", "outputs", diff --git a/src/labthings_fastapi/computed_properties.py b/src/labthings_fastapi/computed_properties.py new file mode 100644 index 00000000..0fec7e74 --- /dev/null +++ b/src/labthings_fastapi/computed_properties.py @@ -0,0 +1,307 @@ +"""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, Mapping +from typing import TYPE_CHECKING, Any, Generic, cast +from weakref import WeakKeyDictionary + +from anyio import create_memory_object_stream, from_thread, to_thread +from anyio.abc import TaskGroup, TaskStatus +from anyio.streams.memory import MemoryObjectSendStream +from typing_extensions import Literal + +from labthings_fastapi.exceptions import PropertyNotObservableError +from labthings_fastapi.message_broker import Message +from labthings_fastapi.properties import FunctionalProperty, Owner, Value + +if TYPE_CHECKING: + from labthings_fastapi.thing import Thing + + +__all__ = ["computed_property"] + + +RECOMPUTE = object() + + +class AccessWrapper: + """Wrap access to the properties of an object.""" + + _access_wrapper_locked = False + + def __init__( + self, + obj: "Thing", + callback: Callable[[str], None], + ) -> None: + """Initialise the AccessWrapper. + + :param obj: the object being wrapped. + :param callback: a function that's called with the name of each dependency. + """ + self._obj = obj + self._callback = callback + self._access_wrapper_locked = True + + 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._callback(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." + ) + + def __setattr__(self, name: str, value: Any) -> None: + """Don't allow attributes to be set. + + Once the wrapper is initialised, it won't allow attributes to be set. + We need to be able to set some attributes during `__init__`, but + after that, everything becomes read-only. + + :param name: the name of the attribute. + :param value: the value to set. + :raises AttributeError: because the wrapper is read-only. + """ + if self._access_wrapper_locked is True: + raise AttributeError("Computed properties may not set values.") + super().__setattr__(name, value) + + +def access_wrapper(obj: Owner, callback: Callable[[str], None]) -> 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 callback: a function that's called for each identified dependency, + with the property name as its argument. + :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, callback=callback) + return cast(Owner, wrapper) + + +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], + constraints: Mapping[str, Any] | None = None, + use_global_lock: Literal[False] | None = None, + ) -> None: + """Set up a FunctionalProperty. + + Create a descriptor for a property that uses a getter function. + + This class also inherits from `builtins.property` to help type checking + tools understand that it functions like a property. + + :param fget: the getter function, called when the property is read. + :param constraints: is passed as keyword arguments to `pydantic.Field` + to add validation constraints to the property. See `pydantic.Field` + for details. + :param use_global_lock: may be set to `False` to disable the global lock + for setting this property. By default, if global locking is enabled, + we hold the global lock while setting the property. + """ + super().__init__( + fget=fget, constraints=constraints, use_global_lock=use_global_lock + ) + self._send_streams = WeakKeyDictionary[ + "Thing", MemoryObjectSendStream[Message] + ]() + self._latest_values = WeakKeyDictionary["Thing", Value]() + + observable: bool = True + + @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 computed property. + + This will use the latest value that was cached. It will fall back to + computing it directly if a cached value is not available. + + :param obj: the object on which this property is being accessed. + :return: the value of the property. + """ + try: + value = self._latest_values[obj] + return value + except KeyError: + return self.fget(obj) + + async def _request_recomputation(self, obj: Owner) -> Value: + """Trigger a recomputation of the property and return the new value. + + :param obj: the object on which the property is defined. + :return: the value of the property. + """ + broker = obj._thing_server_interface.message_broker + send, recv = create_memory_object_stream[Message](max_buffer_size=1) + # Subscribe for updates (get the recomputed value) + await broker.subscribe(obj.name, self.name, send) + # Trigger a recompute + await self._send_streams[obj].send( + # Note that RECOMPUTE messages should never hit the broker: they + # are only ever sent directly to our stream. + Message(obj.name, self.name, "property", RECOMPUTE) + ) + message = await recv.receive() + return message.payload + + def publish(self, obj: Owner) -> Value: + """Recompute and publish the property's value. + + This will signal to the coroutine watching for changes that a recomputation + should happen. The new value will be published to the message broker. + This function will block until the recomputation has happened. + + :param obj: the Thing on which we are publishing the property. + :return: the recomputed value of the property. + """ + return obj._thing_server_interface.call_async_task( + self._request_recomputation, obj + ) + + async def _watch_for_changes(self, obj: Owner, task_status: TaskStatus) -> None: + """Watch for changes in our dependencies, and recompute as needed. + + 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 task_status: anyio handle that reports when this task has + started (i.e. once we've figured out dependencies and started + listening). + """ + broker = obj._thing_server_interface.message_broker + 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 when it's shut down even if there + # are no other subscriptions. + await broker.subscribe(obj.name, "#dummy", send) + # Save the send stream so _request_recomputation can trigger updates + self._send_streams[obj] = send + # This set will be filled with the dependencies each time we recompute. + dependencies: set[str] = set() + + def add_dependency(name: str) -> None: + """Add a dependency to the ``dependencies`` set and subscribe. + + Note: this is called from the worker thread that is used to + recompute the property, hence the need for `anyio.from_thread.run_sync`. + + It's important that we subscribe as-we-go otherwise it's possible to + miss changes to dependencies if they occur during recomputation. + + The dependencies set should be cleared before each recomputation. + + :param name: the name of the property to subscribe to. + """ + from_thread.run(broker.subscribe, obj.name, name, send) + dependencies.add(name) + + # This adds a side-effect to attribute access: it will subscribe to + # the attribute and add the attribute's name to `dependencies`. + obj_access_wrapper = access_wrapper(obj, add_dependency) + + # Add a message to the stream, so we initialise everything immediately + # in the `async for` loop. + initial_message = Message(obj.name, self.name, "property", RECOMPUTE) + await send.send(initial_message) + + # Whenever a dependency changes, we'll recompute and publish an update. + # Each time we update the value, we also check the dependencies in case + # they change. + async for message in recv: + # We need to keep track of dependencies, so we can unsubscribe from + # those we no longer need. + old_dependencies = dependencies.copy() + dependencies.clear() + # The next line recomputes the value, subscribes to dependencies, and + # puts the new dependencies into the `dependencies` set. + value = await to_thread.run_sync(self._fget, obj_access_wrapper) + self._latest_values[obj] = value + # Unsubscribe from any dependencies no longer needed + for affordance in old_dependencies.difference(dependencies): + await broker.unsubscribe(obj.name, affordance, send) + if message is initial_message: + # At this point, we've calculated our dependencies and started + # listening, so we signal that everything's started OK. + # We don't send an initial update, hence the `continue`. + task_status.started() + continue + await broker.publish(Message(obj.name, self.name, "property", value)) + + +def computed_property(fget: Callable[[Owner], Value]) -> ComputedProperty[Owner, Value]: + """Decorate a method as a computed property. + + Computed properties are simple functional properties that only depend on other + properties of their host Thing. Unlike functional properties, it's possible to + observe them for changes, and LabThings will recalculate them automatically when + their dependencies change. + + See :ref:`computed_properties` for full details of how computed properties work and + how to use them. + + :param fget: the getter function, which must depend only on observable properties. + :return: a computed property descriptor. + """ + return ComputedProperty(fget=fget) + + +async def start_watching_computed_properties( + thing: "Thing", + task_group: TaskGroup, +) -> None: + """Watch the computed properties on a Thing. + + :param thing: the Thing on which to watch computed properties. + :param task_group: the task group to use for watching properties. + :raises TypeError: if a property declares itself to be computed, but is not + an instance of ComputedProperty. + """ + for prop in thing.properties.values(): + if prop.is_computed: + computed_property = prop.get_descriptor() + if not isinstance(computed_property, ComputedProperty): + msg = "Computed properties must be instances of ComputedProperty." + raise TypeError(msg) + await task_group.start(computed_property._watch_for_changes, thing) diff --git a/src/labthings_fastapi/properties.py b/src/labthings_fastapi/properties.py index 1515b8b2..d4e82a79 100644 --- a/src/labthings_fastapi/properties.py +++ b/src/labthings_fastapi/properties.py @@ -415,6 +415,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. @@ -535,6 +540,31 @@ def is_resettable(self, obj: Owner | None) -> bool: """ return BaseProperty.reset is not self.__class__.reset + def publish(self, obj: Owner) -> Value: + r"""Publish the property's value to the message broker. + + This is intended to remove the need for code that looks like + ``thing.prop = thing.prop`` to manually trigger publication of the + property's value to the message broker. + + On functional and data properties, it is equivalent to the above. + On computed properties (which are cached by default), it will first + trigger recomputation - see the overridden method there. + + :param obj: the Thing on which we are publishing the property. + :return: the current value of the property. + """ + value = self.__get__(obj) + obj._thing_server_interface.publish( + Message( + thing=obj.name, + affordance=self.name, + message_type="property", + payload=value, + ) + ) + return value + def add_to_fastapi(self, app: FastAPI, thing: Owner) -> None: """Add this action to a FastAPI app, bound to a particular Thing. @@ -1215,6 +1245,23 @@ 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 manually recompute computed properties. + """ + obj = self.owning_object_or_error() + prop = self.get_descriptor() + prop.publish(obj) + def validate(self, value: Any) -> Value: """Use the validation logic in `self.model`. diff --git a/src/labthings_fastapi/server/__init__.py b/src/labthings_fastapi/server/__init__.py index 208a4765..81b2da9c 100644 --- a/src/labthings_fastapi/server/__init__.py +++ b/src/labthings_fastapi/server/__init__.py @@ -14,6 +14,7 @@ from types import MappingProxyType from typing import Any, AsyncGenerator, Optional, Self, TypeVar, overload +import anyio import uvicorn from anyio.from_thread import BlockingPortal from fastapi import APIRouter, FastAPI, Request @@ -24,6 +25,7 @@ from pydantic_core import PydanticSerializationError from labthings_fastapi.actions import ActionManager +from labthings_fastapi.computed_properties import start_watching_computed_properties from labthings_fastapi.exceptions import GlobalLockBusyError from labthings_fastapi.global_lock import GlobalLock from labthings_fastapi.logs import configure_thing_logger @@ -463,8 +465,20 @@ async def lifespan(self, app: FastAPI) -> AsyncGenerator[None, None]: "exception": e, } raise + # We create a task group for watching computed properties + # This uses the AsyncExitStack instead of `with tg:` to save on + # indentation. + tg = anyio.create_task_group() + await stack.enter_async_context(tg) + for thing in self.things.values(): + await start_watching_computed_properties(thing, tg) yield + # Closing streams in the message broker should close websockets + # and terminate the coroutines that recompute computed properties + # (those coroutines run in the task group created above) + await self.message_broker.close_streams() + self.blocking_portal = None def _things_view_router(self) -> APIRouter: diff --git a/src/labthings_fastapi/thing_server_interface.py b/src/labthings_fastapi/thing_server_interface.py index 3acbf011..7433c936 100644 --- a/src/labthings_fastapi/thing_server_interface.py +++ b/src/labthings_fastapi/thing_server_interface.py @@ -20,7 +20,7 @@ from labthings_fastapi.exceptions import FeatureNotEnabledError, ServerNotRunningError from labthings_fastapi.global_lock import GlobalLock -from labthings_fastapi.message_broker import Message +from labthings_fastapi.message_broker import Message, MessageBroker if TYPE_CHECKING: from labthings_fastapi.actions import ActionManager @@ -156,6 +156,11 @@ def publish(self, message: Message) -> None: except ServerNotRunningError: pass # If the server isn't running yet, we can't publish events. + @property + def message_broker(self) -> MessageBroker: + """The message broker, coordinating pub/sub messaging.""" + return self._get_server().message_broker + @property def settings_folder(self) -> str: """The path to a folder where persistent files may be saved.""" diff --git a/tests/test_computed_properties.py b/tests/test_computed_properties.py new file mode 100644 index 00000000..cad20ad6 --- /dev/null +++ b/tests/test_computed_properties.py @@ -0,0 +1,376 @@ +"""Test computed properties in isolation.""" + +from typing import Any, Literal + +import anyio +import pytest +from anyio.to_thread import run_sync +from httpx2 import ASGITransport, AsyncClient + +import labthings_fastapi as lt +from labthings_fastapi.computed_properties import ( + ComputedProperty, + access_wrapper, + computed_property, + start_watching_computed_properties, +) +from labthings_fastapi.exceptions import PropertyNotObservableError +from labthings_fastapi.message_broker import Message +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" + + @property + def non_labthings_property(self) -> int: + """An attribute that's not a LabThings property.""" + return 99 + + +class MyBrokenThing(MyThing): + @computed_property + def broken(self) -> str: + return self.unobservable + + +class RecursionBomb(lt.Thing): + @computed_property + def a(self) -> str: + return self.b + " blah" + + @computed_property + def b(self) -> str: + return self.a + "blah" + + +@pytest.fixture +def server(): + """Return a server with a MyThing attached.""" + server = lt.ThingServer.from_things({"thing": MyThing}) + return server + + +@pytest.fixture +async def aclient(server): + """Create an asynchronous test client.""" + transport = ASGITransport(app=server.app) + async with server.lifespan(server.app): + async with AsyncClient( + transport=transport, base_url="http://testserver" + ) as aclient: + yield aclient + + +def test_access_wrapper(): + """Test the access wrapper works as expected.""" + thing = create_thing_without_server(MyThing) + accesses: list[str] = [] + expected_accesses: list[str] = [] + wrapper = access_wrapper(thing, accesses.append) + + # Accessing an observable property should: + # 1. call the callback, i.e. append the name + # 2. return the value of the property + assert wrapper.quantity == 0 + expected_accesses.append("quantity") + assert accesses == expected_accesses + + # Check that the object and wrapper stay in sync + thing.quantity = 1 + assert wrapper.quantity == 1 + expected_accesses.append("quantity") + assert accesses == expected_accesses + + # Computed properties are observable and thus allowed + assert wrapper.double == 2 + expected_accesses.append("double") + assert accesses == expected_accesses + + # The wrapper should not allow attribute writes + with pytest.raises(AttributeError): + wrapper.quantity = -1 + + # The wrapper should not allow attribute access to + # anything other than observable properties + with pytest.raises(PropertyNotObservableError): + _ = wrapper.unobservable + with pytest.raises(TypeError, match="is not a property"): + _ = wrapper.non_labthings_property + + +def test_running_without_event_loop(): + """The computed properties should work as expected without an event loop. + + The property is recomputed each time it's accessed. + """ + thing = create_thing_without_server(MyThing) + + thing.quantity = 1 + assert thing.double == 2 + + assert thing.selected == 1 + + thing.selector = "q2" + thing.quantity2 = 10 + assert thing.selected == 10 + + +def recompute(obj: lt.Thing, name: str) -> tuple[Any, set[str]]: + """Recompute a computed property and output the dependencies. + + This duplicates logic that lives in `ComputedProperty._watch_for_changes` + so we can run it in isolation to test the access wrapper. It didn't make + sense to put this in a function, because these 4 lines of code are spread + over a for loop in the `ComputedProperty` function, which isn't needed + here. + + :prop obj: the Thing we're working with. + :prop name: the name of the property. + :return: the value of the property, and a set of the names of its + dependencies. + """ + dependencies: set[str] = set() + wrapper = access_wrapper(obj, callback=dependencies.add) + prop = obj.properties[name].get_descriptor() + assert isinstance(prop, ComputedProperty) + # This will compute the value, and populate `dependencies` via the callback + value = prop.fget(wrapper) + return value, dependencies + + +def test_dependencies(): + """Test that the right dependencies are detected when we recompute the property.""" + thing = create_thing_without_server(MyThing) + + thing.quantity = 1 + assert recompute(thing, "double") == (2, {"quantity"}) + thing.selector = "q1" + assert recompute(thing, "selected") == (1, {"selector", "quantity"}) + + thing.selector = "q2" + thing.quantity2 = 10 + assert recompute(thing, "selected") == (10, {"selector", "quantity2"}) + + +def test_unobservable(): + """Check computed properties error if they depend on unobservable quantities.""" + thing = create_thing_without_server(MyBrokenThing) + with pytest.raises(PropertyNotObservableError): + recompute(thing, "broken") + + +async def test_notifications_double(server, aclient): + """Check that we get a notification if a dependent property changes. + + This test also verifies that the cached value is being updated. + """ + thing = server.things["thing"] + assert isinstance(thing, MyThing) + prop = thing.properties["double"].get_descriptor() + assert isinstance(prop, ComputedProperty) + + # This test uses `aclient` so the computed property should already have + # been initialised, meaning the cache should be active + assert thing.double == 0 # the default value + assert prop._latest_values[thing] == 0 # check cache explicitly + + # Subscribe to updates from "double" + send, recv = anyio.create_memory_object_stream[Message](max_buffer_size=1) + await server.message_broker.subscribe("thing", "double", send) + + # Listen for updates and set `quantity` to trigger one + # Note that we need a task group because we must start + # listening before we set the property, otherwise the buffer overflows. + # This most closely matches the way notifications will work in a + # real server - we could also use a bigger buffer and do it sequentially, + # which is done in the next test. + async with anyio.create_task_group() as tg: + handle = tg.start_soon(recv.receive) + await run_sync(thing.properties["quantity"].set, 42) + expected_value = 84 + message = handle.return_value + + # Check the message described the update properly + assert message.affordance == "double" + assert message.thing == "thing" + assert message.message_type == "property" + assert message.payload == expected_value + + # Check the computed property now has the right value + assert thing.double == expected_value + # Explicitly check that the cache is up to date + assert prop._latest_values[thing] == expected_value + + +async def test_notifications_publish(server, aclient): + """Check that publishing a computed property recomputes it.""" + thing = server.things["thing"] + assert isinstance(thing, MyThing) + + # Subscribe to updates from "double" + send, recv = anyio.create_memory_object_stream[Message](max_buffer_size=2) + await server.message_broker.subscribe("thing", "double", send) + + for _ in range(3): + # Publish its value explicitly + await run_sync(thing.properties["double"].publish) + message = await recv.receive() + # Note: we do this 3 times to make sure the messages weren't somehow + # buffered in the stream from start-up: 3 is bigger than the buffer + # size. + + # Check the message described the update properly + assert message.affordance == "double" + assert message.thing == "thing" + assert message.message_type == "property" + assert message.payload == 0 + + +async def test_notifications_selected(server, aclient): + """Check that we get a notification if a dependent property changes.""" + thing = server.things["thing"] + assert isinstance(thing, MyThing) + + # Subscribe to updates from "selected" + # Note: we use a buffer size of 2 so we can call `recv.receive()` *after* + # we trigger the update. This avoids the need to use a task group every + # time. We test using a buffer size of 1 in test_notifications_double. + send, recv = anyio.create_memory_object_stream[Message](max_buffer_size=2) + await server.message_broker.subscribe("thing", "selected", send) + + # Set `quantity` to trigger a recomputation and a message + await run_sync(thing.properties["quantity"].set, 42) + message = await recv.receive() + + # Check the message described the update properly + assert message.affordance == "selected" + assert message.thing == "thing" + assert message.message_type == "property" + assert message.payload == 42 + + # Setting `quantity2` should not trigger an update. + await run_sync(thing.properties["quantity2"].set, 1) + await anyio.sleep(0.05) + with pytest.raises(anyio.WouldBlock): + # There should be no message in the receive stream + recv.receive_nowait() + + # Change the value of `quantity2` to guard against + # messages sitting in the stream. This shouldn't notify, + # because quantity2 is not yet a dependency + await run_sync(thing.properties["quantity2"].set, 2) + # Now change `selector`, which should trigger a recomputation. + await run_sync(thing.properties["selector"].set, "q2") + message = await recv.receive() + + # Check the message described the update properly + assert message.affordance == "selected" + assert message.thing == "thing" + assert message.message_type == "property" + assert message.payload == 2 + + # `quantity2` should now be a dependency, so it ought to trigger + # an update. + await run_sync(thing.properties["quantity2"].set, 3) + message = await recv.receive() + + # Check the message described the update properly + assert message.affordance == "selected" + assert message.thing == "thing" + assert message.message_type == "property" + assert message.payload == 3 + + # Setting `quantity` should not trigger an update any more. + await run_sync(thing.properties["quantity"].set, 1) + await anyio.sleep(0.05) + with pytest.raises(anyio.WouldBlock): + # There should be no message in the receive stream + recv.receive_nowait() + + +def test_recursion_noloop(): + """Two computed properties that depend on each other form an infinite loop. + + This would be a recursion problem even if they were just regular properties. + This test can be safely skipped if it turns out to be slow, but it felt + appropriate to demonstrate that it's a RecursionError even without any + of the computed property code running (without the event loop, there should + be no difference between this and a regular property). + """ + thing = create_thing_without_server(RecursionBomb) + with pytest.raises(RecursionError): + _ = thing.a + + +def test_recursion_in_server(): + """Computed properties that form an infinite loop should crash the server. + + This isn't a problem that's unique to computed properties, so I don't think + there is any need to specifically guard against it: it would be an error + if it was just a plain `builtins.property` (see above test). + """ + server = lt.ThingServer.from_things({"thing": RecursionBomb}) + with pytest.RaisesGroup(RecursionError, flatten_subgroups=True): + with server.test_client(): + pass + + +async def test_start_watching_computed_properties(mocker): + """Check the function to watch all computed properties on a Thing works. + + This checks that the relevant coroutine was started for each computed + property. + """ + thing = create_thing_without_server(MyThing) + tg = mocker.AsyncMock() + await start_watching_computed_properties(thing, tg) + assert tg.start.call_count == 2 + assert await tg.start.has_awaits( + [ + mocker.call( + thing.properties["double"].get_descriptor()._watch_for_changes, thing + ), + mocker.call( + thing.properties["selected"].get_descriptor()._watch_for_changes, thing + ), + ] + ) + + +async def test_start_watching_computed_properties_error(mocker): + """Check the error condition on start_watching_computed_properties. + + This should only be possible if someone creates a property that + claims to be computed, but isn't a ComputedProperty. + """ + + class BadProp(lt.DataProperty[lt.Thing, int]): + is_computed: bool = True + + class BadThing(lt.Thing): + bad: int = BadProp(default=0) # type: ignore + + thing = create_thing_without_server(BadThing) + tg = mocker.AsyncMock() + with pytest.raises(TypeError): + await start_watching_computed_properties(thing, tg) diff --git a/tests/test_property.py b/tests/test_property.py index 34955f98..dd58880a 100644 --- a/tests/test_property.py +++ b/tests/test_property.py @@ -31,6 +31,7 @@ NotConnectedToServerError, PropertyRedefinitionError, ) +from labthings_fastapi.message_broker import Message from labthings_fastapi.properties import ( BaseProperty, DataProperty, @@ -708,6 +709,53 @@ def prop_df(self) -> int: assert Example.prop_df.default_factory() == 42 +def test_notifications(mocker): + """Test property change notifications are sent at the appropriate time.""" + + class Example(lt.Thing): + dataprop: int = lt.property(default=0) + + @lt.property + def funcprop(self) -> int: + return 42 + + @funcprop.setter + def _set_funcprop(self, val: int) -> None: + pass + + funcprop.observable = True + + example = create_thing_without_server(Example) + publish = mocker.spy(example._thing_server_interface, "publish") + + # Check that writing to a data property publishes a message + example.dataprop = 42 + assert publish.call_count == 1 + message = publish.call_args.args[0] + assert isinstance(message, Message) + assert message.thing == example.name + assert message.affordance == "dataprop" + assert message.payload == 42 + + # Check that explicitly publishing a data property duplicates the message + example.properties["dataprop"].publish() + assert publish.call_count == 2 + message2 = publish.call_args.args[0] + assert message2 == message + + # Writing to a functional property, even an observable one, does not publish + example.funcprop = 0 + assert publish.call_count == 2 + + # Functional properties may be published explicitly + example.properties["funcprop"].publish() + assert publish.call_count == 3 + message = publish.call_args.args[0] + assert message.thing == example.name + assert message.affordance == "funcprop" + assert message.payload == 42 + + def test_bad_reset_decorator(): """Check that a resetter can't have the same name as the property."""