plesty.lib.service.resource_manager

Resource manager for tracking and allocating lockable device resources.

Classes

ResourceManager

Manage exclusive access to resources.

Module Contents

class plesty.lib.service.resource_manager.ResourceManager

Manage exclusive access to resources.

Registered resources may be a flat collection or a nested tree:

["Dev1/port0/line0", "Dev1/port0/line1"]
{"Dev1/port0/line0", "Dev1/port0/line1"}

{
    "Dev1": {
        "port0": ["line0", "line1"],
        "port1": ["line0", "line1"],
    }
}

Client requests may be a string ("Dev1/port0" — every leaf under that subtree; "Dev1/port0/line0" — one exact leaf), a flat collection (["Dev1/port0", "Dev1/port1/line0"]), or a nested dictionary tree ({"Dev1": {"port0": ["line0"]}}).

Initialize with empty resource and client tracking sets.

_resources: set[str]

_locked_resources: set[str]

_clients: dict[str, list[str]]

register_resources(resources: Any) → None

Register known resources.

If resources is a tree dict, all leaf paths are registered. If resources is a flat list/set, items are registered as canonical strings.

Parameters:: resources (Any)
Return type:: None

normalize_resources(resources: Any) → list[str]

Convert any supported resource request format into canonical leaf paths.

If known resources are registered, subtree paths are expanded to matching leaves. If no known resources are registered, paths are preserved as given.

Parameters:: resources (Any)
Return type:: list[str]

client_resources(client_id: str) → list[str]

Return the resources currently allocated to a client.

Parameters:: client_id (str)
Return type:: list[str]

resource_available(resources: Any) → bool

Check if the requested resources are available for allocation.

Parameters:: resources (Any)
Return type:: bool

allocate_resource(resources: Any, client_id: str) → bool

Attempt to allocate resources to a client.

Parameters:

resources (Any) – Resource request in str/list/set/dict tree format.
client_id (str) – Identifier used to track the allocation.

Return type:

bool

release_resource(client_id: str) → None

Release a previously allocated resource set.

Parameters:: client_id (str) – The identifier for the resource allocation to release.
Return type:: None

attach_resource(payload: dict, client_id: str) → dict

Attach already allocated resources to an incoming payload.

Parameters:

payload (dict) – The incoming request payload that may contain resource requirements.
client_id (str) – Client identifier used for resource tracking.

Return type:

dict

_unknown_resources(resources: list[str]) → list[str]

Return requested resources that are not registered.

If no known resources were registered, all requests are accepted. This keeps backward compatibility with the original behavior.

Parameters:: resources (list[str])
Return type:: list[str]

_flatten_resource_spec(spec: Any, prefix: str = '', *, expand_known: bool) → list[str]

Flatten str/list/set/dict tree resource specs into canonical string paths.

When expand_known=True, a path that is not an exact leaf but is a known subtree prefix expands to all registered leaf resources below it.

Parameters:

spec (Any)
prefix (str)
expand_known (bool)

Return type:

list[str]

_expand_path(path: str) → list[str]

Expand a path to registered leaf resources.

Examples with registered leaves:: Dev1/port0 -> Dev1/port0/line0, Dev1/port0/line1 Dev1/port0/line0 -> Dev1/port0/line0

Parameters:: path (str)
Return type:: list[str]

static _is_iterable_collection(value: Any) → bool

Parameters:: value (Any)
Return type:: bool

static _join_path(prefix: str, value: str) → str

Parameters:

prefix (str)
value (str)

Return type:

str

static _canonical_path(path: str, prefix: str = '') → str

Parameters:

path (str)
prefix (str)

Return type:

str