diff --git a/src/openflexure_microscope_server/server.py b/src/openflexure_microscope_server/server.py index ab84d4cf..dc261851 100644 --- a/src/openflexure_microscope_server/server.py +++ b/src/openflexure_microscope_server/server.py @@ -11,6 +11,7 @@ from labthings_picamera2.thing import StreamingPiCamera2 from .things.autofocus import AutofocusThing from .things.camera_stage_mapping import CameraStageMapper from .things.system_control import SystemControlThing +from .things.settings_manager import SettingsManager from .serve_static_files import add_static_files import openflexure_microscope_server @@ -22,6 +23,7 @@ thing_server.add_thing(SangaboardThing(), "/stage/") thing_server.add_thing(AutofocusThing(), "/autofocus/") thing_server.add_thing(CameraStageMapper(), "/camera_stage_mapping/") thing_server.add_thing(SystemControlThing(), "/system_control/") +thing_server.add_thing(SettingsManager(), "/settings/") try: add_static_files(thing_server.app) except RuntimeError: diff --git a/src/openflexure_microscope_server/things/settings_manager.py b/src/openflexure_microscope_server/things/settings_manager.py new file mode 100644 index 00000000..d60f7ffd --- /dev/null +++ b/src/openflexure_microscope_server/things/settings_manager.py @@ -0,0 +1,114 @@ +""" +OpenFlexure Settings Management + +This module provides some settings management across the other Things, and +for code that currently lives in clients but needs to persist settings on +the server. +""" +from collections.abc import Mapping +from typing import Any, Mapping, MutableMapping, Optional, Sequence +from fastapi import HTTPException +from labthings_fastapi.dependencies.metadata import ThingStates +from labthings_fastapi.thing import Thing +from labthings_fastapi.decorators import thing_action, thing_property +from pydantic import BaseModel + + +def recursive_update(old_dict: MutableMapping, update: Mapping): + """Update a dictionary recursively""" + for k, v in update.items(): + if isinstance(v, Mapping): + if ( + k in old_dict + and isinstance(old_dict[k], MutableMapping) + ): + recursive_update(old_dict[k], v) + else: + old_dict[k] = dict(**v) + else: + old_dict[k] = v + + +def nested_dict_get(data: dict, key: Sequence[str], create=False) -> Any: + """Index a dict-of-dicts with a sequence of strings""" + subdict = data + for k in key: + if k not in subdict and create: + subdict[k] = {} + subdict = subdict[k] + return subdict + + +class SettingsManager(Thing): + @thing_property + def external_metadata(self) -> Mapping: + """External metadata stored in the server's settings""" + return self.thing_settings.get("external_metadata", {}) + + @thing_action + def update_external_metadata(self, data: Mapping, key: Optional[str] = None) -> None: + """Add or replace keys in the external metadata. + + The data supplied will be merged into the existing dictionary + recursively, i.e. if a key exists, it will be added to rather than + replaced. + + If a key is supplied, we will treat `data` as being relative to that + key, i.e. calling this action with `data={"a":1 }, key="foo/bar"` is + equivalent to calling it with `data={"foo": {"bar": {"a": 1}}}`. + """ + metadata = self.external_metadata + subdict = metadata + # If we're operating only on a part of the dict, make sure + # that key exists, and find it. + if key: + subdict = nested_dict_get(subdict, key.split("/"), create=True) + recursive_update(subdict, data) + self.thing_settings["external_metadata"] = metadata + + @thing_action + def delete_external_metadata(self, key: str) -> None: + """Delete a key from the stored metadata. + + The key may contain forward slashes, which are understood to separate + levels of the dictionary - i.e. `'a/c'` will remove the `c` key from a + dictionary that looks like: `{'a': {'c': 1}, 'b': {'d': 2}}` + """ + metadata = self.external_metadata + try: + subdict = nested_dict_get(metadata, key.split("/")[:-1]) + del subdict[key.split("/")[-1]] + except KeyError: + raise HTTPException( + status_code=404, + detail="The specified key '{key}' was not found" + ) + self.thing_settings["external_metadata"] = metadata + + @thing_action + def get_things_state(self, thing_states_metadata: ThingStates) -> Mapping: + """Metadata summarising the current state of all Things in the server""" + return thing_states_metadata + + @thing_property + def external_metadata_in_state(self) -> Sequence[str]: + """A list of strings that are included in the "state" metadata""" + return self.thing_settings.get("external_metadata_in_state", []) + @external_metadata_in_state.setter + def external_metadata_in_state(self, keys: Sequence[str]): + """Set the keys from external metadata that are returned in state""" + self.thing_settings["external_metadata_in_state"] = keys + + @property + def thing_state(self) -> Mapping: + state = {} + metadata = self.external_metadata + for k in self.external_metadata_in_state: + try: + kl = k.split("/") + v = nested_dict_get(metadata, kl) + subdict = nested_dict_get(state, kl[:-1], create=True) + subdict[kl[-1]] = v + except KeyError: + continue + return state \ No newline at end of file