API#

Base client#

class lvmnps.nps.core.NPSClient[source]#

Bases: ABC

Base NPS client.

async all_off()[source]#

Turns off all outlets.

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

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)[source]#

Retrieves an outlet by ID or name.

async list_running_scripts() → dict[int, str][source]#
Return type:

dict[int, str]

Returns a list of running scripts.

abstract async refresh()[source]#

Refreshes the list of outlets.

async run_script(name, *args, **kwargs) → int[source]#
Return type:

int

Runs a user script.

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

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()[source]#

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

async stop()[source]#

Performs any necessary operations to gracefully disconnect from the NPS.

async stop_script(thread_num=None)[source]#

Stops a running script.

abstract async verify()[source]#

Checks that the NPS is connected and responding.

pydantic model lvmnps.nps.core.OutletModel[source]#

Bases: BaseModel

A model for an outlet status.

Config:

  • arbitrary_types_allowed: bool = True

Fields:
field id: int [Required]#
field name: str [Required]#
field normalised_name: str = ''#
field state: bool = False#
async off()[source]#

Sets the state of the outlet to “off”.

async on()[source]#

Sets the state of the outlet to “on”.

set_client(nps)[source]#

Sets the NPS client.

Implementations#

Digital Loggers Inc#

class lvmnps.nps.implementations.dli.DLIClient(*args, **kwargs) → None[source]#

Bases: NPSClient

An NPS client for a Digital Loggers switch.

async list_running_scripts() → dict[int, str][source]#
Return type:

dict[int, str]

Returns a mapping of running thread number to script name.

async list_scripts()[source]#

Retrieves the list of user scripts.

async refresh()[source]#

Refreshes the list of outlets.

async run_script(name, *args, check_exists=True) → int[source]#

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()[source]#

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

async stop_script(thread_num=None)[source]#

Stops a running script.

Parameters:

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

async verify()[source]#

Checks that the NPS is connected and responding.

pydantic model lvmnps.nps.implementations.dli.DLIOutletModel[source]#

Bases: OutletModel

A model for a DLI outlet status.

Config:

  • arbitrary_types_allowed: bool = True

Fields:
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[source]#

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()[source]#

Refreshes the list of outlets.

async setup()[source]#

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

async verify()[source]#

Checks that the NPS is connected and responding.

Actor#

class lvmnps.actor.NPSActor(*args, schema=None, log=None, **kwargs)[source]#

Bases: LVMActor

LVM network power switches base actor.

async start(**kwargs)[source]#

Starts the actor.

async stop()[source]#

Stops the actor.

Tools#

class lvmnps.tools.APIClient(*args, **kwargs) → None[source]#

Bases: object

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

lvmnps.tools.get_outlet_by_id(outlet_data, id)[source]#

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)[source]#

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)[source]#

Returns a normalised name for an outlet.