Add support to x40 (#28)

* Fix error parsing data for x40 model

* Add audio listening mode for x40 model

anthemav/protocol.py

@@ -9,7 +9,7 @@
 ATTR_CORE = {"Z1POW", "IDM"}
 
 # Audio Listening mode
-ALM_NUMBER = {
+ALM_NUMBER_x20 = {
     "None": 0,
     "AnthemLogic Cinema": 1,
     "AnthemLogic Music": 2,
@@ -27,6 +27,18 @@
     "Dolby Surround": 14,
 }
 
+ALM_NUMBER_x40 = {
+    "None": 0,
+    "AnthemLogic Cinema": 1,
+    "AnthemLogic Music": 2,
+    "Dolby Surround": 3,
+    "DTS neural:X": 4,
+    "DTS Virtual:X": 5,
+    "All Channel Stereo": 6,
+    "Mono": 7,
+    "All Channel Mono": 8,
+}
+
 # Some models (eg:MRX 520) provide a limited list of listening mode
 ALM_RESTRICTED = ["00", "01", "02", "03", "04", "05", "06", "07"]
 
@@ -191,6 +203,7 @@ def __init__(
         self._force_refresh = False
         self._model_series = ""
         self._deviceinfo_received = asyncio.Event()
+        self._alm_number = {"None": 0}
 
         for key in LOOKUP:
             setattr(self, f"_{key}", "")
@@ -279,7 +292,10 @@ def data_received(self, data):
         """Called when asyncio.Protocol detects received data from network."""
         self.buffer += data.decode()
         self.log.debug("Received %d bytes from AVR: %s", len(self.buffer), self.buffer)
-        self._assemble_buffer()
+        try:
+            self._assemble_buffer()
+        except Exception as error:
+            self.log.warning("Unable to parse message. Error: %s", error)
 
     def connection_lost(self, exc):
         """Called when asyncio.Protocol loses the network connection."""
@@ -426,7 +442,7 @@ def _parse_message(self, data: str):
             recognized = True
             self._populate_inputs(int(value))
 
-        if data.startswith("ISN"):  # x20 series: ISN01Turntable
+        if data.startswith("ISN") and len(data) > 5:  # x20 series: ISN01Turntable
             recognized = True
             self._poweron_refresh_successful = True
 
@@ -440,7 +456,9 @@ def _parse_message(self, data: str):
                 self._input_names[input_number] = value
                 self.log.debug("New Value: Input %d is called %s", input_number, value)
                 newdata = True
-        elif data.startswith("IS"):  # x40 series, example "IS3INTurntable"
+        elif (
+            data.startswith("IS") and "IN" in data and len(data) > 5
+        ):  # x40 series, example "IS3INTurntable"
             recognized = True
             self._poweron_refresh_successful = True
             in_position = data.index("IN")
@@ -494,10 +512,12 @@ def set_model_command(self, model: str):
             self._model_series = "x40"
             self.query("EMAC")
             self.query("WMAC")
+            self._alm_number = ALM_NUMBER_x40
         else:
             self.log.debug("Set Command to Model x20")
             self._ignored_commands = COMMANDS_X40
             self._model_series = "x20"
+            self._alm_number = ALM_NUMBER_x20
             self.command("ECH1")
             self.query("IDN")
 
@@ -895,7 +915,7 @@ def audio_listening_mode_list(self):
         """List of available listening mode"""
         if any(m in self.model for m in ALM_RESTRICTED_MODEL):
             return [LOOKUP["Z1ALM"][s] for s in ALM_RESTRICTED]
-        return list(ALM_NUMBER.keys())
+        return list(self._alm_number.keys())
 
     @property
     def audio_listening_mode(self):
@@ -927,8 +947,8 @@ def audio_listening_mode(self, number):
     @audio_listening_mode_text.setter
     def audio_listening_mode_text(self, text):
         self.log.debug("Switching audio listening mode to %s", text)
-        if text in ALM_NUMBER:
-            self.command(f"Z1ALM{ALM_NUMBER[text]:02d}")
+        if text in self._alm_number:
+            self.command(f"Z1ALM{self._alm_number[text]:02d}")
 
     @property
     def dolby_dynamic_range(self):

tests/test_protocol.py

@@ -1,17 +1,19 @@
-from anthemav.protocol import ALM_NUMBER, LOOKUP
+from anthemav.protocol import LOOKUP, ALM_NUMBER_x20
 from anthemav import AVR
 from unittest.mock import patch
 
 
 def test_default_alm_list():
     avr = AVR()
+    avr.set_model_command("MRX 1120")
     almList = avr.audio_listening_mode_list
     assert almList is not None
     assert len(almList) == 15
 
 
 def test_restricted_alm_list():
     avr = AVR()
+    avr.set_model_command("MRX 520")
     avr._IDM = "MRX 520"
     almList = avr.audio_listening_mode_list
     assert almList is not None
@@ -20,14 +22,16 @@ def test_restricted_alm_list():
 
 def test_set_alm_text():
     avr = AVR()
+    avr.set_model_command("MRX 520")
+    avr._IDM = "MRX 520"
     with patch.object(avr, "command") as mock:
         avr.audio_listening_mode_text = "Neo Music"
         mock.assert_called_once_with("Z1ALM06")
 
 
 def test_all_alm_matchnumber():
     for alm in list(LOOKUP["Z1ALM"].values())[1:]:
-        assert alm in ALM_NUMBER
+        assert alm in ALM_NUMBER_x20
 
 
 def test_power_on_force_refresh():
@@ -51,3 +55,29 @@ def test_populate_input():
         avr._populate_inputs(2)
         mock.assert_any_call("ISN01")
         mock.assert_called_with("ISN02")
+
+
+def test_populate_input_x40():
+    avr = AVR()
+    with patch.object(avr, "query") as mock:
+        avr.set_model_command("MRX 1140")
+        avr._populate_inputs(2)
+        mock.assert_any_call("IS1IN")
+        mock.assert_called_with("IS2IN")
+
+
+def test_parse_input_x40():
+    avr = AVR()
+    with patch.object(avr, "query"):
+        avr.set_model_command("MRX 1140")
+        avr._parse_message("IS3INName")
+        assert avr._input_names.get(3, "") == "Name"
+
+
+def test_parse_message_x40():
+    avr = AVR()
+    with patch.object(avr, "query"):
+        avr.set_model_command("MRX 1140")
+        avr._Z1POW = "1"
+        avr._parse_message("IS1ARC1")
+        assert avr._IS1ARC == "1"