From 4c73c4670514440b4bda3bcdf376717a89343140 Mon Sep 17 00:00:00 2001 From: Frederik Gladhorn Date: Wed, 29 Apr 2020 18:05:01 +0200 Subject: [PATCH] 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 | 84 +++++++++++++++++++++++++++++++++++++- tests/test_nad_protocol.py | 80 +++++++++++++++++++++++++++++++++++- 2 files changed, 162 insertions(+), 2 deletions(-) diff --git a/nad_receiver/__init__.py b/nad_receiver/__init__.py index 2e96a10..9448cf3 100644 --- a/nad_receiver/__init__.py +++ b/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 @@ -147,6 +149,86 @@ 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 attr.startswith("_"): + return + if not self._command: + if attr in CMDS.get(self._domain): # type: ignore + return _CallHandler(self._transport, self._domain, attr) + raise TypeError(f"{attr}") + if self._op: + raise TypeError(f"{self} has no attribute {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} 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 TypeError(f"{self} has no attribute {name}") + return _CallHandler(self.transport, name) class NADReceiverTelnet(NADReceiver): """ diff --git a/tests/test_nad_protocol.py b/tests/test_nad_protocol.py index 2bb421d..9e87f45 100644 --- a/tests/test_nad_protocol.py +++ b/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,81 @@ 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