Merge branch 'fix-snapping' into 'develop'

more snapping tests

See merge request las-nq/adwin-control!85

src/adbasic/.gitignore

 *.TC1
+*.ERR

src/adbasic/filter_module.inc

@@ -137,18 +137,20 @@ Function FilterModule(input_int, control, filter_index, aux) as Float
   EndIf
 
   ' Snap state that waits for aux signal
-  If ((control And FCR_SW_SNAP) = 1) Then
+  If ((control And FCR_SW_SNAP) > 0) Then
     filter_output = 0.0
-    snap_value = aux - (snap_config[filter_index] And snap_value_mask)
+    snap_value = aux - (snap_config[filter_index+1] And snap_value_mask)
 
-    If ((snap_config[filter_index] And snap_lg) = 1) Then
+    If ((snap_config[filter_index+1] And snap_lg) > 0) Then
       snap_value = -1 * snap_value
     EndIf
-    If (snap_value >= 0) Then
+    If (snap_value <= 0) Then
       ' Disable snapping
       control = (control And Not(FCR_SW_SNAP))
       ' Enable output
       control = (control Or FCR_SW_OUTPUT)
+      ' Stop ramp
+      rcr = rcr And Not(0fh)
     EndIf
   EndIf
 

src/adwin_control/mockAdwin.py

@@ -105,14 +105,15 @@ class MockADwin():
         c = self.Get_Par(10 + channel)
         # read control bits
         auxSw = general.readBit(c, 9)
+        snapSw = general.readBit(c, 3)
         offsetSw = general.readBit(c, 2)
         outputSw = general.readBit(c, 1)
         inputSw = general.readBit(c, 0)
-        return inputSw, offsetSw, auxSw, outputSw
+        return inputSw, offsetSw, auxSw, outputSw, snapSw
 
     def _constructOutput(self, amount, input, aux):
         channel = self.Get_Par(4)
-        inputSw, offsetSw, auxSw, outputSw = self._readSwitches(channel)
+        inputSw, offsetSw, auxSw, outputSw, snapSw = self._readSwitches(channel)
         output = np.full(amount, 0x8000).astype(int)
         if not outputSw:
             return output
@@ -123,6 +124,8 @@ class MockADwin():
             output = output + offset
         gain = self.GetData_Double(2, channel, 1)
         output = (output - 0x8000) * gain + 0x8000
+        if snapSw:
+            output = np.full(amount, 0x8000).astype(int)
         if auxSw:
             output = output + (aux - 0x8000)
 
@@ -178,17 +181,18 @@ class MockADwin():
         aux = 0x8000
         for servo in range(1, 9):
             control = self._par[9+servo]
-            snap_enabled = general.readBit(control, 3)
+            snap_enabled = control & 8
 
             if snap_enabled:
                 snap_config = self.GetData_Long(7, servo, 1)[0]
-                snap_value = aux - (snap_config & 0x10000)
-                if control & 16 == 1:
+                snap_value = aux - (snap_config & 0xffff)
+                if (snap_config & 0x10000) > 0:
                     snap_value *= -1
-                if snap_value >= 0:
-                    control = general.clearBit(control, 3)
-                    control = general.setBit(control, 1)
+                if snap_value <= 0:
+                    control = general.clearBit(control, 3)  # disable snapSw
+                    control = general.setBit(control, 1)  # enable outputSw
                     self._par[9+servo] = control
+                    self._par[2] &= ~(15)  # stop the ramp
 
             self.Set_Par
 

src/adwin_control/servoDevice.py

@@ -44,7 +44,8 @@ class ServoDevice:
 
     def __init__(self, deviceNumber=None,
                  readFromFile=None,
-                 process=DEFAULT_PROCESS):
+                 process=DEFAULT_PROCESS,
+                 reboot=False):
         """Create a new ServoDevice object."""
         raiseExceptions = 1
         if deviceNumber == 0:
@@ -59,7 +60,7 @@ class ServoDevice:
         self._monitors = [None] * self.NUMBER_OF_MONITORS
 
         try:
-            self._bootAdwin(process)
+            self._bootAdwin(process, reboot=reboot)
         except ADwinError as e:
             if e.errorNumber == 2001:
                 log.warning('No device connected! Starting with mock device!')

src/tests/test_servo.py

@@ -10,6 +10,7 @@ from math import pow
 from pandas import DataFrame
 import os
 import logging as log
+import adwin_control
 from adwin_control.errors import *
 from tempfile import TemporaryDirectory
 from nqlab.analysis import ServoDesign
@@ -19,12 +20,8 @@ log.basicConfig(format='%(levelname)s: %(module)s.%(funcName)s: %(message)s', le
 
 class TestConfiguration(unittest.TestCase):
     def test_run_develop_version(self):
-        import adwin_control
         if 'site-packages' in adwin_control.__path__[0]:
-            self.assertTrue(False)
-            log.critical('Not running the development code!!!')
-            import sys
-            sys.exit()
+            raise Exception('Not running the development code!!!')
 
     def test_user_config(self):
         RUNNING_IN_DOCKER = os.environ.get('RUNNING_IN_DOCKER', False)
@@ -95,9 +92,13 @@ class TestVoltConvertion(unittest.TestCase):
 
 class TestServo(unittest.TestCase):
     def setUp(self):
-        self.sd = ServoDevice(settings.DEVICES_LIST[0])
-        self.sd.reboot()
+        self.sd = ServoDevice(settings.DEVICES_LIST[0], reboot=True)
+        # self.sd.reboot()
         self.s = self.sd.servo(2)
+        import adwin_control
+        log.warning('adwin_control path: {}'.format(adwin_control.__path__[0]))
+        if 'site-packages' in adwin_control.__path__[0]:
+            raise Exception('Not running the development code!!!')
 
     def test_checkNumberAndChannel(self):
         self.assertEqual(self.s._channel, 2)
@@ -458,6 +459,7 @@ class TestServo(unittest.TestCase):
             p.join()
 
     def test_check_input_offset(self):
+        self.s = self.sd.servo(8)
         self.s.fifoStepsize = 1
         self.s._waitForBufferFilling()
         data = self.s._readoutNewData(self.s._fifo['maxlen'])
@@ -519,15 +521,45 @@ class TestServo(unittest.TestCase):
         self.s.loadSettings('servo.json')
         self.assertEqual(servoDesign_str, self.s.servoDesign.__str__())
 
+    def test_snapping_disables_output(self):
+        def out():
+            return self.s._readoutNewData(10000)['output'].iloc[-1]
+
+        self.s.enableFifo()
+        self.s.offset = 9
+        self.s.offsetSw = True
+        self.s.inputSw = True
+        self.s.outputSw = True
+        sleep(.01)
+        self.assertGreater(out(), 2)
+        self.s.snapValue(5, True)
+        self.s.snapSw = True
+        sleep(.01)
+        self.assertEqual(out(), 0)
+
     def test_enable_snapping(self):
         self.assertEqual(self.s._adw.Get_Par(12), 0)
-        self.s.snapValue(5, True)
+        self.s.snapValue(-5, False)
         self.s.snapSw = True
         self.assertEqual(self.s._adw.Get_Par(12), 8)
         # Check if it snaps
         self.s.snapValue(5, False)
         sleep(1e-3)
         self.assertEqual(self.s._adw.Get_Par(12), 2)
+        # Snap when signal is higher
+        self.s.snapValue(-5, True)
+        self.s.snapSw = True
+        sleep(1e-3)
+        self.assertEqual(self.s._adw.Get_Par(12), 2)
+
+    def test_stop_ramp_when_snapping(self):
+        self.s.enableRamp(50, 10, False)
+        self.s.snapValue(5, False)
+        self.s.snapSw = True
+        # Check if it snaps
+        sleep(1e-3)
+        # Ramp stopped?
+        self.assertFalse(self.s.rampEnabled)
 
     def test_apply_old_settings(self):
         self.s.outputSw = True