Adds unit tests; fixes things found by unit tests

Author: xxv

python/query_demo.py

@@ -13,14 +13,15 @@
 t_ser = TracerSerial(tracer, "")
 if not fake:
     ser = serial.Serial('/dev/ttyAMA0', 9600, timeout = 1)
-    ser.write(bytearray(t_ser.to_bytes(query)))
+    ser.write(t_ser.to_bytes(query))
     data = bytearray(ser.read(200))
 else:
     data = fake
 
 print "Read %d bytes" % len(data)
 print ", ".join(map(lambda a: "%0X" % (a), data))
 result = t_ser.from_bytes(data)
+print result
 
 print "PV voltage %s" % result.pv_voltage
 print "Battery voltage %s" % result.batt_voltage

python/test.py

@@ -0,0 +1,86 @@
+from tracer import Result, QueryResult, Command, QueryCommand, ManualCommand, TracerSerial, Tracer
+from unittest2 import TestCase
+import unittest2
+from mock import Mock
+
+fixture_float_bytes=bytearray(b'\xD2\x04')
+fixture_data=bytearray(b'\x41\x42\x43')
+query_result = bytearray([0xD2, 0x4, 0xD3, 0x4, 0x0, 0x0, 0xE, 0x0, 0x53, 0x4, 0xA5, 0x5, 0x1, 0x0, 0x0, 0x1F, 0x0, 0x0, 0x0, 0x0, 0x33, 0x0A, 0x0, 0x0, 0x99, 0x5B, 0x7F])
+full_query_result = bytearray([0xEB, 0x90, 0xEB, 0x90, 0xEB, 0x90, 0x0, 0xA0, 0x18]) + query_result
+class TestResult(TestCase):
+    def test_to_float(self):
+        r=Result("")
+        self.assertEqual(12.34, r.to_float(fixture_float_bytes))
+
+def assertQueryResult(self, qr):
+    self.assertEqual(12.34, qr.batt_voltage)
+    self.assertEqual(12.35, qr.pv_voltage)
+    self.assertEqual(0.14, qr.load_amps)
+    self.assertEqual(11.07, qr.batt_overdischarge_voltage)
+    self.assertEqual(14.45, qr.batt_full_voltage)
+    self.assertEqual(True, qr.load_on)
+    self.assertEqual(False, qr.load_overload)
+    self.assertEqual(False, qr.load_short)
+    self.assertEqual(False, qr.batt_overload)
+    self.assertEqual(False, qr.batt_overdischarge)
+    self.assertEqual(False, qr.batt_full)
+    self.assertEqual(False, qr.batt_charging)
+    self.assertEqual(21, qr.batt_temp)
+    self.assertEqual(0.1, qr.charge_current)
+
+class TestQueryResult(TestCase):
+    def test_decode(self):
+        qr = QueryResult(query_result)
+        assertQueryResult(self, qr)
+
+class TestTracerSerial(TestCase):
+    def setUp(self):
+        tracer = Tracer(16)
+        tracer.get_command_bytes=Mock(return_value=fixture_data)
+        tracer.add_crc = Mock(return_value=fixture_data)
+        tracer.verify_crc = Mock(return_value=True)
+        tracer.get_result = Mock(return_value=Command(0x12, fixture_data))
+        self.ts = TracerSerial(tracer, None)
+
+    def test_to_bytes(self):
+        command = Command(0x12)
+        command.decode_result = Mock()
+        result = self.ts.to_bytes(command)
+        self.assertEqual(bytearray(b'\xAA\x55\xAA\x55\xAA\x55\xEB\x90\xEB\x90\xEB\x90' + fixture_data), result)
+
+    def test_from_bytes(self):
+        result = self.ts.from_bytes(bytearray(b'\xEB\x90\xEB\x90\xEB\x90\x00\x12\x03'+fixture_data + '\x00\x00\x7A'))
+        self.assertEqual(fixture_data, result.data)
+
+    def test_receive_result(self):
+        class FakePort(object):
+            def __init__(self, data):
+                self.data = data
+            read_idx = 0
+            def read(self, count=1):
+                result = self.data[self.read_idx:self.read_idx+count]
+                self.read_idx += count
+                return result
+        self.ts.port = FakePort(full_query_result)
+        self.ts.from_bytes = Mock(return_result=QueryResult(query_result))
+
+        # make the actual call
+        result = self.ts.receive_result()
+
+        self.assertNotEqual(None, result)
+        self.ts.from_bytes.assert_called_with(full_query_result)
+
+class TestTracer(TestCase):
+    def setUp(self):
+        self.t = Tracer(16)
+
+    def test_get_command_bytes(self):
+        result = self.t.get_command_bytes(Command(0x12, fixture_data))
+        self.assertEqual(bytearray(b'\x10\x12\x03' + fixture_data), result)
+
+    def test_get_result(self):
+        result = self.t.get_result(bytearray(b'\x00\xA0\x18') + query_result)
+        self.assertEqual(QueryResult, type(result))
+
+if __name__ == '__main__':
+    unittest2.main()

python/tracer/__init__.py

@@ -11,6 +11,7 @@
 
 class Result(object):
     """A command result from the controller."""
+    props=[]
     def __init__(self, data):
         self.data = data
         self.decode(data)
@@ -23,9 +24,12 @@ def to_float(self, two_bytes):
         """Convert a list of two bytes into a floating point value."""
         # convert two bytes to a float value
         return ((two_bytes[1] << 8) | two_bytes[0]) / 100.0
+    def __str__(self):
+        return "%s{%s}" % (self.__class__.__name__, ", ".join(map(lambda a: "%s: %s" % (a, getattr(self, a)), self.props)))
 
 class QueryResult(Result):
     """The result of a query command."""
+    props=['batt_voltage', 'pv_voltage', 'load_amps', 'batt_overdischarge_voltage', 'batt_full_voltage', 'load_on', 'load_overload', 'load_short', 'batt_overload', 'batt_overdischarge', 'batt_full', 'batt_charging', 'batt_temp', 'charge_current']
     def decode(self, data):
         """Decodes the query result, storing results as fields"""
 	if len(data) < 23:
@@ -49,7 +53,7 @@ def decode(self, data):
 
 class Command(object):
     """A command sent to the controller"""
-    def __init__(self, code, data= []):
+    def __init__(self, code, data=bytearray()):
         self.code = code
         self.data = data
     def decode_result(self, data):
@@ -74,8 +78,8 @@ def __init__(self, state):
 
 class TracerSerial(object):
     """A serial interface to the Tracer"""
-    sync_header = [0xEB, 0x90] * 3
-    comm_init = [0xAA, 0x55] * 3 + sync_header
+    sync_header = bytearray([0xEB, 0x90] * 3)
+    comm_init = bytearray([0xAA, 0x55] * 3) + sync_header
 
     def __init__(self, tracer, port):
         """Create a new Tracer interface on the given serial port
@@ -87,23 +91,45 @@ def __init__(self, tracer, port):
 
     def to_bytes(self, command):
         """Converts the command into the bytes that should be sent"""
-        cmd_data = self.tracer.get_command_bytes(command) + [0x00, 0x00, 0x7F]
+        cmd_data = self.tracer.get_command_bytes(command) + bytearray(b'\x00\x00\x7F')
         crc_data = self.tracer.add_crc(cmd_data)
         to_send = self.comm_init + crc_data
 
         return to_send
 
     def from_bytes(self, data):
         """Given bytes from the serial port, returns the appropriate command result"""
-        if list(data[0:6]) != self.sync_header:
+        if data[0:6] != self.sync_header:
             raise Exception("Invalid sync header")
         if len(data) != data[8] + 12:
-            raise Exception("Invalid length")
-        print list(data[6:])
+            raise Exception("Invalid length. Expecting %d, got %d" % (data[8] + 12, len(data)))
         if not self.tracer.verify_crc(data[6:]):
-            print "invalid crc" 
+            print "invalid crc"
 	    #raise Exception("Invalid CRC")
-        return self.tracer.get_command(data[6:])
+        return self.tracer.get_result(data[6:])
+
+    def send_command(self, command):
+        to_send = self.to_bytes(command)
+        if len(to_send) != self.port.write(to_send):
+            raise IOError("Error sending command: did not send all bytes")
+
+    def receive_result(self):
+        buff = bytearray()
+        read_idx = 0
+
+        b = self.port.read(1)
+        to_read = 200
+
+        while b >= 0 and read_idx < (to_read + 12):
+            buff += b
+            if read_idx < len(self.sync_header) and b[0] != self.sync_header[read_idx]:
+                raise IOError("Error receiving result: invalid sync header")
+            # the location of the read length
+            elif read_idx == 8:
+                to_read = b[0]
+            read_idx += 1
+            b = self.port.read(1)
+        return self.from_bytes(buff)
 
 class Tracer(object):
     """An implementation of the Tracer MT-5 communication protocol"""
@@ -116,17 +142,17 @@ def __init__(self, controller_id):
     def get_command_bytes(self, command):
         """Given a command, gets its byte representation
 
-        This excludes the CRC and trailer."""
-        data = []
+        This excludes the header, CRC, and trailer."""
+        data = bytearray()
         data.append(self.controller_id)
         data.append(command.code)
         data.append(len(command.data))
         data += command.data
 
         return data
 
-    def get_command(self, data):
-        if data[1] == 0xA0:
+    def get_result(self, data):
+        if data[1] == QueryCommand().code:
             return QueryResult(data[3:])
 
     def verify_crc(self, data):
@@ -137,7 +163,6 @@ def verify_crc(self, data):
 
     def add_crc(self, data):
         """Returns a copy of the data with the CRC added"""
-        data = list(data)
         if len(data) < 6:
             raise Exception("data are too short")
         # the input CRC bytes must be zeroed