Utils

Module for MCCS utils.

class DuplicateFilter(cooldown_time=5, clean_logs_time=30, max_entries=1000)

Filters out repeating logs within a given time frame.

__init__(cooldown_time=5, clean_logs_time=30, max_entries=1000)

Initiate a new filter to remove duplicate messages.

Reduces recurring logs to one log per cooldown period that states the amount of time the log appeared in that period.

Parameters:

• cooldown_time (float) – the length of time between recurring logs. defaults to 5.

• clean_logs_time (float) – the time period for cleaning and printing any old logs.

• max_entries (int) – the total number of different log entires allowed.

filter(record)

Filter out repetitive messages to reduce error spam.

The filter logic used to keep track of each log. This checks the message to be logged, and if it has occurred before in the cooldown time it skips logging it. Once the cooldown time it’s over, it prints out the log with the number of occurrences and resets the log.

Parameters:

record (LogRecord) – the error logs

Return type:

bool

Returns:

if it logs the error or not

exception NotExportedError(message='Device not exported')

Raised when device is not exported.

__init__(message='Device not exported')

Initialise a new NotExportedError message.

Parameters:

message (str) – a bespoke message to override the default message.

class ThreadsafeCheckingMeta(name: str, bases: tuple[type], attrs: dict)

Metaclass that checks for methods being run by multiple concurrent threads.

class UniqueQueue(maxsize=0, logger=None)

A class for a queue of unique items.

__init__(maxsize=0, logger=None)

Initialize the queue with an optional logger.

Parameters:

• maxsize (int) – max queue size.

• logger (Optional[Logger]) – optional logger for logging duplicate events.

call_with_json(func, **kwargs)

Call a command with a json string.

Allows the calling of a command that accepts a JSON string as input, with the actual unserialised parameters.

For example, suppose you need to use Allocate(resources) command to tell a controller device to allocate certain stations and tiles to a subarray. Allocate accepts a single JSON string argument. Instead of

Example:

parameters={"id": id, "stations": stations, "tiles": tiles}
json_string=json.dumps(parameters)
controller.Allocate(json_string)

save yourself the trouble and

Example:

call_with_json(controller.Allocate, id=id, stations=stations, tiles=tiles)

Parameters:

• func (Callable) – the function handle to call

• kwargs (Any) – parameters to be jsonified and passed to func

Return type:

Any

Returns:

the return value of func

class json_input(schema_path=None)

Parse and validate json string input.

Method decorator that parses and validates JSON input into a python dictionary, which is then passed to the method as kwargs. The wrapped method is thus called with a JSON string, but can be implemented as if it had been passed a sequence of named arguments.

If the string cannot be parsed as JSON, an exception is raised.

For example, conceptually, MccsController.Allocate() takes as arguments a subarray id, an array of stations, and an array of tiles. In practice, however, these arguments are encoded into a JSON string. Implement the function with its conceptual parameters, then wrap it in this decorator:

Example:

@json_input
def MccsController.Allocate(id, stations, tiles):

The decorator will provide the JSON interface and handle the decoding for you.

__init__(schema_path=None)

Initialise a callable json_input object.

To function as a device method generator.

Parameters:

schema_path (Optional[str]) – an optional path to a schema against which the JSON should be validated. Not working at the moment, so leave it None.

lock_power_state(func)

Return a function that acquires a lock before calling.

This decorator acquires a named lock self._power_command_in_progress before calling the wrapped function. It then calls a named function self._evaluate_power_state after the wrapped function returns.

This function is intended to be used as a decorator to power functions:

@lock_power_state
def _on(self):
    ...

Parameters:

func (TypeVar(Wrapped, bound= Callable[..., Any])) – the wrapped function

Return type:

TypeVar(Wrapped, bound= Callable[..., Any])

Returns:

the wrapped function

threadsafe(func)

Use this method as a decorator for marking a method as threadsafe.

This tells the ThreadsafeCheckingMeta metaclass that it is okay for the decorated method to have more than one thread in it at a time. The metaclass will still raise an exception if the same thread enters the method multiple times, because re-entry is a common cause of deadlock.

Parameters:

func (Callable) – the method to be marked as threadsafe

Return type:

Callable

Returns:

the method, marked as threadsafe