Backported parsing of IS AT command response as I/O data from ZigBee devices

Author: pmalmsten

xbee/ieee.py

@@ -131,7 +131,10 @@ class XBee(XBeeBase):
                             [{'name':'frame_id',    'len':1},
                              {'name':'command',     'len':2},
                              {'name':'status',      'len':1},
-                             {'name':'parameter',   'len':None}]},
+                             {'name':'parameter',   'len':None}],
+                         'parsing': [('parameter',
+                                       lambda xbee,original: xbee._parse_IS_at_response(original))]
+                             },
                      b"\x97":
                         {'name':'remote_at_response',
                          'structure':
@@ -140,8 +143,21 @@ class XBee(XBeeBase):
                              {'name':'source_addr',     'len':2},
                              {'name':'command',         'len':2},
                              {'name':'status',          'len':1},
-                             {'name':'parameter',       'len':None}]},
+                             {'name':'parameter',       'len':None}],
+                         'parsing': [('parameter',
+                                       lambda xbee,original: xbee._parse_IS_at_response(original))]
+                             },
                      }
+                     
+    def _parse_IS_at_response(self, packet_info):
+        """
+        If the given packet is a successful remote AT response for an IS
+        command, parse the parameter field as IO data.
+        """
+        if packet_info['id'] in ('at_response','remote_at_response') and packet_info['command'].lower() == b'is' and packet_info['status'] == b'\x00':
+               return self._parse_samples(packet_info['parameter'])
+        else:
+            return packet_info['parameter']
 
     def __init__(self, *args, **kwargs):
         # Call the super class constructor to save the serial port

xbee/tests/test_ieee.py

@@ -10,6 +10,7 @@
 import unittest, sys, traceback
 from xbee.tests.Fake import FakeDevice, FakeReadDevice
 from xbee.ieee import XBee
+from xbee.frame import APIFrame
 from xbee.python2to3 import byteToInt, intToByte, stringToBytes
 
 class InitXBee(unittest.TestCase):
@@ -561,6 +562,72 @@ def test_read_at_params(self):
                          'parameter':b'\x00\x00\x00'}
         self.assertEqual(info, expected_info)
 
+    def test_is_response_parsed_as_io(self):
+        """
+        I/O data in a AT response for an IS command is parsed.
+        """
+         ## Build IO data
+        # One sample, ADC 0 enabled
+        # DIO 1,3,5,7 enabled
+        header = b'\x01\x02\xAA'
+        
+        # First 7 bits ignored, DIO8 low, DIO 0-7 alternating
+        # ADC0 value of 255
+        sample = b'\x00\xAA\x00\xFF'
+        data = header + sample
+        
+        device = FakeReadDevice(
+            APIFrame(data = b'\x88DIS\x00' + data).output()
+        )
+        
+        xbee = XBee(device)
+        
+        info = xbee.wait_read_frame()
+        expected_info = {'id':'at_response',
+                         'frame_id':b'D',
+                         'command':b'IS',
+                         'status':b'\x00',
+                         'parameter':[{'dio-1':True,
+                                      'dio-3':True,
+                                      'dio-5':True,
+                                      'dio-7':True,
+                                      'adc-0':255}]}
+        self.assertEqual(info, expected_info)
+        
+    def test_is_remote_response_parsed_as_io(self):
+        """
+        I/O data in a Remote AT response for an IS command is parsed.
+        """
+         ## Build IO data
+        # One sample, ADC 0 enabled
+        # DIO 1,3,5,7 enabled
+        header = b'\x01\x02\xAA'
+        
+        # First 7 bits ignored, DIO8 low, DIO 0-7 alternating
+        # ADC0 value of 255
+        sample = b'\x00\xAA\x00\xFF'
+        data = header + sample
+        
+        device = FakeReadDevice(
+            APIFrame(data = b'\x97D\x00\x13\xa2\x00@oG\xe4v\x1aIS\x00' + data).output()
+        )
+        
+        xbee = XBee(device)
+        
+        info = xbee.wait_read_frame()
+        expected_info = {'id':'remote_at_response',
+                         'frame_id':b'D',
+                         'source_addr_long': b'\x00\x13\xa2\x00@oG\xe4',
+                         'source_addr': b'v\x1a',
+                         'command':b'IS',
+                         'status':b'\x00',
+                         'parameter':[{'dio-1':True,
+                                      'dio-3':True,
+                                      'dio-5':True,
+                                      'dio-7':True,
+                                      'adc-0':255}]}
+        self.assertEqual(info, expected_info)
+        
     def test_read_io_data(self):
         """
         XBee class should properly read and parse incoming IO data