Add ScanPointTickerController

malcolm/controllers/scanpointtickercontroller.py

@@ -0,0 +1,62 @@
+import numpy
+import time
+
+from malcolm.core.attribute import Attribute
+from malcolm.core.controller import Controller
+from malcolm.core.mapmeta import REQUIRED
+from malcolm.core.method import takes, Method
+from malcolm.core.pointgeneratormeta import PointGeneratorMeta
+from malcolm.core.stringmeta import StringMeta
+from malcolm.core.numbermeta import NumberMeta
+from malcolm.core.runnabledevicestatemachine import RunnableDeviceStateMachine
+
+
+class ScanPointTickerController(Controller):
+
+    def __init__(self, block):
+        super(ScanPointTickerController, self).__init__(block)
+        self.stateMachine = RunnableDeviceStateMachine("ScanPointTicker")
+
+    def create_attributes(self):
+        self.value = Attribute(NumberMeta("value", "value", numpy.float64))
+        yield self.value
+        self.generator = Attribute(
+            PointGeneratorMeta("generator", "Scan Point Generator"))
+        yield self.generator
+        self.axis_name = Attribute(StringMeta("axis_name", "Name of the axis"))
+        yield self.axis_name
+        self.exposure = Attribute(
+            NumberMeta("exposure", "Exposure time", numpy.float64))
+        yield self.exposure
+
+    @takes(PointGeneratorMeta("generator", "Generator instance"), REQUIRED,
+           StringMeta("axis_name", "Specifier for axis"), REQUIRED,
+           NumberMeta("exposure", "Detector exposure time", numpy.float64), REQUIRED)
+    def configure(self, generator, axis_name, exposure):
+        """
+        Configure the controller
+
+        Args:
+            generator(PointGenerator): Generator to create points
+            axis_name(String): Specifier for axis
+            exposure(Double): Exposure time for detector
+        """
+
+        self.generator.set_value(generator)
+        self.axis_name.set_value(axis_name)
+        self.exposure.set_value(exposure)
+        self.block.notify_subscribers()
+
+    @Method.wrap_method
+    def run(self):
+        """
+        Start the ticker process
+
+        Yields:
+            Point: Scan points from PointGenerator
+        """
+
+        for point in self.generator.value.iterator():
+            self.value.set_value(point)
+            self.block.notify_subscribers()
+            time.sleep(self.exposure.value)

tests/test_controllers/test_scanpointtickercontroller.py

@@ -0,0 +1,75 @@
+import os
+import sys
+import unittest
+sys.path.append(os.path.join(os.path.dirname(__file__), ".."))
+
+import setup_malcolm_paths
+from mock import MagicMock, patch, call
+
+from malcolm.core.attribute import Attribute
+from malcolm.core.runnabledevicestatemachine import RunnableDeviceStateMachine
+from malcolm.controllers.scanpointtickercontroller import \
+    ScanPointTickerController
+
+
+class TestScanPointTickerController(unittest.TestCase):
+
+    @patch("malcolm.core.stringmeta.StringMeta.to_dict")
+    @patch("malcolm.core.numbermeta.NumberMeta.to_dict")
+    @patch("malcolm.core.pointgeneratormeta.PointGeneratorMeta.to_dict")
+    def test_init(self, pgmd_mock, nmd_mock, smd_mock):
+        block = MagicMock()
+        sptc = ScanPointTickerController(block)
+        self.assertEqual(block, sptc.block)
+        self.assertEqual(RunnableDeviceStateMachine, type(sptc.stateMachine))
+        self.assertEqual("ScanPointTicker", sptc.stateMachine.name)
+        self.assertEquals(
+            {"value":None, "meta":nmd_mock.return_value},
+            sptc.value.to_dict())
+        self.assertEquals(
+            {"value":None, "meta":pgmd_mock.return_value},
+            sptc.generator.to_dict())
+        self.assertEquals(
+            {"value":None, "meta":smd_mock.return_value},
+            sptc.axis_name.to_dict())
+        self.assertEquals(
+            {"value":None, "meta":nmd_mock.return_value},
+            sptc.exposure.to_dict())
+
+    def test_configure(self):
+        g = MagicMock()
+        an = MagicMock()
+        e = MagicMock()
+        block = MagicMock()
+        sptc = ScanPointTickerController(block)
+
+        sptc.configure(g, an, e)
+
+        self.assertEqual(g, sptc.generator.value)
+        self.assertEqual(an, sptc.axis_name.value)
+        self.assertEqual(e, sptc.exposure.value)
+        block.notify_subscribers.assert_called_once_with()
+
+    @patch("time.sleep")
+    def test_run(self, sleep_mock):
+        points = [MagicMock(), MagicMock(), MagicMock()]
+        g = MagicMock()
+        g.iterator = MagicMock(return_value=points)
+        an = MagicMock()
+        e = MagicMock()
+        e.__float__ = MagicMock(return_value=0.1)
+        block = MagicMock()
+        sptc = ScanPointTickerController(block)
+        sptc.value.set_value = MagicMock(side_effect=sptc.value.set_value)
+
+        sptc.configure(g, an, e)
+        block.reset_mock()
+        sptc.run()
+
+        self.assertEquals([call(p) for p in points],
+                          sptc.value.set_value.call_args_list)
+        self.assertEquals([call(e)] * len(points), sleep_mock.call_args_list)
+        self.assertEqual([call()] * 3, block.notify_subscribers.call_args_list)
+
+if __name__ == "__main__":
+    unittest.main(verbosity=2)