diff --git a/.github/workflows/tests.yml b/.github/workflows/tests.yml new file mode 100644 index 0000000..16d2e19 --- /dev/null +++ b/.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 + diff --git a/nad_receiver/__init__.py b/nad_receiver/__init__.py index da74a74..beb5fa8 100644 --- a/nad_receiver/__init__.py +++ b/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): """ diff --git a/nad_receiver/nad_fake_transport.py b/nad_receiver/nad_fake_transport.py new file mode 100644 index 0000000..e7a394a --- /dev/null +++ b/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\w+)\.(?P\w+)(?P[=\?\+\-])(?P.*)" + ) + + 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 "" diff --git a/requirements.txt b/requirements.txt new file mode 100644 index 0000000..562f16a --- /dev/null +++ b/requirements.txt @@ -0,0 +1 @@ +pyserial==3.4.0 diff --git a/tests/test_nad_protocol.py b/tests/test_nad_protocol.py new file mode 100644 index 0000000..fb47bdf --- /dev/null +++ b/tests/test_nad_protocol.py @@ -0,0 +1,207 @@ +import re +import pytest # type: ignore + +import nad_receiver +from nad_receiver.nad_fake_transport import Fake_NAD_C_356BE_Transport + +ON = "On" +OFF = "Off" + + +class Fake_NAD_C_356BE(nad_receiver.NADReceiver): + """NAD Receiver with fake transport for testing.""" + def __init__(self): + self.transport = Fake_NAD_C_356BE_Transport() + + +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() + assert receiver.main_power("?") in (ON, OFF) + + # switch off + assert receiver.main_power("=", OFF) == OFF + assert receiver.main_power("?") == OFF + assert receiver.main_power("+") == ON + assert receiver.main_power("+") == OFF + assert receiver.main_power("?") == OFF + + # C 356BE does not reply for commands other than power when off + assert receiver.main_mute("?") is None + + assert receiver.main_power("=", ON) == ON + assert receiver.main_power("?") == ON + + assert receiver.main_mute("=", OFF) == OFF + assert receiver.main_mute("?") == OFF + + # Not a feature for this amp + assert receiver.main_dimmer("?") is None + + # Stepper motor and this thing has no idea about the volume + assert receiver.main_volume("?") is None + + # No exception + assert receiver.main_volume("+") is None + assert receiver.main_volume("-") is None + + assert receiver.main_version("?") == "V1.02" + assert receiver.main_model("?") == "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 + ### FIXME: Protocol V2 represents sources as strings, we should get these: + # assert receiver.main_source("=", "AUX") == "AUX" + # assert receiver.main_source("?") == "AUX" + # assert receiver.main_source("=", "CD") == "CD" + # assert receiver.main_source("?") == "CD" + # assert receiver.main_source("+") == "TUNER" + # assert receiver.main_source("-") == "CD" + # assert receiver.main_source("+") == "TUNER" + # assert receiver.main_source("+") == "DISC/MDC" + # assert receiver.main_source("+") == "AUX" + # assert receiver.main_source("+") == "TAPE2" + # assert receiver.main_source("+") == "MP" + # assert receiver.main_source("+") == "CD" + # assert receiver.main_source("-") == "MP" + + # Tape monitor / tape 1 is independent of sources + assert receiver.main_tape_monitor("=", OFF) == OFF + assert receiver.main_tape_monitor("?") == OFF + assert receiver.main_tape_monitor("=", ON) == ON + assert receiver.main_tape_monitor("+") == OFF + + assert receiver.main_speaker_a("=", OFF) == OFF + assert receiver.main_speaker_a("?") == OFF + assert receiver.main_speaker_a("=", ON) == ON + assert receiver.main_speaker_a("?") == ON + assert receiver.main_speaker_a("+") == OFF + assert receiver.main_speaker_a("+") == ON + assert receiver.main_speaker_a("-") == OFF + assert receiver.main_speaker_a("-") == ON + + assert receiver.main_speaker_b("=", OFF) == OFF + 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