146 lines
5 KiB
Python
146 lines
5 KiB
Python
"""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 socket import gethostname
|
|
from typing import Annotated, Any, MutableMapping, Optional, Sequence
|
|
from uuid import UUID, uuid4
|
|
from fastapi import Depends, HTTPException, Request
|
|
import labthings_fastapi as lt
|
|
|
|
|
|
def thing_server_from_request(request: Request) -> lt.ThingServer:
|
|
"""Wrap lt.find_thing_server."""
|
|
return lt.find_thing_server(request.app)
|
|
|
|
|
|
ThingServerDep = Annotated[lt.ThingServer, Depends(thing_server_from_request)]
|
|
|
|
|
|
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(lt.Thing):
|
|
"""Provides functionality to other Things about the current server state.
|
|
|
|
The SettingsManager is used to get information the microscope ID, the hostname
|
|
and the state of other Things.
|
|
"""
|
|
|
|
external_metadata = lt.ThingSetting(
|
|
initial_value={},
|
|
model=Mapping,
|
|
)
|
|
"""External metadata stored in the server's settings."""
|
|
|
|
external_metadata_in_state = lt.ThingSetting(
|
|
initial_value=[],
|
|
model=Sequence[str],
|
|
)
|
|
"""Keys from ``external_metadata`` that are included in the "state" metadata."""
|
|
|
|
@lt.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.external_metadata = metadata
|
|
|
|
@lt.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.external_metadata = metadata
|
|
|
|
_microscope_id: Optional[str] = None
|
|
|
|
@lt.thing_setting
|
|
def microscope_id(self) -> UUID:
|
|
"""A unique identifier for this microscope."""
|
|
if self._microscope_id is None:
|
|
self._microscope_id = str(uuid4())
|
|
return UUID(self._microscope_id)
|
|
|
|
@microscope_id.setter
|
|
def microscope_id(self, uuid: UUID):
|
|
# TODO make read only but still settable from disk
|
|
self._microscope_id = uuid
|
|
|
|
@lt.thing_property
|
|
def hostname(self) -> str:
|
|
"""The hostname of the microscope, as reported by its operating system."""
|
|
return gethostname()
|
|
|
|
@lt.thing_action
|
|
def get_things_state(self, metadata_getter: lt.deps.GetThingStates) -> Mapping:
|
|
"""Metadata summarising the current state of all Things in the server."""
|
|
return metadata_getter()
|
|
|
|
@property
|
|
def thing_state(self) -> Mapping:
|
|
"""Summary metadata describing the current state of the Thing."""
|
|
state = {
|
|
"hostname": self.hostname,
|
|
"microscope-uuid": str(self.microscope_id),
|
|
}
|
|
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
|