new syncops module

Author: cav71

src/luxos/asyncops.py

@@ -1,6 +1,7 @@
 from __future__ import annotations
 
 import asyncio
+import contextlib
 import functools
 import json
 import logging
@@ -126,9 +127,9 @@ def validate_message(
     port: int,
     res: dict[str, Any],
     extrakey: str | None = None,
-    minfields: None | int = None,
-    maxfields: None | int = None,
-) -> dict[str, Any]:
+    minfields: None | int = 1,
+    maxfields: None | int = 1,
+) -> Any:
     # all miner message comes with a STATUS
     for key in ["STATUS", "id", *([extrakey] if extrakey else [])]:
         if key in res:
@@ -142,17 +143,17 @@ def validate_message(
 
     n = len(res[extrakey])
     msg = None
-    if minfields and (n < minfields):
+    if (minfields is not None) and (n < minfields):
         msg = f"found {n} fields for {extrakey} invalid: " f"{n} <= {minfields}"
-    elif maxfields and (n > maxfields):
+    elif (maxfields is not None) and (n > maxfields):
         msg = f"found {n} fields for {extrakey} invalid: " f"{n} >= {maxfields}"
     if msg is None:
         return res[extrakey]
     raise exceptions.MinerCommandMalformedMessageError(host, port, msg, res)
 
 
 @wrapped
-async def logon(host: str, port: int, timeout: float | None = 3) -> str:
+async def logon(host: str, port: int, timeout: float | None = None) -> str:
     timeout = TIMEOUT if timeout is None else timeout
     res = await roundtrip(host, port, {"command": "logon"}, timeout=timeout)
 
@@ -166,7 +167,7 @@ async def logon(host: str, port: int, timeout: float | None = 3) -> str:
         )
     sessions = validate_message(host, port, res, "SESSION", 1, 1)
 
-    session = sessions[0]  # type: ignore
+    session = sessions[0]
 
     if "SessionID" not in session:
         raise exceptions.MinerCommandSessionAlreadyActive(
@@ -177,56 +178,15 @@ async def logon(host: str, port: int, timeout: float | None = 3) -> str:
 
 @wrapped
 async def logoff(
-    host: str, port: int, sid: str, timeout: float | None = 3
+    host: str, port: int, sid: str, timeout: float | None = None
 ) -> dict[str, Any]:
     timeout = TIMEOUT if timeout is None else timeout
     return await roundtrip(
         host, port, {"command": "logoff", "parameter": sid}, timeout=timeout
     )
 
 
-@wrapped
-async def execute_command(
-    host: str,
-    port: int,
-    timeout_sec: float | None,
-    cmd: str,
-    parameters: list[str] | None = None,
-    verbose: bool = False,
-    asjson: bool | None = True,
-    add_address: bool = False,
-) -> tuple[tuple[str, int], dict[str, Any]] | dict[str, Any]:
-    timeout = TIMEOUT if timeout_sec is None else timeout_sec
-    parameters = parameters or []
-
-    sid = None
-    if api.logon_required(cmd):
-        sid = await logon(host, port)
-        parameters = [sid, *parameters]
-        log.debug("session id requested & obtained for %s:%i (%s)", host, port, sid)
-    else:
-        log.debug("no logon required for command '%s' on %s:%i", cmd, host, port)
-
-    try:
-        packet = {"command": cmd}
-        if parameters:
-            packet["parameter"] = ",".join(parameters)
-        log.debug(
-            "executing command '%s' on '%s:%i' with parameters: %s",
-            cmd,
-            host,
-            port,
-            packet.get("parameter", ""),
-        )
-        ret = await roundtrip(host, port, packet, timeout=timeout, asjson=asjson)
-        log.debug("received from %s:%s: %s", host, port, str(ret))
-        return ((host, port), ret) if add_address else ret
-    finally:
-        if sid:
-            await logoff(host, port, sid)
-
-
-def _rexec_parameters(
+def parameters_to_list(
     parameters: str | list[Any] | dict[str, Any] | None = None,
 ) -> list[str]:
     if isinstance(parameters, dict):
@@ -256,9 +216,7 @@ async def rexec(
     retry: int | None = None,
     retry_delay: float | None = None,
 ) -> dict[str, Any] | None:
-    from . import api
-
-    parameters = _rexec_parameters(parameters)
+    parameters = parameters_to_list(parameters)
 
     timeout = TIMEOUT if timeout is None else timeout
     retry = RETRIES if retry is None else retry
@@ -328,3 +286,14 @@ async def rexec(
         await logoff(host, port, sid)
     if isinstance(failure, Exception):
         raise failure
+
+
+@contextlib.asynccontextmanager
+async def with_atm(host, port, enabled: bool, timeout: float | None = None):
+    res = await rexec(host, port, "atm", timeout=timeout)
+    if not res:
+        raise exceptions.MinerConnectionError(host, port, "cannot check atm")
+    current = validate_message(host, port, res, "ATM")[0]["Enabled"]
+    await rexec(host, port, "atmset", {"enabled": enabled}, timeout=timeout)
+    yield current
+    await rexec(host, port, "atmset", {"enabled": current}, timeout=timeout)

src/luxos/exceptions.py

@@ -32,3 +32,19 @@ class MinerCommandSessionAlreadyActive(MinerConnectionError):
 
 class MinerCommandMalformedMessageError(MinerConnectionError):
     pass
+
+
+class LuxosLaunchError(MinerConnectionError):
+    def __init__(self, tback: str, host: str, port: int, *args, **kwargs):
+        self.tback = tback
+        super().__init__(host, port, *args, **kwargs)
+
+    def __str__(self):
+        from .text import indent
+
+        msg = indent(str(self.tback), "| ")
+        return f"{self.address}: \n{msg}"
+
+
+class LuxosLaunchTimeoutError(LuxosLaunchError, asyncio.TimeoutError):
+    pass

src/luxos/ips.py

@@ -129,6 +129,27 @@ def iter_ip_ranges(
             cur += 1
 
 
+def ip_ranges(
+    txt: str, rsep: str = "-", gsep: str = ":"
+) -> list[tuple[str, int | None]]:
+    """return a list of ips given a text expression.
+
+    Eg.
+        >>> for ip in ip_ranges("127.0.0.1"):
+        ...     print(ip)
+        127.0.0.1
+
+        >>> for ip in ip_ranges("127.0.0.1-127.0.0.3"):
+        ...     print(ip)
+        127.0.0.1
+        127.0.0.2
+        127.0.0.3
+
+    NOTE: use the `:` (gsep) to separate ips groups, and `-` (rsep) to define a range.
+    """
+    return list(iter_ip_ranges(txt, rsep=rsep, gsep=gsep))
+
+
 def load_ips_from_csv(path: Path | str, port: int = 4028) -> list[tuple[str, int]]:
     """loads ip addresses from a csv file
 

src/luxos/syncops.py

@@ -1,25 +1,31 @@
 from __future__ import annotations
 
+import contextlib
 import json
 import logging
 import socket
 from typing import Any
 
 from luxos.api import logon_required
 
+from .asyncops import TIMEOUT, parameters_to_list, validate_message
+from .exceptions import MinerCommandSessionAlreadyActive
+
 log = logging.getLogger(__name__)
 
 
 # internal_send_cgminer_command sends a command to the
 # cgminer API server and returns the response.
 def internal_send_cgminer_command(
-    host: str, port: int, command: str, timeout_sec: int, verbose: bool
+    host: str, port: int, command: str, timeout: float | None
 ) -> dict[str, Any]:
+    timeout_sec = TIMEOUT if timeout is None else timeout
     # Create a socket connection to the server
     with socket.socket(socket.AF_INET, socket.SOCK_STREAM) as sock:
         try:
             # set timeout
-            sock.settimeout(timeout_sec)
+            if timeout_sec is not None:
+                sock.settimeout(timeout)
 
             # Connect to the server
             sock.connect((host, port))
@@ -62,22 +68,23 @@ def internal_send_cgminer_command(
 # send_cgminer_command sends a command to the cgminer API server and
 # returns the response.
 def send_cgminer_command(
-    host: str, port: int, cmd: str, param: str, timeout: int, verbose: bool
+    host: str, port: int, cmd: str, param: str, timeout: float | None = None
 ) -> dict[str, Any]:
+    timeout = TIMEOUT if timeout is None else timeout
     req = str(f'{{"command": "{cmd}", "parameter": "{param}"}}\n')
     log.debug(f"Executing command: {cmd} with params: {param} to host: {host}")
-
-    return internal_send_cgminer_command(host, port, req, timeout, verbose)
+    return internal_send_cgminer_command(host, port, req, timeout)
 
 
 # send_cgminer_simple_command sends a command with no params
 # to the miner and returns the response.
 def send_cgminer_simple_command(
-    host: str, port: int, cmd: str, timeout: int, verbose: bool
+    host: str, port: int, cmd: str, timeout: float | None = None
 ) -> dict[str, Any]:
+    timeout = TIMEOUT if timeout is None else timeout
     req = str(f'{{"command": "{cmd}"}}\n')
     log.debug(f"Executing command: {cmd} to host: {host}")
-    return internal_send_cgminer_command(host, port, req, timeout, verbose)
+    return internal_send_cgminer_command(host, port, req, timeout)
 
 
 # check_res_structure checks that the response has the expected structure.
@@ -128,28 +135,26 @@ def get_str_field(struct: dict[str, Any], name: str) -> str:
     return s
 
 
-def logon(host: str, port: int, timeout: int, verbose: bool) -> str:
+def logon(host: str, port: int, timeout: float | None = None) -> str:
     # Send 'logon' command to cgminer and get the response
-    res = send_cgminer_simple_command(host, port, "logon", timeout, verbose)
-
+    timeout = TIMEOUT if timeout is None else timeout
+    res = send_cgminer_simple_command(host, port, "logon", timeout)
+    if res["STATUS"][0]["STATUS"] == "E":
+        raise MinerCommandSessionAlreadyActive(host, port, res["STATUS"][0]["Msg"])
     # Check if the response has the expected structure
-    check_res_structure(res, "SESSION", 1, 1)
-
-    # Extract the session data from the response
-    session = res["SESSION"][0]
+    session = validate_message(host, port, res, "SESSION")[0]
+    return str(session["SessionID"])
 
-    # Get the 'SessionID' field from the session data
-    s = get_str_field(session, "SessionID")
 
-    # If 'SessionID' is empty, raise an error indicating invalid session id
-    if s == "":
-        raise ValueError("error: invalid session id")
-
-    # Return the extracted 'SessionID'
-    return s
+def logoff(
+    host: str, port: int, sid: str, timeout: float | None = None
+) -> dict[str, Any]:
+    timeout = TIMEOUT if timeout is None else timeout
+    res = send_cgminer_command(host, port, "logoff", sid, timeout)
+    return validate_message(host, port, res)
 
 
-def add_session_id_parameter(session_id, parameters):
+def add_session_id_parameter(session_id, parameters) -> list[Any]:
     # Add the session id to the parameters
     return [session_id, *parameters]
 
@@ -160,40 +165,62 @@ def parameters_to_string(parameters):
 
 
 def execute_command(
-    host: str, port: int, timeout_sec: int, cmd: str, parameters: list, verbose: bool
+    host: str,
+    port: int,
+    timeout_sec: int | float | None,
+    cmd: str,
+    parameters: str | list[Any] | dict[str, Any] | None = None,
+    verbose: bool = False,
 ):
+    timeout_sec = TIMEOUT if timeout_sec is None else timeout_sec
     # Check if logon is required for the command
     logon_req = logon_required(cmd)
 
-    try:
-        if logon_req:
-            # Get a SessionID
-            sid = logon(host, port, timeout_sec, verbose)
-            # Add the SessionID to the parameters list at the left.
-            parameters = add_session_id_parameter(sid, parameters)
+    parameters = parameters_to_list(parameters)
+
+    if logon_req:
+        # Get a SessionID
+        sid = logon(host, port, timeout_sec)
+        # Add the SessionID to the parameters list at the left.
+        parameters = add_session_id_parameter(sid, parameters)
 
-            log.debug("Command requires a SessionID, logging in for host: %s", host)
-            log.info("SessionID obtained for %s: %s", host, sid)
+        log.debug("SessionID obtained for %s: %s", host, sid)
 
-        # TODO verify this
-        elif not logon_required:  # type: ignore
-            log.debug("Logon not required for executing %s", cmd)
+    # TODO verify this
+    elif not logon_req:
+        log.debug("Logon not required for executing %s", cmd)
 
-        # convert the params to a string that LuxOS API accepts
-        param_string = parameters_to_string(parameters)
+    # convert the params to a string that LuxOS API accepts
+    param_string = ",".join(parameters)
 
-        log.debug("%s on %s with parameters: %s", cmd, host, param_string)
+    log.debug("%s on %s with parameters: %s", cmd, host, param_string)
 
-        # Execute the API command
-        res = send_cgminer_command(host, port, cmd, param_string, timeout_sec, verbose)
+    # Execute the API command
+    res = send_cgminer_command(host, port, cmd, param_string, timeout_sec)
 
-        log.debug(res)
+    log.debug(res)
 
-        # Log off to terminate the session
-        if logon_req:
-            send_cgminer_command(host, port, "logoff", sid, timeout_sec, verbose)
+    # Log off to terminate the session
+    if logon_req:
+        logoff(host, port, sid, timeout_sec)
+
+    return res
+
+
+def execute(
+    host: str,
+    port: int,
+    cmd: str,
+    parameters: str | list[Any] | dict[str, Any] | None = None,
+    timeout: float | None = None,
+):
+    return execute_command(host, port, timeout, cmd, parameters)
 
-        return res
 
-    except Exception:
-        log.exception("Error executing %s on %s", cmd, host)
+@contextlib.contextmanager
+def with_atm(host, port, enabled: bool, timeout: float | None = None):
+    res = execute(host, port, "atm", timeout=timeout)
+    current = validate_message(host, port, res, "ATM")[0]["Enabled"]
+    execute(host, port, "atmset", {"enabled": enabled}, timeout=timeout)
+    yield current
+    execute(host, port, "atmset", {"enabled": current}, timeout=timeout)