Initial working structure with DLI client

src/lvmnps/__init__.py

@@ -15,3 +15,7 @@
 log = get_logger(NAME)
 
 __version__ = get_package_version(path=__file__, package_name=NAME)
+
+
+from .actor import NPSActor, NPSCommand
+from .nps import DLIClient, NPSClient

src/lvmnps/exceptions.py

@@ -0,0 +1,25 @@
+#!/usr/bin/env python
+# -*- coding: utf-8 -*-
+#
+# @Author: José Sánchez-Gallego ([email protected])
+# @Date: 2023-11-22
+# @Filename: exceptions.py
+# @License: BSD 3-clause (http://www.opensource.org/licenses/BSD-3-Clause)
+
+from __future__ import annotations
+
+
+class NPSException(Exception):
+    """Base NPS exception."""
+
+
+class VerificationError(NPSException):
+    """Failed to connect to the power supply."""
+
+
+class ResponseError(NPSException):
+    """Invalid response from the power supply API."""
+
+
+class NPSWarning(UserWarning):
+    """Base NPS warning."""

src/lvmnps/nps/__init__.py

@@ -0,0 +1,12 @@
+#!/usr/bin/env python
+# -*- coding: utf-8 -*-
+#
+# @Author: José Sánchez-Gallego ([email protected])
+# @Date: 2023-11-22
+# @Filename: __init__.py
+# @License: BSD 3-clause (http://www.opensource.org/licenses/BSD-3-Clause)
+
+from __future__ import annotations
+
+from .core import NPSClient
+from .implementations import *

src/lvmnps/nps/core.py

@@ -0,0 +1,133 @@
+#!/usr/bin/env python
+# -*- coding: utf-8 -*-
+#
+# @Author: José Sánchez-Gallego ([email protected])
+# @Date: 2023-11-22
+# @Filename: core.py
+# @License: BSD 3-clause (http://www.opensource.org/licenses/BSD-3-Clause)
+
+from __future__ import annotations
+
+import abc
+
+from typing import Any, Sequence
+
+from pydantic import BaseModel, ConfigDict
+
+from lvmnps import log
+from lvmnps.tools import get_outlet_by_id, get_outlet_by_name, normalise_outlet_name
+
+
+__all__ = ["NPSClient", "OutletModel", "OutletArgType"]
+
+
+class OutletModel(BaseModel):
+    """A model for an outlet status."""
+
+    model_config = ConfigDict(arbitrary_types_allowed=True)
+
+    id: int
+    name: str
+    name_normalised: str = ""
+    state: bool = False
+
+    client: NPSClient | None = None
+
+    def model_post_init(self, __context: Any) -> None:
+        self.name_normalised = normalise_outlet_name(self.name)
+        return super().model_post_init(__context)
+
+    def set_client(self, nps: NPSClient):
+        """Sets the NPS client."""
+
+        self.client = nps
+
+    async def on(self):
+        """Sets the state of the outlet to "on"."""
+
+        if not self.client:
+            raise RuntimeError("NPS client not set.")
+
+        await self.client.set_state(self, on=True)
+
+    async def off(self):
+        """Sets the state of the outlet to "off"."""
+
+        if not self.client:
+            raise RuntimeError("NPS client not set.")
+
+        await self.client.set_state(self, on=False)
+
+
+OutletArgType = OutletModel | int | str | Sequence[str | int | OutletModel]
+
+
+class NPSClient(abc.ABC):
+    """Base NPS client."""
+
+    def __init__(self):
+        self.outlets: dict[str, OutletModel] = {}
+
+        # Time after switching an outlet on during which switching outlets on is
+        # delayed to prevent simultaneous inrush currents on power-on time.
+        self.delay: float = 1
+
+    async def setup(self):
+        """Sets up the power supply, setting any required configuration options."""
+
+    @abc.abstractmethod
+    async def verify(self):
+        """Checks that the NPS is connected and responding."""
+
+        pass
+
+    @abc.abstractmethod
+    async def refresh(self):
+        """Refreshes the list of outlets."""
+
+        pass
+
+    async def set_state(self, outlets: OutletArgType, on: bool = False):
+        """Sets the state of an outlet or list of outlets.
+
+        Parameters
+        ----------
+        outlets
+            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
+            Whether to turn the outlet on (if `True`) or off.
+
+        """
+
+        _outlets: list[OutletModel] = []
+
+        if isinstance(outlets, str) or not isinstance(outlets, Sequence):
+            outlets = [outlets]
+
+        for outlet in outlets:
+            if isinstance(outlet, str):
+                _outlets.append(get_outlet_by_name(self.outlets, outlet))
+            elif isinstance(outlet, int):
+                _outlets.append(get_outlet_by_id(self.outlets, outlet))
+            else:
+                _outlets.append(outlet)
+
+        names = [outlet.name for outlet in _outlets]
+        log.debug(f"Setting outlets {names} to state on={on}.")
+        await self._set_state_internal(_outlets, on=on)
+
+        await self.refresh()
+
+    @abc.abstractmethod
+    async def _set_state_internal(self, outlets: list[OutletModel], on: bool = False):
+        """Internal method for setting the outlet state.
+
+        This method is intended to be overridden by each specific implementation.
+        All implementations should handle switching single outlets or multiple ones,
+        and do it in a way that is safe and efficient given the hardware specifications.
+
+        """

src/lvmnps/nps/implementations/__init__.py

@@ -0,0 +1,15 @@
+#!/usr/bin/env python
+# -*- coding: utf-8 -*-
+#
+# @Author: José Sánchez-Gallego ([email protected])
+# @Date: 2023-11-22
+# @Filename: __init__.py
+# @License: BSD 3-clause (http://www.opensource.org/licenses/BSD-3-Clause)
+
+from __future__ import annotations
+
+
+__all__ = ["DLIClient"]
+
+
+from .dli import DLIClient

src/lvmnps/nps/implementations/dli.py

@@ -0,0 +1,166 @@
+#!/usr/bin/env python
+# -*- coding: utf-8 -*-
+#
+# @Author: José Sánchez-Gallego ([email protected])
+# @Date: 2023-11-22
+# @Filename: dli.py
+# @License: BSD 3-clause (http://www.opensource.org/licenses/BSD-3-Clause)
+
+from __future__ import annotations
+
+import warnings
+
+import httpx
+from pydantic import ConfigDict, SecretStr
+from pydantic.dataclasses import dataclass
+
+from lvmnps.exceptions import VerificationError
+from src.lvmnps.nps.core import NPSClient, OutletModel
+
+
+__all__ = ["DLIClient"]
+
+
+class DLIOutletModel(OutletModel):
+    """A model for an outlet status."""
+
+    index: int = 0
+    physical_state: bool = False
+    transient_state: bool = False
+    critical: bool = False
+    locked: bool = False
+    cycle_delay: float | None = None
+
+
+@dataclass
+class APIClient:
+    """A wrapper around ``httpx.AsyncClient`` to yield a new client."""
+
+    base_url: str
+    user: str
+    password: SecretStr
+
+    def __post_init__(self):
+        self.client: httpx.AsyncClient | None = None
+
+    async def __aenter__(self):
+        """Yields a new client."""
+
+        auth = httpx.DigestAuth(self.user, self.password.get_secret_value())
+        self.client = httpx.AsyncClient(
+            auth=auth,
+            base_url=self.base_url,
+            headers={},
+        )
+
+        return self.client
+
+    async def __aexit__(self, exc_type, exc, tb):
+        """Closes the client."""
+
+        if self.client and not self.client.is_closed:
+            await self.client.aclose()
+
+
+@dataclass(config=ConfigDict(extra="forbid"))
+class DLIClient(NPSClient):
+    """An NPS client for a Digital Loggers switch."""
+
+    host: str
+    port: int = 80
+    user: str = "admin"
+    password: SecretStr = SecretStr("admin")
+    api_route: str = "restapi/"
+
+    def __post_init__(self):
+        super().__init__()
+
+        self.base_url = f"http://{self.host}:{self.port}/{self.api_route}"
+        self.api_client = APIClient(self.base_url, self.user, self.password)
+
+        self.outlet: dict[str, DLIOutletModel] = {}
+
+    async def setup(self):
+        """Sets up the power supply, setting any required configuration options."""
+
+        try:
+            await self.verify()
+        except VerificationError as err:
+            warnings.warn(
+                "Cannot setup DLI. Power switch "
+                f"verification failed with error: {err}"
+            )
+            return
+
+        async with self.api_client as client:
+            # Change in-rush delay to 1 second.
+            response = await client.put(
+                url="/relay/sequence_delay/",
+                data={"value": 1},
+                headers={"X-CSRF": "x"},
+            )
+            self._validate_response(response, 204)
+
+        await self.refresh()
+
+    def _validate_response(self, response: httpx.Response, expected_code: int = 200):
+        """Validates an HTTP response."""
+
+        if response.status_code != expected_code:
+            raise VerificationError(
+                f"Request returned response with status code {response.status_code}."
+            )
+
+    async def verify(self):
+        """Checks that the NPS is connected and responding."""
+
+        async with self.api_client as client:
+            try:
+                response = await client.get(url="/", headers={"Range": "dli-depth=1"})
+            except httpx.ConnectError as err:
+                raise VerificationError(f"Failed to connect to DLI: {err}")
+
+        # 206 because we have asked for the API to only do depth=1
+        self._validate_response(response, 206)
+
+    async def refresh(self):
+        """Refreshes the list of outlets."""
+
+        url = "/relay/outlets/"
+        async with self.api_client as client:
+            response = await client.get(url=url)
+
+        self._validate_response(response)
+
+        data = response.json()
+
+        self.outlets = {}
+
+        for outlet_id in range(1, len(data) + 1):
+            outlet_data = data[outlet_id - 1]
+            outlet_data["id"] = outlet_id
+            outlet_data["index"] = outlet_id - 1
+
+            outlet = DLIOutletModel(**outlet_data)
+            outlet.client = self
+            self.outlets[outlet.name_normalised] = outlet
+
+    async def _set_state_internal(
+        self, outlets: list[DLIOutletModel], on: bool = False
+    ):
+        """Sets the state of a list of outlets."""
+
+        outlet_indices = [outlet.index for outlet in outlets]
+
+        # Use a matrix URI to set all the states at once.
+        outlet_path = "=" + ",".join(map(str, outlet_indices))
+
+        async with self.api_client as client:
+            response = await client.put(
+                url=f"/relay/outlets/{outlet_path}/state/",
+                data={"value": on},
+                headers={"X-CSRF": "x"},
+            )
+            self._validate_response(response, 207)
+
+        return