Merge pull request #16 from rvodden/fix_mandatory_default_fields

fix #14: reduce number of mandatory fields in SwitchDevice

.github/workflows/code.yml

@@ -21,7 +21,7 @@ jobs:
       - name: Run black, flake8, mypy
         uses: dls-controls/pipenv-run-action@v1
         with:
-          python-version: "3.8"
+          python-version: "3.10"
           pipenv-run: lint
 
   wheel:
@@ -56,7 +56,7 @@ jobs:
       fail-fast: false
       matrix:
         os: ["ubuntu-latest"] # can add windows-latest, macos-latest
-        python: ["3.8", "3.9"]
+        python: ["3.10", "3.11", "3.12"]
         pipenv: ["skip-lock"]
 
         include:

.github/workflows/docs.yml

@@ -26,7 +26,7 @@ jobs:
         uses: dls-controls/pipenv-run-action@v1
         with:
           pipenv-run: docs
-          python-version: "3.8"
+          python-version: "3.10"
 
       # - name: Check links resolve
       #   run: pipenv run docs -b linkcheck

docker/Dockerfile

@@ -1,5 +1,5 @@
 # python stage: install the maaspower python package
-FROM python:3.9-slim AS python
+FROM python:3.10-slim AS python
 
 RUN python -m pip install --upgrade pip && \
     pip install maaspower

setup.cfg

@@ -10,8 +10,9 @@ long_description_content_type = text/x-rst
 classifiers =
     Development Status :: 4 - Beta
     License :: OSI Approved :: Apache Software License
-    Programming Language :: Python :: 3.8
-    Programming Language :: Python :: 3.9
+    Programming Language :: Python :: 3.10
+    Programming Language :: Python :: 3.11
+    Programming Language :: Python :: 3.12
 
 [options]
 python_requires = >=3.8

src/maaspower/__main__.py

@@ -13,11 +13,12 @@
 from .devices.shell_cmd import CommandLine
 from .devices.smart_thing import SmartThing
 from .devices.web_ui import WebGui
+from .devices.web_device import WebDevice
 from .maasconfig import MaasConfig
 from .webhook import run_web_hook
 
 # avoid linter complaints
-required_to_find_subclasses = [SmartThing, CommandLine, WebGui]
+required_to_find_subclasses = [SmartThing, CommandLine, WebGui, WebDevice]
 
 cli = typer.Typer()
 yaml = YAML()

src/maaspower/devices/shell_cmd.py

@@ -10,15 +10,23 @@
 import subprocess
 from dataclasses import dataclass
 
-from typing_extensions import Literal
+from typing import Optional
+from typing_extensions import Literal, Annotated as A
 
-from maaspower.maasconfig import SwitchDevice
+from maaspower.maasconfig import RegexSwitchDevice
+from maaspower.maas_globals import desc
 
 
-@dataclass
-class CommandLine(SwitchDevice):
+@dataclass(kw_only=True)
+class CommandLine(RegexSwitchDevice):
     """A device controlled via a command line utility"""
 
+    on: A[str, desc("command line string to switch device on")]
+    off: A[str, desc("command line string to switch device off")]
+    query: A[str, desc("command line string to query device state")]
+    query_on_regex: A[str, desc("match the on status return from query")] = "on"
+    query_off_regex: A[str, desc("match the off status return from query")] = "off"
+
     type: Literal["CommandLine"] = "CommandLine"
 
     def execute_command(self, command: str):
@@ -38,5 +46,5 @@ def turn_on(self):
     def turn_off(self):
         self.execute_command(self.off)
 
-    def query_state(self) -> str:
+    def run_query(self) -> str:
         return self.execute_command(self.query)

src/maaspower/devices/smart_thing.py

@@ -9,21 +9,26 @@
 """
 import asyncio
 from dataclasses import dataclass
+from typing import Optional
 
 import aiohttp
 from pysmartthings import SmartThings
-from typing_extensions import Annotated as A
-from typing_extensions import Literal
+from typing_extensions import Literal, Annotated as A
 
-from maaspower.maasconfig import SwitchDevice
+from maaspower.maasconfig import RegexSwitchDevice
+from maaspower.maas_globals import desc
 
-from ..maas_globals import desc
 
-
-@dataclass
-class SmartThing(SwitchDevice):
+@dataclass(kw_only=True)
+class SmartThing(RegexSwitchDevice):
     """A device controlled via SmartThings"""
 
+    on: A[str, desc("command line string to switch device on")]
+    off: A[str, desc("command line string to switch device off")]
+    query: A[str, desc("command line string to query device state")]
+    query_on_regex: A[str, desc("match the on status return from query")] = "on"
+    query_off_regex: A[str, desc("match the off status return from query")] = "off"
+
     type: Literal["SmartThingDevice"] = "SmartThingDevice"
 
     api_token: A[
@@ -42,7 +47,7 @@ def turn_on(self):
     def turn_off(self):
         asyncio.run(self.switch(self.off))
 
-    def query_state(self) -> str:
+    def run_query(self) -> str:
         return asyncio.run(self.switch(self.query, True))
 
     async def switch(self, cmd: str, query: bool = False):

src/maaspower/devices/web_device.py

@@ -4,18 +4,24 @@
 from dataclasses import dataclass
 from typing import Optional, cast
 
-from typing_extensions import Literal
+from typing_extensions import Literal, Annotated as A
 
 from maaspower.maasconfig import MaasConfig, SwitchDevice
+from maaspower.maas_globals import desc
 
 from ..webhook import app
 from .web_ui import WebGui
 
 
-@dataclass
+@dataclass(kw_only=True)
 class WebDevice(SwitchDevice):
     """Commands for a  device controlled via a Web GUI"""
 
+    on: A[str, desc("command line string to switch device on")]
+    off: A[str, desc("command line string to switch device off")]
+    query: A[str, desc("command line string to query device state")]
+    query_on_regex: A[str, desc("match the on status return from query")] = "on"
+    query_off_regex: A[str, desc("match the off status return from query")] = "off"
     type: Literal["WebDevice"] = "WebDevice"
 
     # this gets called after the dataclass __init__

src/maaspower/devices/web_ui.py

@@ -14,9 +14,9 @@
 from typing_extensions import Annotated as A
 from typing_extensions import Literal
 
-from maaspower.maasconfig import SwitchDevice
+from maaspower.maasconfig import RegexSwitchDevice
+from maaspower.maas_globals import desc
 
-from ..maas_globals import desc
 
 command_regex = re.compile(r"([^\/]*)\/([^\/]*)\/?([^\/]*)?\/?([^\/]*)?$")
 index_regex = re.compile(r"(.*)\[([0-9]*)\]")
@@ -30,8 +30,8 @@ class FindBy(Enum):
     css = By.CSS_SELECTOR
 
 
-@dataclass
-class WebGui(SwitchDevice):
+@dataclass(kw_only=True)
+class WebGui(RegexSwitchDevice):
     """A device controlled via a Web GUI"""
 
     type: Literal["WebGui"] = "WebGui"

src/maaspower/maasconfig.py

@@ -13,34 +13,44 @@
 """
 
 import re
+from abc import ABC, abstractmethod
 from copy import deepcopy
-from dataclasses import dataclass
+from dataclasses import dataclass, fields
 from typing import Any, ClassVar, Dict, Mapping, Optional, Sequence, Type
 
 from apischema import deserialize, identity
 from apischema.conversions import Conversion, deserializer
-from typing_extensions import Annotated as A
+from typing_extensions import Annotated as A, override
 
 from .maas_globals import MaasResponse, T, desc
 
 
-@dataclass
-class SwitchDevice:
+@dataclass(kw_only=True)
+class SwitchDevice(ABC):
     """
     A base class for the switching devices that the webhook server will control.
+
+    Concrete subclasses MUST provide a `type` field akin to this:
+
+    type: Literal["ConreteDevice"] = "ConcreteDevice"
+
     Concrete subclasses are found in the devices subfolder
     """
 
     name: A[str, desc("A name for the switching device")]
 
-    on: A[str, desc("command line string to switch device on")]
-    off: A[str, desc("command line string to switch device off")]
-    query: A[str, desc("command line string to query device state")]
-    query_on_regex: A[str, desc("match the on status return from query")] = "on"
-    query_off_regex: A[str, desc("match the off status return from query")] = "off"
+    # command functions to be implemented in the derived classes
+    @abstractmethod
+    def turn_on(self) -> None:
+        ...
+
+    @abstractmethod
+    def turn_off(self) -> None:
+        ...
 
-    description: A[Optional[str], desc("A description of the device")] = ""
-    type: str = "none"  # a literal to distinguish the subclasses of Device
+    @abstractmethod
+    def query_state(self) -> str:
+        ...
 
     def __post_init__(self):
         # allow regular expressions for names but if the name is an
@@ -55,15 +65,27 @@ def __init_subclass__(cls):
         # Deserializers stack directly as a Union
         deserializer(Conversion(identity, source=cls, target=SwitchDevice))
 
-    # command functions to be implemented in the derived classes
-    def turn_on(self) -> None:
-        raise (NotImplementedError)
+    def copy(self, new_name: str, match) -> "SwitchDevice":
+        """
+        Create a copy of this device with a new name.
+        All the fields of the object are reformatted with substitutions in
+        regex matches using {name} for the whole match and {m1} {m2} etc
+        for matching subgroups.
 
-    def turn_off(self) -> None:
-        raise (NotImplementedError)
+        This is used for creating a specific instance of a device from
+        a regex defined device.
+        """
+        result = deepcopy(self)
 
-    def query_state(self) -> str:
-        raise (NotImplementedError)
+        # TODO can't find an easy way to iterate over dataclass field instances
+        for field in fields(self):
+            if field.name == "name":
+                continue
+            setattr(result, field.name, match.expand(getattr(result, field.name)))
+
+        result.name = new_name
+
+        return result
 
     def do_command(self, command) -> Optional[str]:
         result = None
@@ -73,41 +95,49 @@ def do_command(self, command) -> Optional[str]:
         elif command == "off":
             self.turn_off()
         elif command == "query":
-            query_response = self.query_state()
-            if re.search(self.query_on_regex, query_response, flags=re.MULTILINE):
-                result = MaasResponse.on.value
-            elif re.search(self.query_off_regex, query_response, flags=re.MULTILINE):
-                result = MaasResponse.off.value
-            else:
-                raise ValueError(
-                    f"Unknown power state response: \n{query_response}\n"
-                    f"\nfor regexes {self.query_on_regex}, {self.query_off_regex}"
-                )
+            return self.query_state()
         else:
             raise ValueError("Illegal Command")
         return result
 
-    def copy(self, new_name: str, match) -> "SwitchDevice":
-        """
-        Create a copy of this device with a new name.
-        All the fields of the object are reformatted with substitutions in
-        regex matches using {name} for the whole match and {m1} {m2} etc
-        for matching subgroups.
 
-        This is used for creating a specific instance of a device from
-        a regex defined device.
+@dataclass(kw_only=True)
+class RegexSwitchDevice(SwitchDevice, ABC):
+    """
+    An abstract `SwitchDevice` which has the ability to interpret reponses
+    and convert them to the requisit MaasReponse values using regex.
+
+    """
+    query_on_regex: A[str, desc("match the on status return from query")] = "on"
+    query_off_regex: A[str, desc("match the off status return from query")] = "off"
+
+    @abstractmethod
+    def run_query(self) -> str:
         """
-        result = deepcopy(self)
-        result.name = new_name
+        Ths method should be overridden by concrete classes. This method
+        is called by query_state and it's response is run through query_regex_on
+        and query_regex_off.
 
-        # TODO can't find an easy way to iterate over dataclass field instances
-        result.on = match.expand(result.on)
-        result.off = match.expand(result.off)
-        result.query = match.expand(result.query)
-        result.query_on_regex = match.expand(result.query_on_regex)
-        result.query_off_regex = match.expand(result.query_off_regex)
+        returns: A value to be parsed by query_regex_on and query_regex_off.
+        """
+        ...
 
-        return result
+    @override
+    def query_state(self) -> str:
+        """
+        Uses the regex patterns defined in query_on_regex and query_off_regex
+        to ascertain the correct response.
+        """
+        query_response = self.run_query()
+        if re.search(self.query_on_regex, query_response, flags=re.MULTILINE):
+            return MaasResponse.on.value
+        elif re.search(self.query_off_regex, query_response, flags=re.MULTILINE):
+            return MaasResponse.off.value
+        else:
+            raise ValueError(
+                f"Unknown power state response: \n{query_response}\n"
+                f"\nfor regexes {self.query_on_regex}, {self.query_off_regex}"
+            )
 
 
 @dataclass
@@ -150,7 +180,6 @@ def find_device(self, name: str):
 
         TODO https://github.com/gilesknap/maaspower/issues/10#issue-1222292918
         """
-
         if name in self._devices:
             return self._devices[name]
         for device in self.devices: