[Backport release-0.6] Add drivers for EnerGenie Power Management System (PMS) products

docs/source/reference/package-apis/drivers/energenie.md

@@ -0,0 +1 @@
+../../../../../packages/jumpstarter-driver-energenie/README.md

docs/source/reference/package-apis/drivers/index.md

@@ -21,6 +21,7 @@ Drivers that control the power state and basic operation of devices:
   control
 * **[DUT Link](dutlink.md)** (`jumpstarter-driver-dutlink`) - [DUT Link
   Board](https://github.com/jumpstarter-dev/dutlink-board) hardware control
+* **[Energenie PDU](energenie.md)** (`jumpstarter-driver-energenie`) - Energenie PDUs
 
 ### Communication Drivers
 
@@ -79,6 +80,7 @@ General-purpose utility drivers:
 can.md
 corellium.md
 dutlink.md
+energenie.md
 flashers.md
 http.md
 network.md

packages/jumpstarter-driver-energenie/.gitignore

@@ -0,0 +1,3 @@
+__pycache__/
+.coverage
+coverage.xml

packages/jumpstarter-driver-energenie/README.md

@@ -0,0 +1,78 @@
+# EnerGenie
+
+Drivers for EnerGenie products.
+
+## EnerGenie driver
+
+This driver provides a client for the [EnerGenie Programmable power switch](https://energenie.com/products.aspx?sg=239). The driver was tested on EG-PMS2-LAN device only but should be easy to support other devices.
+
+**driver**: `jumpstarter_driver_energenie.driver.EnerGenie`
+
+## Installation
+
+```shell
+pip3 install --extra-index-url https://pkg.jumpstarter.dev/simple/ jumpstarter-driver-energenie
+```
+
+### Configuration
+
+```yaml
+export:
+  power:
+    type: jumpstarter_driver_energenie.driver.EnerGenie
+    config:
+      host: "192.168.0.1"
+      password: "password"
+      slot: "1"
+```
+
+### Config parameters
+
+| Parameter | Description | Type | Required | Default |
+|-----------|-------------|------|----------|---------|
+| host | The ip address of the EnerGenie system | string | yes | None |
+| password | The password of the EnerGenie system | string | no | None |
+| slot | The slot number to be managed, 1, 2, 3, 4 | int | yes | 1 |
+
+### PowerClient API
+
+The EnerGenie driver provides a `PowerClient` with the following API:
+
+```{eval-rst}
+.. autoclass:: jumpstarter_driver_power.client.PowerClient()
+    :no-index:
+    :members: on, off
+```
+
+### Examples
+
+Powering on and off a device
+
+```{testcode}
+:skipif: True
+client.power.on()
+time.sleep(1)
+client.power.off()
+```
+
+### CLI
+
+```bash
+$ sudo uv run jmp exporter shell -c ./packages/jumpstarter-driver-energenie/examples/exporter.yaml
+
+$$ j
+Usage: j [OPTIONS] COMMAND [ARGS]...
+
+  Generic composite device
+
+Options:
+  --help  Show this message and exit.
+
+Commands:
+  power   Generic power
+
+$$ j power on
+
+
+$$ exit
+```

packages/jumpstarter-driver-energenie/examples/exporter.yaml

@@ -0,0 +1,14 @@
+apiVersion: jumpstarter.dev/v1alpha1
+kind: ExporterConfig
+metadata:
+  namespace: default
+  name: demo
+endpoint: grpc.jumpstarter.192.168.0.203.nip.io:8082
+token: "<token>"
+export:
+  power:
+    type: jumpstarter_driver_energenie.driver.EnerGenie
+    config:
+      host: "192.168.1.51"
+      password: "1"
+      slot: 1

packages/jumpstarter-driver-energenie/jumpstarter_driver_energenie/driver.py

@@ -0,0 +1,92 @@
+
+from collections.abc import AsyncGenerator
+from dataclasses import dataclass, field
+
+import requests
+from jumpstarter_driver_power.driver import PowerInterface, PowerReading
+
+from jumpstarter.driver import Driver, export
+
+
+@dataclass(kw_only=True)
+class EnerGenie(PowerInterface, Driver):
+    """
+    driver for the EnerGenie Programmable surge protector with LAN interface.
+
+    This driver was tested on EG-PMS2-LAN device only but should be easy to support other devices.
+    """
+
+    host: str | None = field(default=None)
+    password: str | None = field(default="1")
+    slot: int = 1
+
+    def login(self):
+        """
+        Log in to the programmable power switch.
+
+        :return: True if login is successful, False otherwise.
+        """
+        login_url = f"{self.base_url}/login.html"
+        try:
+            response = requests.post(login_url, data={"pw": self.password}, timeout=10)
+            return response.status_code == 200
+        except (requests.exceptions.ConnectionError, requests.exceptions.Timeout,
+                requests.exceptions.RequestException) as e:
+            self.logger.error(f"Login failed: {str(e)}")
+            return False
+
+    def __post_init__(self):
+        if hasattr(super(), "__post_init__"):
+            super().__post_init__()
+        # Programmable power switch initialitzation. The EG-PMS2-LAN device has up to 4 slots.
+        if self.slot < 1 or self.slot > 4:
+            raise ValueError("Slot must be between 1 and 4")
+        if self.host is None:
+            raise ValueError("Host must be specified")
+        self.logger.debug(f"Using Host: {self.host}, Slot: {self.slot}")
+        self.base_url = f"http://{self.host}"
+
+
+    def set_switch(self, switch_number, state):
+        """
+        Set the state of a specific switch.
+
+        :param switch_number: The switch number (1, 2, etc.).
+        :param state: The state to set (1 for ON, 0 for OFF).
+        :return: True if the operation is successful, False otherwise.
+        """
+        if state not in [0, 1]:
+            self.logger.error(f"Invalid state: {state}")
+            return False
+
+        if self.login():
+            self.logger.debug("Login successful!")
+        else:
+            self.logger.debug("Login failed!")
+            return False
+        data = {f"cte{switch_number}": state}
+        try:
+            response = requests.post(self.base_url, data=data, timeout=10)
+            if response.status_code != 200:
+                self.logger.error(f"Set switch {switch_number} to {state} state failed!")
+                return False
+        except (requests.exceptions.ConnectionError, requests.exceptions.Timeout,
+                requests.exceptions.RequestException) as e:
+            self.logger.error(f"Set switch failed: {str(e)}")
+            return False
+
+        self.logger.debug(f"Set switch {switch_number} to {state} state")
+
+        return True
+
+    @export
+    def on(self) -> None:
+        self.set_switch(self.slot, 1)
+
+    @export
+    def off(self) -> None:
+        self.set_switch(self.slot, 0)
+
+    @export
+    def read(self) -> AsyncGenerator[PowerReading, None]:
+        raise NotImplementedError

packages/jumpstarter-driver-energenie/jumpstarter_driver_energenie/driver_test.py

@@ -0,0 +1,46 @@
+from pytest_httpserver import HTTPServer
+
+from .driver import EnerGenie
+from jumpstarter.common.utils import serve
+
+
+def test_drivers_energenie(httpserver: HTTPServer):
+    # Configure mock responses
+    # 1. Login response - Match raw data string
+    httpserver.expect_request(
+        "/login.html",
+        method="POST",
+        data="pw=1"
+    ).respond_with_data("Login successful") # Defaults to status 200
+
+    # 2. Response for turning ON switch 1 - Match raw data string
+    httpserver.expect_request(
+        "/",
+        method="POST",
+        data="cte1=1"
+    ).respond_with_data("Switch turned ON") # Defaults to status 200
+
+    # 3. Response for turning OFF switch 1 - Match raw data string
+    httpserver.expect_request(
+        "/",
+        method="POST",
+        data="cte1=0"
+    ).respond_with_data("Switch turned OFF") # Defaults to status 200
+
+    # Get the mock server's host and port
+    host = f"{httpserver.host}:{httpserver.port}"
+
+    # Create EnerGenie instance with the mock server's URL
+    instance = EnerGenie(host=host)
+
+    with serve(instance) as client:
+        client.on()
+        client.off()
+
+    # check_assertions will verify that all expected requests were received
+    # in the correct order and that no unexpected requests arrived.
+    try:
+        httpserver.check_assertions()
+    except AssertionError as e:
+        print(f"httpserver assertions FAILED: {e}")
+        raise # Re-raise the assertion error to fail the test

packages/jumpstarter-driver-energenie/pyproject.toml

@@ -0,0 +1,37 @@
+[project]
+name = "jumpstarter-driver-energenie"
+dynamic = ["version", "urls"]
+description = "Energenie is an advanced surge protector with power management features"
+readme = "README.md"
+license = { text = "Apache-2.0" }
+authors = [
+    { name = "Enric Balletbo i Serra", email = "eballetbo@redhat.com" }
+]
+requires-python = ">=3.11"
+dependencies = [
+    "anyio>=4.6.2.post1",
+    "jumpstarter",
+    "jumpstarter-driver-power"
+]
+
+[project.entry-points."jumpstarter.drivers"]
+EnerGenie = "jumpstarter_driver_energenie.driver:EnerGenie"
+
+[dependency-groups]
+dev = [
+    "pytest-cov>=6.0.0",
+    "pytest>=8.3.3",
+    "pytest-httpserver>=1.0.0",
+]
+
+[tool.hatch.metadata.hooks.vcs.urls]
+Homepage = "https://jumpstarter.dev"
+source_archive = "https://github.com/jumpstarter-dev/repo/archive/{commit_hash}.zip"
+
+[tool.hatch.version]
+source = "vcs"
+raw-options = { 'root' = '../../'}
+
+[build-system]
+requires = ["hatchling", "hatch-vcs"]
+build-backend = "hatchling.build"

pyproject.toml

@@ -11,6 +11,7 @@ jumpstarter-driver-can = { workspace = true }
 jumpstarter-driver-composite = { workspace = true }
 jumpstarter-driver-corellium = { workspace = true }
 jumpstarter-driver-dutlink = { workspace = true }
+jumpstarter-driver-energenie = { workspace = true }
 jumpstarter-driver-flashers = { workspace = true }
 jumpstarter-driver-http = { workspace = true }
 jumpstarter-driver-raspberrypi = { workspace = true }