chore: Unvendor pydantic_argparse and wire up uv correctly

docs/api.md

@@ -47,11 +47,3 @@ All available transports are documented in {doc}`../transports`.
     :members:
     :show-inheritance:
 ```
-
-## opennetzteil.netzteil
-
-```{eval-rst}
-.. automodule:: opennetzteil.netzteil
-    :members:
-    :show-inheritance:
-```

docs/automation.md

@@ -11,14 +11,12 @@ SPDX-License-Identifier: CC0-1.0
 Power supplies are mostly used for power cycling the current device under test.
 There is no limit in accessing power supplies, e.g. voltage or current settings can be controlled as well.
 
-Own drivers can be included by implementing the {class}`opennetzteil.netzteil.BaseNetzteil` interface[^1].
+Own drivers can be included by implementing the {class}`gallia.power_supply.BasePowerSupply` interface.
 On the commandline there is the `--power-supply` argument to specify a relevant power supply.
 Further, there is `--power-cycle` to automatically power-cycle the device under test.
-There is an experimental cli tool `opennetzteil` included in `gallia`.
+There is an experimental cli tool `netzteil` included in `gallia`.
 This cli tool can be used to control all supported power supplies via the cli.
 
-[^1]: `opennetzteil` is included and shipped with `gallia`.
-
 The argument for `--power-supply` is a URI of the following form:
 
 ``` text

pyproject.toml

@@ -30,6 +30,7 @@ dependencies = [
     "aiofiles >=24.1.0,<25.0",
     "aiosqlite >=0.18",
     "argcomplete >=2,<4",
+    "boltons>=24.1.0",
     "construct >=2.10,<3.0",
     "more-itertools >=10.3.0,<11.0.0",
     "msgspec >=0.11,<0.19",
@@ -39,15 +40,15 @@ dependencies = [
     "zstandard >=0.19",
 ]
 
-[project.entry_points."gallia_plugins"]
+[project.entry-points."gallia_plugins"]
 "uds" = "gallia.plugins.uds:UDSPlugin"
 "xcp" = "gallia.plugins.xcp:XCPPlugin"
 
 [project.scripts]
-"gallia" = "gallia.cli:main"
-"netzteil" = "opennetzteil.cli:main"
-"cursed-hr" = "cursed_hr.cursed_hr:main"
-"hr" = "hr:main"
+"gallia" = "gallia.cli.gallia:main"
+"netzteil" = "gallia.cli.netzteil:main"
+"cursed-hr" = "gallia.cli.cursed_hr:main"
+"hr" = "gallia.cli.hr:main"
 
 [dependency-groups]
 dev = [
@@ -78,10 +79,26 @@ strict = true
 plugins = [
     "pydantic.mypy"
 ]
+exclude = [
+    "src/gallia/pydantic_argparse/.*"
+]
+
+# TODO: Make this compatible to --strict.
+[[tool.mypy.overrides]]
+module = ["gallia.pydantic_argparse"]
+check_untyped_defs = true
+disallow_untyped_defs = true
+disallow_incomplete_defs = true
+disallow_untyped_decorators = true
+disallow_any_unimported = true
+warn_return_any = true
+warn_unused_ignores = true
+no_implicit_optional = true
 
 [tool.ruff]
 target-version = "py311"
 line-length = 100
+extend-exclude = ["src/gallia/pydantic_argparse"]
 
 [tool.ruff.lint]
 select = [

src/cursed_hr/cursed_hr.py → src/gallia/cli/cursed_hr.py

@@ -1316,8 +1316,8 @@ def add_entries(message: str, prefix: str = "") -> None:
         add_entries("")
         add_entries("Action keys:")
 
-        for option in options:
-            add_entries(f"{options[option]}", f"    {option}: ")
+        for key, val in options.items():
+            add_entries(f"{key}", f"    {val}: ")
 
         add_entries("")
         add_entries("Log level keys (ordered from highest to lowest):")

src/gallia/cli.py → src/gallia/cli/gallia.py

@@ -14,8 +14,6 @@
 
 import argcomplete
 from pydantic import Field, create_model
-from pydantic_argparse import ArgumentParser
-from pydantic_argparse import BaseCommand as PydanticBaseCommand
 from pydantic_core import PydanticUndefined
 
 from gallia import exitcodes
@@ -25,6 +23,8 @@
 from gallia.config import Config, load_config_file
 from gallia.log import Loglevel, setup_logging
 from gallia.plugins.plugin import CommandTree, load_commands, load_plugins
+from gallia.pydantic_argparse import ArgumentParser
+from gallia.pydantic_argparse import BaseCommand as PydanticBaseCommand
 from gallia.utils import get_log_level
 
 setup_logging(Loglevel.DEBUG)
@@ -82,6 +82,7 @@ def _create_parser_from_tree(
     )
 
 
+# TODO: Move this function into some CLI library package.
 def create_parser(
     commands: type[BaseCommand] | MutableMapping[str, CommandTree | type[BaseCommand]],
 ) -> ArgumentParser[PydanticBaseCommand]:
@@ -118,6 +119,7 @@ def get_command(config: BaseCommandConfig) -> BaseCommand:
     return cmd(config)
 
 
+# TODO: Move this function into some CLI library package.
 def parse_and_run(
     commands: type[BaseCommand] | MutableMapping[str, CommandTree | type[BaseCommand]],
     auto_complete: bool = True,

src/opennetzteil/cli.py → src/gallia/cli/netzteil.py

@@ -6,15 +6,15 @@
 
 from pydantic import field_serializer, model_validator
 
-from gallia.cli import parse_and_run
+from gallia.cli.gallia import parse_and_run
 from gallia.command import AsyncScript
 from gallia.command.base import AsyncScriptConfig, ScannerConfig
 from gallia.command.config import Field, Idempotent
-from gallia.powersupply import PowerSupplyURI
+from gallia.power_supply import power_supply_drivers
+from gallia.power_supply.base import BasePowerSupplyDriver
+from gallia.power_supply.uri import PowerSupplyURI
 from gallia.transports import TargetURI
 from gallia.utils import strtobool
-from opennetzteil import netzteile
-from opennetzteil.netzteil import BaseNetzteil
 
 
 class CLIConfig(AsyncScriptConfig):
@@ -38,8 +38,8 @@ def serialize_target_uri(self, target_uri: TargetURI | None) -> Any:
 
     @model_validator(mode="after")
     def check_power_supply_requirements(self) -> Self:
-        for netzteil in netzteile:
-            if self.power_supply.product_id == netzteil.PRODUCT_ID:
+        for driver in power_supply_drivers:
+            if self.power_supply.product_id == driver.PRODUCT_ID:
                 break
         else:
             raise ValueError(f"powersupply {self.power_supply.product_id} is not supported")
@@ -60,10 +60,10 @@ def __init__(self, config: CLIConfig):
         super().__init__(config)
         self.config: CLIConfig = config
 
-    async def _client(self) -> BaseNetzteil:
-        for netzteil in netzteile:
-            if self.config.power_supply.product_id == netzteil.PRODUCT_ID:
-                return await netzteil.connect(self.config.power_supply, timeout=1.0)
+    async def _client(self) -> BasePowerSupplyDriver:
+        for driver in power_supply_drivers:
+            if self.config.power_supply.product_id == driver.PRODUCT_ID:
+                return await driver.connect(self.config.power_supply, timeout=1.0)
 
         assert False
 

src/gallia/command/base.py

@@ -27,7 +27,8 @@
 from gallia.db.handler import DBHandler
 from gallia.dumpcap import Dumpcap
 from gallia.log import add_zst_log_handler, get_logger, tz
-from gallia.powersupply import PowerSupply, PowerSupplyURI
+from gallia.power_supply import PowerSupply
+from gallia.power_supply.uri import PowerSupplyURI
 from gallia.services.uds.core.exception import UDSException
 from gallia.transports import BaseTransport, TargetURI
 from gallia.utils import camel_to_snake, get_file_log_level

src/gallia/command/config.py

@@ -21,11 +21,11 @@
 
 from pydantic import BeforeValidator
 from pydantic.fields import _FromFieldInfoInputs
-from pydantic_argparse import BaseCommand
-from pydantic_argparse.utils.field import ArgFieldInfo
 from pydantic_core import PydanticUndefined
 
 from gallia.config import Config
+from gallia.pydantic_argparse import BaseCommand
+from gallia.pydantic_argparse.utils.field import ArgFieldInfo
 from gallia.utils import unravel, unravel_2d
 
 

src/gallia/plugins/__init__.py

@@ -0,0 +1,4 @@
+# SPDX-FileCopyrightText: AISEC Pentesting Team
+#
+# SPDX-License-Identifier: Apache-2.0
+

src/gallia/power_supply/__init__.py

@@ -0,0 +1,68 @@
+# SPDX-FileCopyrightText: AISEC Pentesting Team
+#
+# SPDX-License-Identifier: Apache-2.0
+
+import asyncio
+from collections.abc import Awaitable, Callable
+from typing import Self
+
+from gallia.log import get_logger
+from gallia.power_supply.base import BasePowerSupplyDriver
+from gallia.power_supply.devices.rs.hmc804 import HMC804
+from gallia.power_supply.uri import PowerSupplyURI
+
+power_supply_drivers: list[type[BasePowerSupplyDriver]] = [HMC804]
+
+logger = get_logger(__name__)
+
+
+class PowerSupply:
+    def __init__(self, driver: BasePowerSupplyDriver, channel_id: int | list[int]) -> None:
+        self.channel_id = channel_id
+        self.driver = driver
+        self.mutex = asyncio.Lock()
+
+    @classmethod
+    async def connect(cls, target: PowerSupplyURI) -> Self:
+        if target.product_id == "":
+            raise ValueError("no device_id specified")
+
+        for driver in power_supply_drivers:
+            if target.product_id == driver.PRODUCT_ID:
+                client = await driver.connect(target, timeout=1.0)
+                return cls(client, target.channel)
+        raise ValueError(f"{target.product_id} is not supported")
+
+    async def _power(self, op: bool) -> None:
+        assert self.driver
+        if isinstance(self.channel_id, list):
+            for id_ in self.channel_id:
+                if id_ == 0:
+                    await self.driver.set_master(op)
+                else:
+                    await self.driver.set_output(id_, op)
+        elif isinstance(self.channel_id, int):
+            if self.channel_id == 0:
+                await self.driver.set_master(op)
+            else:
+                await self.driver.set_output(self.channel_id, op)
+
+    async def power_up(self) -> None:
+        logger.info("power up")
+        await self._power(True)
+
+    async def power_down(self) -> None:
+        logger.info("power down")
+        await self._power(False)
+
+    async def power_cycle(
+        self,
+        sleep: float = 2.0,
+        callback: Callable[[], Awaitable[None]] | None = None,
+    ) -> None:
+        async with self.mutex:
+            await self.power_down()
+            await asyncio.sleep(sleep)
+            await self.power_up()
+            if callback is not None:
+                await callback()

src/opennetzteil/netzteil.py → src/gallia/power_supply/base.py

@@ -2,15 +2,13 @@
 #
 # SPDX-License-Identifier: Apache-2.0
 
-from __future__ import annotations
-
 from abc import ABC, abstractmethod
-from typing import Any
+from typing import Any, Self
 
 from gallia.transports import TargetURI
 
 
-class BaseNetzteil(ABC):
+class BasePowerSupplyDriver(ABC):
     #: The product_id is used to choose the relevant implementation.
     PRODUCT_ID = ""
 
@@ -20,7 +18,7 @@ def __init__(self, target: TargetURI, timeout: float | None) -> None:
         self.ident = ""
 
     @classmethod
-    async def connect(cls, target: TargetURI, timeout: float | None) -> BaseNetzteil:
+    async def connect(cls, target: TargetURI, timeout: float | None) -> Self:
         """Connects to ``target`` and checks for connectivity using :meth:`probe()`."""
         nt = cls(target, timeout)
         await nt.probe()

src/opennetzteil/devices/rs/hmc804.py → src/gallia/power_supply/devices/rs/hmc804.py

@@ -2,17 +2,15 @@
 #
 # SPDX-License-Identifier: Apache-2.0
 
-from __future__ import annotations
-
 import asyncio
 from typing import Any
 
+from gallia.power_supply.base import BasePowerSupplyDriver
+from gallia.power_supply.exceptions import OperationNotSupportedError
 from gallia.utils import strtobool
-from opennetzteil.exceptions import OperationNotSupportedError
-from opennetzteil.netzteil import BaseNetzteil
 
 
-class HMC804(BaseNetzteil):
+class HMC804(BasePowerSupplyDriver):
     """Rohde&Schwarz power supply: R&S HMC804x
     https://www.rohde-schwarz.com/de/produkte/messtechnik/dc-netzgeraete/rs-hmc804x-dc-netzgeraeteserie_63493-61542.html
     """

src/gallia/power_supply/uri.py

@@ -0,0 +1,32 @@
+# SPDX-FileCopyrightText: AISEC Pentesting Team
+#
+# SPDX-License-Identifier: Apache-2.0
+
+from functools import partial
+
+from gallia.log import get_logger
+from gallia.transports import TargetURI
+
+logger = get_logger(__name__)
+
+
+class PowerSupplyURI(TargetURI):
+    @property
+    def id(self) -> int:
+        if "id" in self.qs:
+            return int(self.qs["id"][0], 0)
+        return 0
+
+    @property
+    def channel(self) -> int | list[int]:
+        if "channel" in self.qs:
+            if len(ch := self.qs["channel"]) == 1:
+                return int(ch[0], 0)
+            return list(map(partial(int, base=0), ch))
+        return 0
+
+    @property
+    def product_id(self) -> str:
+        if "product_id" in self.qs:
+            return self.qs["product_id"][0]
+        return ""