Utils

Module for utils.

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

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

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, Controller.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 Controller.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.

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