mock connection

nanpy/mock/mockconnection.py

@@ -0,0 +1,77 @@
+import logging
+
+from nanpy.mock.mockeeprom import MockEEPROM
+from nanpy.mock.mockinfo import MockInfo
+from nanpy.mock.mockram import MockRAM
+from nanpy.mock.mockservo import MockServo
+from nanpy.mock.mockstepper import MockStepper
+
+
+log = logging.getLogger(__name__)
+
+
+ALL_CLASSES = [
+               MockInfo, 
+               MockRAM, 
+               MockEEPROM,
+               MockServo,
+               MockStepper,
+               ]
+
+
+class MockConnection(object):
+    state = 0
+    obj_id = 0
+    param_count = 0
+
+    def __init__(self):
+        self.objects = dict([(c.firmware_id, c()) for c in ALL_CLASSES])
+        self.objects['Info'].class_names = [x.firmware_id for x in ALL_CLASSES]
+
+    def elaborate(self,):
+        for n, x in self.objects.items():
+            if n == self.class_name:
+                self.return_value = x.elaborate(self.obj_id, self.func_name, self.params)
+                return
+        assert 0
+
+    def write(self, value):
+        log.debug('sending:%s', repr(value))
+        assert value.endswith('\x00')
+
+        value = value[0:-1]
+
+        assert value == value.strip()
+
+        if self.state == 0:
+            self.class_name = value
+            self.state += 1
+        elif self.state == 1:
+            self.obj_id = int(value)
+            self.state += 1
+        elif self.state == 2:
+            self.param_count = int(value)
+            self.state += 1
+        elif self.state == 3:
+            self.func_name = value
+            self.params = []
+            self.state += 1
+            if self.param_count == 0:
+                self.elaborate()
+                self.state = 0
+        elif self.state == 4:
+            self.params.append(value)
+            self.param_count -= 1
+            if self.param_count == 0:
+                self.elaborate()
+                self.state = 0
+
+        assert self.state<=4
+        assert self.param_count<=10
+        assert self.param_count>=0
+
+    def readline(self):
+        s = '%s\r\n' % self.return_value
+        log.debug('received:%s', repr(s))
+        return s
+

nanpy/mock/mockeeprom.py

@@ -0,0 +1,17 @@
+import logging
+
+log = logging.getLogger(__name__)
+
+class MockEEPROM(object):
+    firmware_id = "EEPROM"
+
+    def __init__(self):
+        self.memory = [0, 1, 2, 3, 4, 5, 255, 255]
+
+    def elaborate(self, obj_id, name, params):
+        if name == "size":
+            return len(self.memory)
+        elif name == "read":
+            return self.memory[int(params[0])]
+        elif name == "write":
+            self.memory[int(params[0])] = int(params[1])

nanpy/mock/mockinfo.py

@@ -0,0 +1,18 @@
+import logging
+
+log = logging.getLogger(__name__)
+
+
+class MockInfo(object):
+    firmware_id = "Info"
+
+    def __init__(self, class_names=None):
+        self.class_names = class_names
+
+
+    def elaborate(self, obj_id, name, params):
+        if name == "count":
+            return len(self.class_names)
+        elif name == "name":
+            return self.class_names[int(params[0])]
+

nanpy/mock/mockobject.py

@@ -0,0 +1,12 @@
+class MockObject(object):
+    def __init__(self):
+        self.objects = []
+
+    def elaborate(self, obj_id, name, params):
+        if name == "new":
+            x = self.factory(params)
+            self.objects.append(x)
+            return len(self.objects) - 1
+        if name == "remove":
+            del self.objects[int(params[0])]
+            return 0

nanpy/mock/mockram.py

@@ -0,0 +1,20 @@
+import logging
+
+log = logging.getLogger(__name__)
+
+class MockRAM(object):
+    firmware_id = "RAM"
+
+    def __init__(self):
+        self.memory = [0, 1, 2, 3, 4, 5, 255, 255]
+        self.free = 3
+
+    def elaborate(self, obj_id, name, params):
+        if name == "size":
+            return len(self.memory)
+        elif name == "read":
+            return self.memory[int(params[0])]
+        elif name == "write":
+            self.memory[int(params[0])] = int(params[1])
+        elif name == "free":
+            return self.free

nanpy/mock/mockservo.py

@@ -0,0 +1,33 @@
+import logging
+
+from nanpy.mock.mockobject import MockObject
+
+
+log = logging.getLogger(__name__)
+
+class MockServo(MockObject):
+    firmware_id = "Servo"
+
+    def factory(self, params):
+        d = dict(
+             pin=int(params[0]),
+             angle=0,
+             )
+        return d
+
+    def elaborate(self, obj_id, name, params):
+        if name == "read":
+            return self.objects[obj_id]['angle']
+        elif name == "write":
+            self.objects[obj_id]['angle'] = int(params[0])
+            return 0
+        elif name == "writeMicroseconds":
+            raise NotImplementedError
+        elif name == "readMicroseconds":
+            raise NotImplementedError
+        elif name == "detach":
+            self.objects[obj_id]['pin'] = None
+            return 0
+        elif name == "attached":
+            return self.objects[obj_id]['pin'] is not None
+        return MockObject.elaborate(self, obj_id, name, params)

nanpy/mock/mockstepper.py

@@ -0,0 +1,29 @@
+import logging
+
+from nanpy.mock.mockobject import MockObject
+
+
+log = logging.getLogger(__name__)
+
+
+class MockStepper(MockObject):
+    firmware_id = "Stepper"
+
+    def factory(self, params):
+        d = dict(
+             revsteps=int(params[0]),
+             pin1=int(params[1]),
+             pin2=int(params[2]),
+             speed=0,
+             )
+        return d
+
+    def elaborate(self, obj_id, name, params):
+        if name == "setSpeed":
+            self.objects[obj_id]['speed'] = int(params[0])
+            return 0
+        elif name == "step":
+            assert self.objects[obj_id]
+            return 0
+        return MockObject.elaborate(self, obj_id, name, params)
+

tests/test_offline/test_mockconnection/test_eeprom.py

@@ -0,0 +1,16 @@
+from nose.tools import eq_, ok_
+
+from nanpy.mock.mockconnection import MockConnection
+from nanpy.eepromobj import EepromLib
+
+
+def test():
+    mock = MockConnection()
+    x = EepromLib(connection=mock)
+
+    mock.objects['EEPROM'].memory = [1, 2, 3]
+    eq_(x.size, 3)
+    eq_(x.read(0), 1)
+    eq_(x.read(1), 2)
+    eq_(x.read(2), 3)
+

tests/test_offline/test_mockconnection/test_ram.py

@@ -0,0 +1,18 @@
+from nanpy.ram import RAM
+from nose.tools import eq_, ok_
+
+from nanpy.mock.mockconnection import MockConnection
+
+
+def test():
+    mock = MockConnection()
+    x = RAM(connection=mock)
+
+    mock.objects['RAM'].memory = [1, 2, 3]
+    mock.objects['RAM'].free = 2
+    eq_(x.size, 3)
+    eq_(x.free(), 2)
+    eq_(x.read(0), 1)
+    eq_(x.read(1), 2)
+    eq_(x.read(2), 3)
+

tests/test_offline/test_mockconnection/test_servo.py

@@ -0,0 +1,32 @@
+from nanpy.servo import Servo
+from nose.tools import eq_, ok_
+
+from nanpy.mock.mockconnection import MockConnection
+
+
+def test():
+    mock = MockConnection()
+    x = Servo(pin=3, connection=mock)
+
+    ok_(x.attached())
+    x.write(42)
+    eq_(x.read(), 42)
+
+
+def test_more():
+    mock = MockConnection()
+
+    def objs():
+        return mock.objects['Servo'].objects
+
+    eq_(len(objs()), 0)
+
+    x1 = Servo(pin=5, connection=mock)
+
+    eq_(len(objs()), 1)
+
+    x2 = Servo(pin=6, connection=mock)
+
+    eq_(len(objs()), 2)
+
+

tests/test_offline/test_mockconnection/test_stepper.py

@@ -0,0 +1,12 @@
+from nanpy.stepper import Stepper
+from nose.tools import eq_, ok_
+
+from nanpy.mock.mockconnection import MockConnection
+
+
+def test():
+    mock = MockConnection()
+    x = Stepper(revsteps=100, pin1=3, pin2=4, connection=mock)
+    x.step(3)
+    x.setSpeed(3)
+