Dynamic api

.github/workflows/tests.yml

@@ -0,0 +1,35 @@
+name: Python package
+
+on: [push, pull_request]
+
+jobs:
+  build:
+
+    runs-on: ubuntu-latest
+    strategy:
+      matrix:
+        python-version: [3.7, 3.8]
+
+    steps:
+    - uses: actions/checkout@v2
+    - name: Set up Python ${{ matrix.python-version }}
+      uses: actions/setup-python@v1
+      with:
+        python-version: ${{ matrix.python-version }}
+    - name: Install dependencies
+      run: |
+        python -m pip install --upgrade pip
+        pip install -r requirements.txt
+    - name: Lint with flake8
+      run: |
+        pip install flake8
+        # stop the build if there are Python syntax errors or undefined names
+        flake8 . --count --select=E9,F63,F7,F82 --show-source --statistics
+        # exit-zero treats all errors as warnings. The GitHub editor is 127 chars wide
+        flake8 . --count --exit-zero --max-complexity=10 --max-line-length=127 --statistics
+    - name: Test with pytest
+      run: |
+        pip install .
+        pip install pytest
+        pytest
+

nad_receiver/__init__.py

@@ -8,10 +8,20 @@
 import codecs
 import socket
 from time import sleep
+from typing import Any, Optional
+
 from nad_receiver.nad_commands import CMDS
-from nad_receiver.nad_transport import (SerialPortTransport, TelnetTransport,
+from nad_receiver.nad_transport import (NadTransport, SerialPortTransport, TelnetTransport,
                                         DEFAULT_TIMEOUT)
 
+import logging
+
+
+logging.basicConfig()
+_LOGGER = logging.getLogger("nad_receiver")
+# Uncomment this line to see all communication with the device:
+# _LOGGER.setLevel(logging.DEBUG)
+
 
 class NADReceiver:
     """NAD receiver."""
@@ -40,6 +50,7 @@ def exec_command(self, domain, function, operator, value=None):
 
         try:
             msg = self.transport.communicate(cmd)
+            _LOGGER.debug(f"sent: '{cmd}' reply: '{msg}'")
             return msg.split('=')[1]
         except IndexError:
             pass
@@ -141,6 +152,87 @@ def tuner_fm_preset(self, operator, value=None):
         """Execute Tuner.FM.Preset."""
         return self.exec_command('tuner', 'fm_preset', operator, value)
 
+    def __getattr__(self, name: str) -> Any:
+        """Dynamically allow accessing domain, command and operator based on the command dict.
+
+        This allows directly using main.power.set('On') without needing any explicit functions
+        to be added. All that is needed for maintenance is to keep the dict in nad_commands.py
+        up to date.
+        """
+        class _CallHandler:
+            _operator_map = {
+                "get": "?",
+                "set": "=",
+                "increase": "+",
+                "decrease": "-",
+            }
+
+            def __init__(
+                self,
+                transport: NadTransport,
+                domain: str,
+                command: Optional[str] = None,
+                op: Optional[str] = None,
+            ):
+                self._transport = transport
+                self._domain = domain
+                self._command = command
+                self._op = op
+
+            def __repr__(self) -> str:
+                command = f".{self._command}" if self._command else ""
+                op = f".{self._op}" if self._op else ""
+                return f"NADReceiver.{self._domain}{command}{op}"
+
+            def __getattr__(self, attr: str) -> Any:
+                if not self._command:
+                    if attr in CMDS.get(self._domain):  # type: ignore
+                        return _CallHandler(self._transport, self._domain, attr)
+                    raise AttributeError(f"{self} has no attribute '{attr}'")
+                if self._op:
+                    raise AttributeError(f"{self} has no attribute {attr}")
+                op = _CallHandler._operator_map.get(attr, None)
+                if not op:
+                    raise AttributeError(f"{self} has no function {attr}")
+                return _CallHandler(self._transport, self._domain, self._command, attr)
+
+            def __call__(self, value: Optional[str] = None) -> Optional[str]:
+                """Executes the command.
+
+                Returns a string when possible or None.
+                Throws a ValueError in case the command was not successful."""
+                if not self._op:
+                    raise TypeError(f"{self} object is not callable.")
+
+                function_data = CMDS.get(self._domain).get(self._command)  # type: ignore
+                op = _CallHandler._operator_map.get(self._op, None)
+                if not op or op not in function_data.get("supported_operators"):  # type: ignore
+                    raise TypeError(
+                        f"{self} does not support '{self._op}', try one of {_CallHandler._operator_map.keys()}"
+                    )
+
+                cmd = f"{function_data.get('cmd')}{op}{value if value else ''}"  # type: ignore
+                reply = self._transport.communicate(cmd)
+                _LOGGER.debug(f"command: {cmd} reply: {reply}")
+                if not reply:
+                    raise ValueError(f"Did not receive reply from receiver for {self}.")
+                if reply:
+                    # Try to return the new value
+                    index = reply.find("=")
+                    if index < 0:
+                        if reply == cmd:
+                            # On some models, no value, but the command is returned.
+                            # That means success, but the receiver cannot report the state.
+                            return None
+                        raise ValueError(
+                            f"Unexpected reply from receiver for {self}: {reply}."
+                        )
+                    reply = reply[index + 1 :]
+                return reply
+
+        if name not in CMDS:
+            raise AttributeError(f"{self} has no attribute {name}")
+        return _CallHandler(self.transport, name)
 
 class NADReceiverTelnet(NADReceiver):
     """

nad_receiver/nad_fake_transport.py

@@ -0,0 +1,85 @@
+from nad_receiver.nad_transport import NadTransport
+import re
+
+class Fake_NAD_C_356BE_Transport(NadTransport):
+    """A fake NAD C 356BE device.
+
+    Behaves just like the real device (although faster).
+    This is convenient for testing or when integrating this
+    library into other applications, such as Home Assistant.
+    """
+
+    def __init__(self):
+        self._toggle = {
+            "Power": False,
+            "Mute": False,
+            "Tape1": False,
+            "SpeakerA": True,
+            "SpeakerB": False,
+        }
+        self._model = "C356BEE"
+        self._version = "V1.02"
+        self._sources = "CD TUNER DISC/MDC AUX TAPE2 MP".split()
+        self._source = "CD"
+        self._command_regex = re.compile(
+            r"(?P<component>\w+)\.(?P<function>\w+)(?P<operator>[=\?\+\-])(?P<value>.*)"
+        )
+
+    def _toggle_property(self, property: str, operator: str, value: str) -> str:
+        assert value in ["", "On", "Off"]
+        val = self._toggle[property]
+        if operator in ("+", "-"):
+            val = not val
+        if operator == "=":
+            val = value == "On"
+        self._toggle[property] = val
+        return "On" if val else "Off"
+
+    def communicate(self, command: str) -> str:
+        match = self._command_regex.fullmatch(command)
+        if not match or match.group("component") != "Main":
+            return ""
+        component = match.group("component")
+        function = match.group("function")
+        operator = match.group("operator")
+        value = match.group("value")
+
+        response = lambda val: f"{component}.{function}{'=' + val if val else ''}"
+
+        if function == "Version" and operator == "?":
+            return response(self._version)
+        if function == "Model" and operator == "?":
+            return response(self._model)
+
+        if function == "Power":
+            return response(self._toggle_property(function, operator, value))
+
+        if not self._toggle["Power"]:
+            # Except for power, all other functions return "" when power is off.
+            return ""
+
+        if function in self._toggle.keys():
+            return response(self._toggle_property(function, operator, value))
+
+        if function == "Volume":
+            # this thing doesn't report volume, but increase/decrease works and we do get the original command back
+            if operator in ("+", "-"):
+                return response(None) + operator
+
+        if function == "Source":
+            index = self._sources.index(self._source)
+            assert index >= 0
+            if operator == "+":
+                index += 1
+            if operator == "-":
+                index -= 1
+            if operator == "=":
+                index = self._sources.index(value)
+            if index < 0:
+                index = len(self._sources) - 1
+            if index == len(self._sources):
+                index = 0
+            self._source = self._sources[index]
+            return response(self._source)
+
+        return ""

requirements.txt

@@ -0,0 +1 @@
+pyserial==3.4.0