diff --git a/src/opensemantic/base/_controller.py b/src/opensemantic/base/_controller.py index 7a82208..44b6855 100644 --- a/src/opensemantic/base/_controller.py +++ b/src/opensemantic/base/_controller.py @@ -116,6 +116,25 @@ def __init__(self, **kwargs): def set_client(self, client): self._driver.set_client(client) + @property + def _offline(self) -> bool: + """True while the remote is unreachable and writes are buffered. + + The flag lives on the driver. Consumers check it on the controller + (DataToolMixin._handle_data_change) to detect the online/offline + transition, so it has to be readable here. + """ + return getattr(self._driver, "_offline", False) + + @property + def _emulate_offline(self) -> bool: + """Test hook: force the driver to behave as if the remote is down.""" + return getattr(self._driver, "_emulate_offline", False) + + @_emulate_offline.setter + def _emulate_offline(self, value: bool): + self._driver._emulate_offline = value + async def create_tool(self, params: TSDCMixin.CreateToolParams): return await self._driver.create_tool(params.tool_osw_id) diff --git a/src/opensemantic/base/_controller_mixin.py b/src/opensemantic/base/_controller_mixin.py index 45a2ed9..4ccd25d 100644 --- a/src/opensemantic/base/_controller_mixin.py +++ b/src/opensemantic/base/_controller_mixin.py @@ -12,7 +12,7 @@ from abc import abstractmethod from datetime import datetime from enum import Enum -from typing import Any, List, Optional, Union +from typing import Any, Dict, List, Optional, Union from oold.model import BaseController from pydantic import BaseModel, ConfigDict @@ -62,12 +62,7 @@ class DataToolController(DataToolMixin, DataTool): pass def __init__(self, *args, **data): super().__init__(*args, **data) self._compute_subobject_ids() - if not isinstance(getattr(self, "_channel_dict", None), dict): - object.__setattr__(self, "_channel_dict", {}) - for channel in self.get_all_channels(): - # Use node_id if available (OPC UA), fall back to uuid - key = getattr(channel, "node_id", None) or channel.uuid - self._channel_dict[key] = channel + self.rebuild_channel_dict() # Warn about unloaded channel characteristics self._check_channel_characteristics() # TODO: update wiki OpcUaServer model to include endpoint/url field @@ -84,6 +79,21 @@ def __init__(self, *args, **data): # get_osw_id() and get_iri() are inherited from OswBaseModel # via the model base class (DataTool -> Entity -> OswBaseModel) + def rebuild_channel_dict(self): + """Index all channels of self and its subdevices for fast lookup. + + Called from __init__. Call it again after mutating data_channels or + subdevices, otherwise incoming notifications for the new channels are + not routed. + """ + if not isinstance(getattr(self, "_channel_dict", None), dict): + object.__setattr__(self, "_channel_dict", {}) + self._channel_dict.clear() + for channel in self.get_all_channels(): + # Use node_id if available (OPC UA), fall back to uuid + key = getattr(channel, "node_id", None) or channel.uuid + self._channel_dict[key] = channel + # TODO: Consider moving _compute_subobject_ids to OswBaseModel def _compute_subobject_ids(self, parent_chain=None): """Compute composite osw_ids for inline subobject children. @@ -337,6 +347,123 @@ def _init_archive_database(self, db): ) return None + # -- Component hierarchy -- + + @staticmethod + def _component_refs(entity) -> list: + """Return (component_type_iri, component_instance) pairs of a tool. + + The IRIs are read from ``__iris__`` instead of the attributes so the + lazy backend resolution of ``component_instance`` is not triggered. + Resolving it would load the child with autofetch_schema=True and + generate an ad-hoc model instead of using the installed package. + """ + + def _first(value): + if isinstance(value, list): + return value[0] if value else None + return value + + refs = [] + for comp in getattr(entity, "components", None) or []: + iris = getattr(comp, "__iris__", None) or {} + instance = _first(iris.get("component_instance")) + if instance is None: + # Inline object instead of a reference + instance = comp.__dict__.get("component_instance") + if instance is None: + continue + refs.append((_first(iris.get("component_type")), instance)) + return refs + + @classmethod + def load_from_osw( + cls, + osw, + iri: str, + model_by_component_type: Optional[Dict[str, type]] = None, + default_model: Optional[type] = None, + depth: int = -1, + **kwargs, + ): + """Load a tool and its component hierarchy from the OSW backend. + + ``components`` is the OSW backend's parent/child relation for tools. + The controller-only ``subdevices`` list is populated from it, so + get_all_channels(), get_channel_owner() and archiving work on the + whole hierarchy. + + Parameters + ---------- + osw + An ``osw.core.OSW`` instance. Only ``load_entity`` and its + ``LoadEntityParam`` are used, so osw stays an optional dependency. + iri + IRI of the root tool, e.g. ``Item:OSW``. + model_by_component_type + Maps a component type IRI to the class the child is loaded as. + Use it when the children are not all of the same kind. The classes + have to be controllers (composing this mixin), otherwise the tree + traversal in get_subdevices() / get_all_channels() fails. + default_model + Controller class for children without a mapping. Defaults to + ``cls``. + depth + Component levels to follow, -1 for unlimited, 0 for the root only. + kwargs + Extra attributes to set on the root, e.g. ``url=...``. + """ + return cls._load_tree( + osw=osw, + iri=iri, + model=cls, + mapping=model_by_component_type or {}, + default_model=default_model or cls, + depth=depth, + extra=kwargs, + ) + + @classmethod + def _load_tree(cls, osw, iri, model, mapping, default_model, depth, extra): + param = type(osw).LoadEntityParam( + titles=iri, autofetch_schema=False, model_to_use=model + ) + entity = osw.load_entity(param).entities[0] + for key, value in (extra or {}).items(): + setattr(entity, key, value) + + if depth == 0: + return entity + + subdevices = [] + for component_type, ref in cls._component_refs(entity): + child_model = mapping.get(component_type, default_model) + if isinstance(ref, str): + subdevices.append( + cls._load_tree( + osw=osw, + iri=ref, + model=child_model, + mapping=mapping, + default_model=default_model, + depth=depth - 1, + extra={}, + ) + ) + elif isinstance(ref, child_model): + subdevices.append(ref) + else: + subdevices.append(child_model(ref)) + + if subdevices: + # Bypass validation: assigning to the field would revalidate every + # child (pydantic v1 replaces them with copies), which detaches the + # controller state the caller still holds a reference to. + entity.__dict__["subdevices"] = subdevices + if hasattr(entity, "rebuild_channel_dict"): + entity.rebuild_channel_dict() + return entity + def get_subdevices(self) -> list: if self.subdevices is None: return [] diff --git a/src/opensemantic/base/v1/_controller.py b/src/opensemantic/base/v1/_controller.py index 66a36ee..f473d21 100644 --- a/src/opensemantic/base/v1/_controller.py +++ b/src/opensemantic/base/v1/_controller.py @@ -116,6 +116,25 @@ def __init__(self, **kwargs): def set_client(self, client): self._driver.set_client(client) + @property + def _offline(self) -> bool: + """True while the remote is unreachable and writes are buffered. + + The flag lives on the driver. Consumers check it on the controller + (DataToolMixin._handle_data_change) to detect the online/offline + transition, so it has to be readable here. + """ + return getattr(self._driver, "_offline", False) + + @property + def _emulate_offline(self) -> bool: + """Test hook: force the driver to behave as if the remote is down.""" + return getattr(self._driver, "_emulate_offline", False) + + @_emulate_offline.setter + def _emulate_offline(self, value: bool): + self._driver._emulate_offline = value + async def create_tool(self, params: TSDCMixin.CreateToolParams): return await self._driver.create_tool(params.tool_osw_id) diff --git a/tests/test_controller.py b/tests/test_controller.py index 0dc0d99..f9173a0 100644 --- a/tests/test_controller.py +++ b/tests/test_controller.py @@ -1523,3 +1523,225 @@ async def _bench(buffered): assert ( t_buf < t_unbuf / 5 ), f"Buffered ({t_buf:.3f}s) not 5x faster than unbuffered ({t_unbuf:.3f}s)" + + +# -- Offline flag forwarding tests -- + + +def _make_offline_archive(**kwargs): + """PostgREST controller without a client, offline buffer in a temp file.""" + from opensemantic.base import PostgrestTimeSeriesDatabaseController + + return PostgrestTimeSeriesDatabaseController( + name="offline_test", + label=[Label(text="Offline Test")], + buffer_offline_location=tempfile.NamedTemporaryFile( + suffix=".sqlite", delete=False + ).name, + **kwargs, + ) + + +def test_postgrest_controller_forwards_offline_flag(): + """The driver's _offline flag must be readable on the controller.""" + db = _make_offline_archive() + assert db._offline is False + db._driver._offline = True + assert db._offline is True + + +def test_postgrest_controller_forwards_emulate_offline_flag(): + """_emulate_offline is settable on the controller and reaches the driver.""" + db = _make_offline_archive() + assert db._emulate_offline is False + db._emulate_offline = True + assert db._driver._emulate_offline is True + + +def test_handle_data_change_calls_on_archive_error_when_offline(): + """Going offline during archiving must trigger the _on_archive_error hook.""" + import datetime + from uuid import uuid4 + + from opensemantic.base import DataToolController + from opensemantic.base._controller_mixin import DataToolMixin + from opensemantic.base._model import DataChannel + + class RecordingController(DataToolController): + def _on_archive_error(self): + self._archive_errors = getattr(self, "_archive_errors", 0) + 1 + + archive = _make_offline_archive(buffered=True, buffer_batch_size=1) + archive._emulate_offline = True + + ch = DataChannel(uuid=str(uuid4()), osw_id="ch1", name="ch1") + ctrl = RecordingController( + name="test", + label=[Label(text="Test")], + data_channels=[ch], + auto_archive=True, + ) + ctrl.archive_database = archive + + async def _test(): + await ctrl._handle_data_change( + DataToolMixin.ChannelDataChangeNotificationParams( + channel=ch, + value=42.0, + timestamp=datetime.datetime.now(datetime.timezone.utc), + ) + ) + + asyncio.run(_test()) + + assert archive._offline is True + assert getattr(ctrl, "_archive_errors", 0) == 1 + + if os.path.exists(archive.buffer_offline_location): + os.unlink(archive.buffer_offline_location) + + +# -- Component hierarchy tests -- + + +class _FakeOsw: + """Minimal stand-in for osw.core.OSW used by load_from_osw.""" + + class LoadEntityParam: + def __init__(self, titles, autofetch_schema=True, model_to_use=None): + self.titles = titles if isinstance(titles, list) else [titles] + self.autofetch_schema = autofetch_schema + self.model_to_use = model_to_use + + class _Result: + def __init__(self, entities): + self.entities = entities + + def __init__(self, pages): + self.pages = pages + self.calls = [] + + def load_entity(self, param): + title = param.titles[0] + self.calls.append((title, param.model_to_use, param.autofetch_schema)) + return self._Result([param.model_to_use(**self.pages[title])]) + + +def _osw_fixture(): + from uuid import uuid4 + + from opensemantic.base._model import Component, DataChannel + + def _channel(name): + return DataChannel(uuid=str(uuid4()), osw_id=name, name=name) + + def _component(instance_iri, type_iri): + return Component( + uuid=str(uuid4()), + component_id="Component01", + component_instance=instance_iri, + component_type=type_iri, + ) + + pages = { + "Item:OSWroot": { + "name": "root", + "label": [Label(text="Root")], + "data_channels": [_channel("root_ch")], + "components": [ + _component("Item:OSWchildA", "Category:OSWtypeA"), + _component("Item:OSWchildB", "Category:OSWtypeB"), + ], + }, + "Item:OSWchildA": { + "name": "child_a", + "label": [Label(text="Child A")], + "data_channels": [_channel("a_ch")], + "components": [_component("Item:OSWgrandchild", "Category:OSWtypeA")], + }, + "Item:OSWchildB": { + "name": "child_b", + "label": [Label(text="Child B")], + "data_channels": [_channel("b_ch")], + }, + "Item:OSWgrandchild": { + "name": "grandchild", + "label": [Label(text="Grandchild")], + "data_channels": [_channel("g_ch")], + }, + } + return _FakeOsw(pages) + + +def test_load_from_osw_builds_subdevice_tree(): + from opensemantic.base import DataToolController + + osw = _osw_fixture() + root = DataToolController.load_from_osw(osw, "Item:OSWroot") + + assert root.name == "root" + assert [sub.name for sub in root.subdevices] == ["child_a", "child_b"] + assert {sub.name for sub in root.get_subdevices()} == { + "child_a", + "child_b", + "grandchild", + } + assert {ch.name for ch in root.get_all_channels()} == { + "root_ch", + "a_ch", + "b_ch", + "g_ch", + } + + +def test_load_from_osw_indexes_subdevice_channels(): + """The channel dict has to cover the subdevices wired in after __init__.""" + from opensemantic.base import DataToolController + + osw = _osw_fixture() + root = DataToolController.load_from_osw(osw, "Item:OSWroot") + + indexed = {ch.name for ch in root._channel_dict.values()} + assert indexed == {"root_ch", "a_ch", "b_ch", "g_ch"} + + +def test_load_from_osw_never_autofetches_schemas(): + """Schemas come from the installed packages, not from ad-hoc generation.""" + from opensemantic.base import DataToolController + + osw = _osw_fixture() + DataToolController.load_from_osw(osw, "Item:OSWroot") + + assert osw.calls, "no entity was loaded" + assert all(autofetch is False for _, _, autofetch in osw.calls) + + +def test_load_from_osw_selects_model_per_component_type(): + from opensemantic.base import DataToolController + + class TypeAController(DataToolController): + pass + + osw = _osw_fixture() + root = DataToolController.load_from_osw( + osw, + "Item:OSWroot", + model_by_component_type={"Category:OSWtypeA": TypeAController}, + ) + + by_name = {sub.name: sub for sub in root.get_subdevices()} + assert isinstance(by_name["child_a"], TypeAController) + assert isinstance(by_name["grandchild"], TypeAController) + assert not isinstance(by_name["child_b"], TypeAController) + + +def test_load_from_osw_respects_depth_and_extra_kwargs(): + from opensemantic.base import DataToolController + + osw = _osw_fixture() + root = DataToolController.load_from_osw( + osw, "Item:OSWroot", depth=1, auto_archive=False + ) + + assert {sub.name for sub in root.get_subdevices()} == {"child_a", "child_b"} + assert root.subdevices[0].subdevices == []