diff --git a/labgrid/run/run.py b/labgrid/run/run.py index 71584b7667b7e108df410db09a773113bba72cb2..c75f47ceabaaceb946f211be9d474e393e6cd435 100644 --- a/labgrid/run/run.py +++ b/labgrid/run/run.py @@ -1,17 +1,24 @@ #! /usr/bin/env python3 from __future__ import annotations -import argparse -import os -from argparse import REMAINDER, Action, ArgumentParser -from contextlib import contextmanager -from dataclasses import dataclass +from argparse import REMAINDER, Action, ArgumentParser, Namespace +from dataclasses import dataclass, replace from os import environ, linesep from pathlib import Path, PurePath from shlex import join from subprocess import CalledProcessError from sys import argv, stderr, stdout -from typing import Collection, Iterator, Protocol, Tuple +from typing import ( + Any, + Final, + Iterator, + MutableMapping, + Optional, + Sequence, + Text, + TypeVar, + Union, +) from python.runfiles import Runfiles @@ -19,6 +26,8 @@ from bazel.labgrid.strategy import transition from bazel.labgrid.util.target import TemporaryDirectory from labgrid import Environment +T = TypeVar("T") + class RunfileNotFoundError(FileNotFoundError): pass @@ -32,6 +41,65 @@ def runfile(path: str) -> Path: return resolved +@dataclass(frozen=True) +class State: + desired: str + initial: str | None = None + final: str | None = None + + +class StateAction(Action): + __default: Final[MutableMapping] = {} + + def __init__( + self, + option_strings: Sequence[str], + dest: str, + default: str | None, + const: Optional[T] = None, + help: Optional[str] = None, + ): + dest, found, attr = dest.partition(".") + if not found: + raise ValueError("`dest` must contain `.`") + self.attr = attr + super().__init__( + option_strings, + dest, + nargs=0, + const=const, + default=default, + type=None, + choices=None, + required=False, + help=help, + metavar=None, + ) + + @property + def default(self) -> State: + return State(**self.__default) + + @default.setter + def default(self, value: str | None) -> None: + self.__default[self.attr] = value + + def __call__( + self, + parser: ArgumentParser, + namespace: Namespace, + values: Union[str, Sequence[Any], None], + option_string: Optional[Text] = None, + ) -> None: + assert isinstance(values, str) + try: + state = getattr(namespace, self.dest) + except AttributeError: + state = self.default + state = replace(state, **{self.attr: values}) + setattr(namespace, self.dest, state) + + def arguments(prsr: ArgumentParser) -> None: prsr.add_argument( "program", metavar="PROG", help="The program to run on the device.", type=Path @@ -48,20 +116,23 @@ def arguments(prsr: ArgumentParser) -> None: prsr.add_argument( "--initial-state", help="The state to transition the LabGrid strategy to initially", - dest="initial", + dest="state.initial", action=StateAction, + default=environ.get("LG_INITIAL_STATE", None), ) prsr.add_argument( "--state", help="The state to transition the LabGrid strategy to", - dest="desired", + dest="state.desired", action=StateAction, + default=environ["LG_STATE"], ) prsr.add_argument( "--final-state", help="The state to transition the LabGrid strategy to at the end", - dest="final", + dest="state.final", action=StateAction, + default=environ.get("BZL_LG_FINAL_STATE", None), ) prsr.add_argument( "--mktemp", @@ -116,18 +187,6 @@ class Get: return path -@dataclass -class State: - initial: str | None = environ.get("LG_INITIAL_STATE", None) - desired: str = environ["LG_STATE"] - final: str | None = environ.get("BZL_LG_FINAL_STATE", None) - - -class StateAction(argparse.Action): - def __call__(self, parser, namespace, values, option_string=None): - setattr(namespace.state, self.dest, values) - - # TODO: move this out to `bazel.labgrid.run` on day to allow downstream to re-use def run( program: Path, @@ -186,7 +245,6 @@ def main(exe: Path, *args: str) -> int: prsr = ArgumentParser( prog=str(exe), description="Runs a command on a LabGrid device." ) - prsr.set_defaults(state=State()) arguments(prsr) parsed = prsr.parse_args(args)