diff --git a/nad_receiver/__init__.py b/nad_receiver/__init__.py index 363b75b..beb5fa8 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 @@ -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): """ diff --git a/tests/test_nad_protocol.py b/tests/test_nad_protocol.py index 2bb421d..fb47bdf 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,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