openflexure_microscope_server.utilities

module documentation

(source)

Utility functions and classes.

Class	`VersionData`	A BaseModel containing the information about the server version.
Function	`coerce_thing_selector`	Return a valid selector for a mapping of things or None if empty.
Function	`get_invocation_logs`	Get logs from an ongoing action invocation.
Function	`is_path_safe`	Check if a file path has any unsafe elements in it.
Function	`load_patched_config`	Load a json configuration file, if it a patch, return the full config.
Function	`make_name_safe`	Replace unsafe characters in a filename or identifier with underscores.
Function	`merge_patch`	Merge json data using the methods from IETF RFC 7396.
Function	`requires_lock`	Decorate a class method so that it requires the class lock to run.
Function	`resolve_path_from_dir`	Convert a relative or abs path specified in one dir to the working dir.
Function	`robust_version_strings`	Return a version string and information on its source.
Function	`save_invocation_logs`	Save the invocation logs from and ongoing action.
Constant	`COMMIT_REGEX`	Undocumented
Constant	`LOGGER`	Undocumented
Constant	`P`	Undocumented
Constant	`REF_REGEX`	Undocumented
Constant	`REPO_DIR`	Undocumented
Type Variable	`LockableClass`	Undocumented
Type Variable	`T`	Undocumented
Type Alias	`JSONScalar`	Undocumented
Type Alias	`JSONType`	Undocumented
Function	`_get_hash_from_git_dir`	Get the commit hash from a .git directory directly.
Function	`_get_hash_from_git_ref`	Get the commit hash from a ref in the .git directory.
Function	`_get_version_from_toml`	Get the version string from a pyproject.toml file.
Function	`_is_lock_like`	Check if an object is a lock.
Function	`_sanitise_reserved`	Check a name against Windows reserved names.
Constant	`_NAME_UNSAFE_PATTERN`	Undocumented
Constant	`_POSIX_UNSAFE_PATTERN`	Undocumented
Constant	`_WINDOWS_RESERVED_NAMES`	Undocumented
Constant	`_WINDOWS_UNSAFE_PATTERN`	Undocumented

def coerce_thing_selector(thing_mapping: Mapping[str, lt.Thing], selected: str | None, default: str | None = None) -> str | None: (source) ¶

Return a valid selector for a mapping of things or None if empty.

Parameters
thing_mapping:`Mapping[str, lt.Thing]`	A mapping of str -> `Thing`
selected:`str \| None`	The key of the selected thing (or `None`)
default:`str \| None`	The default key to fallback to if selected is `None` or the key is missing

Returns
`str \| None`	`None` if the mapping is empty, else a key that is in the mapping. Order of preference is: selected, default, the first key.

def get_invocation_logs(thing: lt.Thing, logged_since: float = 0.0) -> list[logging.LogRecord]: (source) ¶

Get logs from an ongoing action invocation.

This must be called from the action's context (thread).

Note that only 1000 logs are stored in LabThings. For very long tasks that log regularly it may be good to periodically poll this and request only the logs since the last log was created.

Parameters
thing:`lt.Thing`	The thing that the action is called from.
logged_since:`float`	The timestamp that records must come after. This could be the `record.created` time of the last log.

Returns
`list[logging.LogRecord]`	A list of `LogRecord` objects.

def is_path_safe(unsafe_path_string: str) -> bool: (source) ¶

Check if a file path has any unsafe elements in it.

The path is not coerced into a safe form because if we ask for a file to be written to a location we shouldn't be changing that location as we have no "consent" to write to this other location.

Parameters
unsafe_path_string:`str`	The original path string to sanitise.

Returns
`bool`	A boolean indicating if any unsafe features were found.

def load_patched_config(config_path: str) -> dict: (source) ¶

Load a json configuration file, if it a patch, return the full config.

Load a json file. If it does not contains the "base_config_file" key then return the data as read.

If the configuration specifies a "base_config_file" but no "patch" then return the data from reading base_config_file as json.

If the configuration specifies a "base_config_file" and "patch" then apply the patch to the data in base_config_file and return the result. The patching method is merge_patch()

Parameters
config_path:`str`	The path of the configuration file.

Returns
`dict`	The contents of the configuration file after loading and patching the specified base configuration file if applicable.

def make_name_safe(unsafe_name_string: str) -> str: (source) ¶

Replace unsafe characters in a filename or identifier with underscores.

This excludes all path separators, ensuring the result is safe to use as a standalone filename or identifier component.

This also handles Windows reserved names and trailing dots/spaces.

Parameters
unsafe_name_string:`str`	The original name string to sanitise.

Returns
`str`	A version of the input string safe to use as a file name or identifier.

@overload

def merge_patch(target: dict[str, JSONType], patch: dict[str, JSONType], enforce_dict: Literal[True]) -> dict[str, JSONType]:

@overload

def merge_patch(target: JSONType, patch: JSONType, enforce_dict: Literal[False]) -> JSONType:

@overload

def merge_patch(target: JSONType, patch: JSONType) -> JSONType:

(source) ¶

Merge json data using the methods from IETF RFC 7396.

Primarily this is designed to merge dictionaries, but IETF RFC 7396 provides defined methods for handling non-dictionary data loaded from JSON, so this has been provided in full.

Parameters
target:`JSONType`	The target object
patch:`JSONType`	The patch to be applied
enforce_dict:`bool`	Boolean, set True enforces that the target and patch are both dictionaries.

Returns
`JSONType`	Undocumented

def requires_lock(method: Callable[Concatenate[LockableClass, P], T]) -> Callable[Concatenate[LockableClass, P], T]: (source) ¶

Decorate a class method so that it requires the class lock to run.

The class should have a reentrant lock with the name self._lock.

def resolve_path_from_dir(path: str, directory: str) -> str: (source) ¶

Convert a relative or abs path specified in one dir to the working dir.

This also expands any environment variables.

Parameters
path:`str`	The specified path (could be absolute or relative)
directory:`str`	The directory from which the path was specified

Returns
`str`	The normalised path.

def robust_version_strings() -> VersionData: (source) ¶

Return a version string and information on its source.

Returning a python version string is not without problems. If the package is installed in an editable way, often METADATA is never updated. This provides 4 ways to provide a version string:

If both a Git directory and a pyproject.toml are available. Return the version

string from the toml file and the git hash from the .git directory as its source.
If a pyproject.toml is available but no Git directory then this is likely an

installation from source. Return the version string from the toml file and return the literal string "TOML" as its source.
If a Git directory is available but no pyproject.toml then something is very

weird - log an error. Then return "Undefined" as the version string, but still return the Git hash as the source.
Finally if neither a Git directory nor a pyproject.toml are available then the server has been installed from a distribution package (i.e. a wheel). In this case use the standard library function importlib.metadata.version for the version string (prepending a "v") and return "Dist" as its source.

Returns
`VersionData`	a VersionData instance with the version string and source.

def save_invocation_logs(filename: str, thing: lt.Thing, last_saved: float = 0.0, append: bool = True) -> float: (source) ¶

Save the invocation logs from and ongoing action.

This must be called from the action's context (thread).

Parameters
filename:`str`	The filename to write the logs to.
thing:`lt.Thing`	The thing that the action is called from.
last_saved:`float`	The timestamp that records must come after.
append:`bool`	If True (default) append to the file if it exists.

Returns
`float`	The timestamp of the most recent log.

COMMIT_REGEX = (source) ¶

Undocumented

Value

re.compile(r'[0-9a-f]{40}')

LOGGER = (source) ¶

Undocumented

Value

logging.getLogger(__name__)

P = (source) ¶

Undocumented

Value

ParamSpec('P')

REF_REGEX = (source) ¶

Undocumented

Value

re.compile(r'^ref:\s(.*)$')

REPO_DIR = (source) ¶

Undocumented

Value

os.path.dirname(os.path.dirname(os.path.dirname(__file__)))

LockableClass = (source) ¶

Undocumented

Value

TypeVar('LockableClass')

T = (source) ¶

Undocumented

Value

TypeVar('T')

JSONScalar: TypeAlias = (source) ¶

Undocumented

Value

str | int | float | bool | None

JSONType: TypeAlias = (source) ¶

Undocumented

Value

dict[str, JSONType] | list[JSONType] | JSONScalar

def _get_hash_from_git_dir(git_dir_path: str) -> str: (source) ¶

Get the commit hash from a .git directory directly.

This function doesn't rely on GIT being on the PATH or even installed. It doesn't require any packages outside the Python standard library. It directly inspects the .git directory to get the commit hash:

If the repository is on a detached HEAD then the hash will be in the file

.git/HEAD. A detached HEAD is could be from checking out a tag or checking out a commit directly.
In normal operation .git/HEAD points to a reference or "ref". This ref file

contains the hash.

Parameters
git_dir_path:`str`	The path to the .git directory.

Returns
`str`	The hash, or "Undefined" if there is any error, errors are logged not raised.

def _get_hash_from_git_ref(git_ref_path: str) -> str: (source) ¶

Get the commit hash from a ref in the .git directory.

For more detail see _get_hash_from_git_dir

Parameters
git_ref_path:`str`	The full path to the ref file in the .git directory.

Returns
`str`	The hash, or "Undefined" if there is any error, errors are logged not raised.

def _get_version_from_toml(toml_path: str) -> str: (source) ¶

Get the version string from a pyproject.toml file.

Parameters
toml_path:`str`	The path to the toml file.

Returns
`str`	The version string directly as reported in the toml file, or "Undefined" if there is any error, errors are logged not raised.

def _is_lock_like(obj: Any) -> bool: (source) ¶

Check if an object is a lock.

Cannot use isinstance(obj, threading.RLock), may be possible to use isinstance(obj, threading._RLock). But this uses a private method and may break. Instead making the check that both "acquire" and "release" exist.

def _sanitise_reserved(name: str, is_filename: bool = True) -> str: (source) ¶

Check a name against Windows reserved names.

Parameters
name:`str`	The component to check.
is_filename:`bool`	Whether the component is a filename. If True, names will be forced into lowercase.

Returns
`str`	The sanitised component, in lower case.

_NAME_UNSAFE_PATTERN = (source) ¶

Undocumented

Value

re.compile(r'[^a-zA-Z0-9_\.-]')

_POSIX_UNSAFE_PATTERN = (source) ¶

Undocumented

Value

re.compile(r'[^a-zA-Z0-9_\.-/ ]')

_WINDOWS_RESERVED_NAMES: set[str] = (source) ¶

Undocumented

Value

set(['CON',
     'PRN',
     'AUX',
     'NUL',
     'COM1',
     'COM2',
     'COM3',
...

_WINDOWS_UNSAFE_PATTERN = (source) ¶

Undocumented

Value

re.compile(r'[^a-zA-Z0-9_\.-:/ \\]')