Ctrl+K

luxos.utils

luxos.utils#

user facing functions for normal use

async luxos.utils.launch(addresses: list[tuple[str, int]], function: Callable[[str, int], Awaitable[Any]], batch: int = 0, asobj: bool = False) → list[LuxosLaunchError | LuxosLaunchTimeoutError | Any]#

Launch an async function on a list of (host, port) miners.

This function takes a list of (host, port) tuples (points) to miners, and for each “point” call function on it.

Parameters:

  • addresses – list of (host: str, port: int)

  • function – async callable with (host: str, port: int) call signature

  • batch – limit the number of concurrent calls (unlimited by default)

  • asobj – if True all results will be instances subclasses of LuxosLaunchBaseResult

Examples

This will gather the miners versions in a dict:

async printme(host: str, port: int):
    res = await rexec(host, port, "version"))
addresses = load_ips_from_csv("miners.csv")
asyncio.run(launch(addresses, printme))
async luxos.utils.rexec(host: str, port: int, cmd: str, parameters: str | int | float | bool | list[Any] | dict[str, Any] | None = None, timeout: float | None = None, retry: int | None = None, retry_delay: float | None = None) → dict[str, Any] | None#

Send a command to a host.

Parameters:

  • host – A string representing the host IP or a name.

  • port – An integer representing the port number to connect to.

  • cmd – A string representing the command to execute.

  • parameters – Any additional parameters for the command.

  • timeout – A float representing the maximum time in seconds to wait for a response before timing out.

  • retry – Optional. An integer representing the number of times to retry the command execution in case of failure.

  • retry_delay – Optional. A float representing the delay in seconds between each retry attempt.

Returns:

A dictionary containing the response from the execution of the command.

Raises:

Any exception that occurs during the execution of the command. –

Notes

If timeout/retry/retry_delay aren’t provided (or None), they will default to the module level values (TIMEOUT, RETRIES, and RETRIES_DELAY).

This function will handle logon/logoff automatically.

luxos.utils.validate(res: dict[str, Any], extrakey: str | None = None, minfields: None | int = None, maxfields: None | int = None) → Any#

Validate a message returned from a miner.

Parameters:

  • res – The dictionary returned from a miner.

  • extrakey – if present it will try to extract it from res.

  • cmd – A string representing the command to execute.

  • minfields – The min length of the field extrakey.

  • maxfields – The max length of the field extrakey.

Returns:

A dictionary containing the response from the execution of the command if extrakey is None, else a dict or a list of items.

Raises:
Contents