API

Base client

class lvmnps.nps.core.NPSClient

Bases: ABC

Base NPS client.

async all_off()

Turns off all outlets.

async cycle(outlets, delay=3) → list[OutletModel]

Turns off the selected outlets and turns them on again after a delay.

Parameters:

• outlets (Union[OutletModel, int, str, Sequence[str | int | OutletModel]]) – An outlet or list of outlets whose state will be set. An outlet can be specified by its name, number, or model instance.

• delay (float) – Number of seconds the code will wait before turning on the outlets again.

Return type:

list[OutletModel]

get(outlet)

Retrieves an outlet by ID or name.

async list_running_scripts() → dict[int, str]

Returns a list of running scripts.

Return type:

dict[int, str]

abstract async refresh()

Refreshes the list of outlets.

async run_script(name, *args, **kwargs) → int

Runs a user script.

Return type:

int

async set_state(outlets, on=False, off_after=None) → list[OutletModel]

Sets the state of an outlet or list of outlets.

Parameters:

• outlets (Union[OutletModel, int, str, Sequence[str | int | OutletModel]]) – An outlet or list of outlets whose state will be set. An outlet can be specified by its name, number, or model instance. If a list of outlet is provided the behaviour will depend on the client implementation. Outlets may be switched concurrently or sequentially, with a delay to avoid in-rush currents.

• on (bool) – Whether to turn the outlet on (if True) or off.

• off_after (float | None) – Turns off the outlet after the specified number of seconds. Only relevant if on=True.

Return type:

list[OutletModel]

async setup()

Sets up the power supply, setting any required configuration options.

async stop()

Performs any necessary operations to gracefully disconnect from the NPS.

async stop_script(thread_num=None)

Stops a running script.

abstract async verify()

Checks that the NPS is connected and responding.

pydantic model lvmnps.nps.core.OutletModel

Bases: BaseModel

A model for an outlet status.

Config:

• arbitrary_types_allowed: bool = True

Fields:

• id (int)

• name (str)

• normalised_name (str)

• state (bool)

field id: int [Required]

field name: str [Required]

field normalised_name: str = ''

field state: bool = False

async off()

Sets the state of the outlet to “off”.

async on()

Sets the state of the outlet to “on”.

set_client(nps)

Sets the NPS client.

Implementations

Digital Loggers Inc

class lvmnps.nps.implementations.dli.DLIClient(*args, **kwargs) → None

Bases: NPSClient

An NPS client for a Digital Loggers switch.

async list_running_scripts() → dict[int, str]

Returns a mapping of running thread number to script name.

Return type:

dict[int, str]

async list_scripts()

Retrieves the list of user scripts.

async refresh()

Refreshes the list of outlets.

async run_script(name, *args, check_exists=True) → int

Runs a user script.

Parameters:

• name (str) – The script name.

• args – Arguments with which to call the script function.

• check_exists (bool) – If True, checks that the script exists in the DLI before executing it.

Returns:

The thread identifier for the running script.

Return type:

thread_num

async setup()

Sets up the power supply, setting any required configuration options.

async stop_script(thread_num=None)

Stops a running script.

Parameters:

thread_num (int | None) – The thread to stop. If not specified, stops all threads.

async verify()

Checks that the NPS is connected and responding.

pydantic model lvmnps.nps.implementations.dli.DLIOutletModel

Bases: OutletModel

A model for a DLI outlet status.

Config:

• arbitrary_types_allowed: bool = True

Fields:

• critical (bool)

• cycle_delay (float | None)

• index (int)

• locked (bool)

• physical_state (bool)

• transient_state (bool)

field critical: bool = False

field cycle_delay: float | None = None

field index: int = 0

field locked: bool = False

field physical_state: bool = False

field transient_state: bool = False

NetIO

class lvmnps.nps.implementations.netio.NetIOClient(*args, **kwargs) → None

Bases: NPSClient

An NPS client for a NetIO switch.

This implementation uses the JSON API, see https://www.netio-products.com/files/NETIO-M2M-API-Protocol-JSON.pdf

async refresh()

Refreshes the list of outlets.

async setup()

Sets up the power supply, setting any required configuration options.

async verify()

Checks that the NPS is connected and responding.

Actor

class lvmnps.actor.NPSActor(*args, schema=None, log=None, **kwargs)

Bases: LVMActor

LVM network power switches base actor.

async start(**kwargs)

Starts the actor.

async stop()

Stops the actor.

Tools

class lvmnps.tools.APIClient(*args, **kwargs) → None

Bases: object

A wrapper around httpx.AsyncClient to yield a new client.

lvmnps.tools.get_outlet_by_id(outlet_data, id)

Gets an outlet by id.

Parameters:

• outlet_data (dict[str, OutletModel]) – The mapping of outlet name to outlet model data.

• id (int) – The id of the outlet to retrieve.

Returns:

The outlet matching the id.

Return type:

outlet

Raises:

ValueError – If the outlet cannot be found.

lvmnps.tools.get_outlet_by_name(outlet_data, name)

Gets an outlet from a list of outlets.

Parameters:

• outlet_data (dict[str, OutletModel]) – The mapping of outlet name to outlet model data.

• name (str) – The name of the outlet to retrieve.

Returns:

The outlet matching the input name.

Return type:

outlet

Raises:

ValueError – If the outlet cannot be found.

lvmnps.tools.normalise_outlet_name(name)

Returns a normalised name for an outlet.