Dynamic API

We have a complete description of the protocol in nad_commands. Let's
use it. This dynamically generates objects from the command dictionary,
so that new functions do not need any code any more. Instead one can
simply call:
receiver.main.volume.increase()

This works by inspecting nad_commands when asking for attributes of the
receiver class or it's dynamically created subclasses.

The old API is around unchanged to stay compatible.

nad_receiver/__init__.py

@@ -8,8 +8,10 @@
 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
@@ -150,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):
     """

tests/test_nad_protocol.py

@@ -14,7 +14,7 @@ def __init__(self):
         self.transport = Fake_NAD_C_356BE_Transport()
 
 
-def test_NAD_C_356BE():
+def test_NAD_C_356BE_old_api():
     # This test can be run with the real amplifier, just instantiate
     # the real transport instead of the fake one
     receiver = Fake_NAD_C_356BE()
@@ -88,3 +88,120 @@ def test_NAD_C_356BE():
     assert receiver.main_speaker_b("?") == OFF
 
     assert receiver.main_power("=", OFF) == OFF
+
+
+def test_NAD_C_356BE_new_api():
+    # This test can be run with the real amplifier, just instantiate
+    # the real transport instead of the fake one
+    receiver = Fake_NAD_C_356BE()
+    assert receiver.main.power.get() in (ON, OFF)
+
+    # switch off
+    assert receiver.main.power.set(OFF) == OFF
+    assert receiver.main.power.get() == OFF
+    assert receiver.main.power.increase() == ON
+    assert receiver.main.power.increase() == OFF
+    assert receiver.main.power.get() == OFF
+
+    # C 356BE does not reply for commands other than power when off
+    with pytest.raises(ValueError):
+        receiver.main.mute.get()
+
+    assert receiver.main.power.set(ON) == ON
+    assert receiver.main.power.get() == ON
+
+    assert receiver.main.mute.set(OFF) == OFF
+    assert receiver.main.mute.get() == OFF
+
+    # Not a feature for this amp
+    with pytest.raises(ValueError):
+        receiver.main.dimmer.get()
+
+    # Stepper motor and this thing has no idea about the volume
+    with pytest.raises(ValueError):
+        receiver.main.volume.get()
+
+    # No exception
+    assert receiver.main.volume.increase() is None
+    assert receiver.main.volume.decrease() is None
+
+    assert receiver.main.version.get() == "V1.02"
+    assert receiver.main.model.get() == "C356BEE"
+
+    # Here the RS232 NAD manual seems to be slightly off / maybe the model is different
+    # The manual claims:
+    # CD Tuner Video Disc Ipod Tape2 Aux
+    # My Amp:
+    # CD Tuner Disc/MDC Aux Tape2 MP
+    assert receiver.main.source.set("AUX") == "AUX"
+    assert receiver.main.source.get() == "AUX"
+    assert receiver.main.source.set("CD") == "CD"
+    assert receiver.main.source.get() == "CD"
+    assert receiver.main.source.increase() == "TUNER"
+    assert receiver.main.source.decrease() == "CD"
+    assert receiver.main.source.increase() == "TUNER"
+    assert receiver.main.source.increase() == "DISC/MDC"
+    assert receiver.main.source.increase() == "AUX"
+    assert receiver.main.source.increase() == "TAPE2"
+    assert receiver.main.source.increase() == "MP"
+    assert receiver.main.source.increase() == "CD"
+    assert receiver.main.source.decrease() == "MP"
+
+    # Tape monitor / tape 1 is independent of sources
+    assert receiver.main.tape_monitor.set(OFF) == OFF
+    assert receiver.main.tape_monitor.get() == OFF
+    assert receiver.main.tape_monitor.set(ON) == ON
+    assert receiver.main.tape_monitor.increase() == OFF
+
+    assert receiver.main.speaker_a.set(OFF) == OFF
+    assert receiver.main.speaker_a.get() == OFF
+    assert receiver.main.speaker_a.set(ON) == ON
+    assert receiver.main.speaker_a.get() == ON
+    assert receiver.main.speaker_a.increase() == OFF
+    assert receiver.main.speaker_a.increase() == ON
+    assert receiver.main.speaker_a.decrease() == OFF
+    assert receiver.main.speaker_a.decrease() == ON
+
+    assert receiver.main.speaker_b.set(OFF) == OFF
+    assert receiver.main.speaker_b.get() == OFF
+
+    assert receiver.main.power.set(OFF) == OFF
+
+
+def test_dynamic_api():
+    receiver = Fake_NAD_C_356BE()
+    assert receiver.main.power.get() in (ON, OFF)
+
+    # invalid attributes result in attribute error
+    with pytest.raises(AttributeError):
+        receiver.foo
+    with pytest.raises(AttributeError):
+        receiver.foo.bar
+
+    # valid attributes work and have a good __repr__
+    assert str(receiver.main) == "NADReceiver.main"
+    assert str(receiver.main.power) == "NADReceiver.main.power"
+    assert str(receiver.main.power.get) == "NADReceiver.main.power.get"
+    assert str(receiver.main.power.increase) == "NADReceiver.main.power.increase"
+    assert str(receiver.main.power.decrease) == "NADReceiver.main.power.decrease"
+    assert str(receiver.main.power.set) == "NADReceiver.main.power.set"
+
+    # functions on dynamic objects can be called
+    assert callable(receiver.main.power.get)
+    assert callable(receiver.main.power.set)
+    assert callable(receiver.main.power.increase)
+    assert callable(receiver.main.power.decrease)
+
+    # attributes can not be called
+    with pytest.raises(TypeError, match="object is not callable"):
+        receiver.main()
+    with pytest.raises(TypeError, match="object is not callable"):
+        receiver.main.power()
+
+    # invalid properties are AttributeErrors
+    with pytest.raises(AttributeError):
+        receiver.main.power.invalid
+    with pytest.raises(AttributeError):
+        receiver.main.power.invalid()
+    with pytest.raises(AttributeError):
+        receiver.main.power.get.invalid_too