beamline settings...

Author: Kay Kasemir (ky9)

doc/getting_started.rst

@@ -268,9 +268,16 @@ Unfortunately, the "correct" way to set a PV is not discernible from the outside
 It requires knowledge about the implementation of a PV in the IOC.
 And even if the correct way to set a PV is known, having to type
 all parameters for each `Set` command can be cumbersone.
-The PyScanClient library offers :ref:`scan_settings` where the ideal
-parameters for each PV can be configured once, and wrappers for the
-basic `Set` and `Wait` commands will then automatically use them as a default.
+The PyScanClient :ref:`scan_settings` provide a way to configure default parameters
+for each PV.
+Wrappers for the basic `Set`, `Loop` and `Wait` commands then use these defaults.
+For an example, see `example/beamline_setup.py`:
+
+.. literalinclude:: ../example/beamline_setup.py
+
+The code in `example/beamline1.py`, `example/beamline2.py` and
+`example/beamline3.py` shows how to take advantage of the
+scan settings.
 
 
 Table Scan

example/beamline1.py

@@ -1,14 +1,18 @@
 from beamline_setup import *
 
-# N-Dim scan
-cmds = ndim(('xpos', 1, 10), Delay(0.2))
-print cmds
-id = scan_client.submit(cmds)
-print scan_client.waitUntilDone(id)
+# N-Dim scan with motors
+# At each point, wait for pcharge which rises at about 1e9 per second
+# Then log how many neutrons we got
+cmds = ndim(('motor_x', 1, 3), ('motor_y', 1, 3),
+            TakeData('pcharge', 3*1e9),
+            'neutrons')
+
+# Note how the defaults apply a readback to motors,
+# how TakeData is expanded into basic commands etc.
+print(cmds)
 
-# Simulated pcharge rises at about 1e9 per second
-cmds = ndim(('xpos', 1, 3), ('ypos', 1, 3), TakeData('pcharge', 3*1e9), 'neutrons')
-print cmds
+# The scan_client is already configured with the correct IP address
 id = scan_client.submit(cmds)
-print scan_client.scanInfo(id)
-# scan_client.waitUntilDone(id)
+print(scan_client.scanInfo(id))
+print(scan_client.waitUntilDone(id))
+

example/beamline2.py

@@ -2,17 +2,17 @@
 
 # Table
 cmds = table_scan(
-  [ 'xpos',    'ypos', 'Wait For', 'Value'],
+  [ 'motor_x', 'motor_y', 'Wait For', 'Value'],
 [
-  [     1, [ 2, 4, 6],  'pcharge',  3*1e9 ],
-  [     3,          8,  'pcharge',  3*1e9 ],
+  [         1, [ 2, 4, 6], 'pcharge',  3*1e9 ],
+  [         3,          8, 'pcharge',  3*1e9 ],
 ])
-print cmds
+print(CommandSequence(cmds))
 id = scan_client.submit(cmds)
 scan_client.waitUntilDone(id)
 
 # Dump motor positions
 data = scan_client.getData(id)
-table = createTable(data, 'xpos', 'ypos')
+table = createTable(data, 'motor_x', 'motor_y')
 for (x, y) in zip(*table):
-    print "%d, %d" % (x, y)
+    print("%d, %d" % (x, y))

example/beamline3.py

@@ -8,12 +8,12 @@
 cmds.append(SetChopper(1.7, 45.0))
 cmds.append(TakeData('pcharge', 5e9))
 cmds.append(Set('setpoint', 1))
-print cmds
+print(cmds)
 
 result = scan_client.simulate(cmds)
-print result['simulation']
+print(result['simulation'])
 
 id = scan_client.submit(cmds)
-print "Submitted scan #%d" % id
+print("Submitted scan #%d" % id)
 
 

example/beamline_setup.py

@@ -1,31 +1,49 @@
 """Example for beamline specific setup
 
-Uses the scan server example database
-plus some local PVs to fake devices.
+   All scripts that access the scan server
+   should start with
+
+     'from beamline_setup import *'
 """
 
+# Get all scan commands
 from scan import *
 
-# Note that the basic Loop/Set/Wait commands are replaced by
-# those that utilize custom scan settings
+# Replace basic Loop/Set/Wait commands with wrappers
+# that utilize custom scan settings
 from scan.util import SettingsBasedLoop as Loop
 from scan.util import SettingsBasedSet as Set
 from scan.util import SettingsBasedWait as Wait
 
-# Custom scan settings
+# Each "beamline" or more generally setup that uses
+# the PyScanClient should create its own ScanSettings 
 class BeamlineScanSettings(ScanSettings):
     def __init__(self):
         super(BeamlineScanSettings, self).__init__()
-        # Define several PVs to use completion etc.
-        self.defineDeviceClass("shutter", readback=True)
-        self.defineDeviceClass("chopper:.*", completion=True)
-        self.defineDeviceClass(".*daq.*", completion=True)
+
+        # Define settings based on PV name patterns
+        # (regular expressions).
+        # Order matters! List most generic patterns first,
+        # specific PV names last.
+        # For example, motors should use completion, and check a readback
         self.defineDeviceClass("motor_.", completion=True, readback=True)
-        self.defineDeviceClass("setpoint", completion=True, readback="readback", tolerance=0.1)
+        # The specific motors in the simulation.db, however, don't support completion
+        self.defineDeviceClass("motor_x", completion=False, readback=True)
+        self.defineDeviceClass("motor_y", completion=False, readback=True)
+        self.defineDeviceClass("shutter", readback=True)
+        # The simulated "setpoint" uses a different PV "readback" as its readback.
+        # (readback=False skips readback check
+        #  readback=True calls getReadbackName(pv) which typically returns the PV itself,
+        #  readback='whatever' uses provided PV
+        # )
+        self.defineDeviceClass("setpoint", completion=False, readback="readback", tolerance=0.1, timeout=20)
+        # When waiting for proton charge or neutron counts, use "increase by"
         self.defineDeviceClass("pcharge", comparison="increase by")
         self.defineDeviceClass("neutrons", comparison="increase by")
 
-        # Used by UI to suggest devices on which one can 'wait'
+        # Each site may add more to its site-specific configuration.
+        # The example/opi for alignment uses these to
+        # populate drop-downs
         self.settable = [ "motor_x", "motor_y" ]
         self.waitable = [ "seconds", "pcharge" ]
         self.loggable = [ "signal" ]
@@ -38,11 +56,10 @@ def getReadbackName(self, device_name):
         if "motor" in device_name:
            return device_name + ".RBV"
 
-scan_settings = BeamlineScanSettings()
 # Install beam line specific scan settings
+scan_settings = BeamlineScanSettings()
 setScanSettings(scan_settings)
 
-
 # 'Meta Commands'
 def Start():
     """Start data acquisition"""
@@ -58,13 +75,12 @@ def TakeData(counter, limit):
     return  Sequence(Start(), Wait(counter, limit), Stop())
 
 def SetChopper(wavelength, phase):
-    return  Sequence(scan_settings.Set('loc://chopper:run(0)', 0),
-                     scan_settings.Set('loc://chopper:wlen(0)', wavelength),
-                     scan_settings.Set('loc://chopper:phs(0)', phase),
-                     scan_settings.Set('loc://chopper:run(0)', 1)
+    return  Sequence(Set('loc://chopper:run(0)', 0),
+                     Set('loc://chopper:wlen(0)', wavelength),
+                     Set('loc://chopper:phs(0)', phase),
+                     Set('loc://chopper:run(0)', 1)
                     )
 
-
 def table_scan(headers, rows):
     """Create table scan with pre/post/start/stop for this beam line"""
     table = TableScan(headers, rows,
@@ -78,37 +94,15 @@ def table_scan(headers, rows):
 # Create a scan client, using the host name that executes the scan server
 scan_client = ScanClient('localhost')
 
-
-
-
-
-import unittest
-
-class TestScanClient(unittest.TestCase):
-    def testSet(self):
-        self.assertEqual(str(Set('setpoint', 1)), "Set('setpoint', 1, completion=True, readback='readback')")
-        self.assertEqual(str(Set('setpoint', 1, completion=False)), "Set('setpoint', 1, readback='readback')")
-        self.assertEqual(str(Set('setpoint', 1, readback=False)), "Set('setpoint', 1, completion=True)")
-        self.assertEqual(str(Set('setpoint', 1, timeout=10)), "Set('setpoint', 1, completion=True, readback='readback', timeout=10)")
-
-    def testLoop(self):
-        self.assertEqual(str(Loop('x', 1, 5, 0.5)), "Loop('x', 1, 5, 0.5)")
-        self.assertEqual(str(Loop('motor_x', 1, 5, 0.5)), "Loop('motor_x', 1, 5, 0.5, completion=True, readback='motor_x')")
-        self.assertEqual(str(Loop('motor_x', 1, 5, 0.5, completion=False)), "Loop('motor_x', 1, 5, 0.5, readback='motor_x')")
-        self.assertEqual(str(Loop('motor_x', 1, 5, 0.5, completion=False, readback=False)), "Loop('motor_x', 1, 5, 0.5)")
-        self.assertEqual(str(Loop('motor_x', 1, 5, 0.5, Comment(''), completion=False, readback=False)),
-                         "Loop('motor_x', 1, 5, 0.5, [ Comment('') ])")
-        self.assertEqual(str(Loop('motor_x', 1, 5, 0.5, Comment('a'), Comment('b'), completion=False, readback=False)),
-                         "Loop('motor_x', 1, 5, 0.5, [ Comment('a'), Comment('b') ])")
-        self.assertEqual(str(Loop('motor_x', 1, 5, 0.5, [ Comment('a'), Comment('b') ], completion=False, readback=False)),
-                         "Loop('motor_x', 1, 5, 0.5, [ Comment('a'), Comment('b') ])")
-        
-    def testWait(self):
-        self.assertEqual(str(Wait('whatever', 1)), "Wait('whatever', 1, comparison='>=')")
-        self.assertEqual(str(Wait('pcharge', 1)), "Wait('pcharge', 1, comparison='increase by')")
-        self.assertEqual(str(Wait('pcharge', 1, timeout=5)), "Wait('pcharge', 1, comparison='increase by', timeout=5)")
-        self.assertEqual(str(Wait('pcharge', 1, timeout=5, comparison='>=')), "Wait('pcharge', 1, comparison='>=', timeout=5)")
-
-
+# As a convenience, `python beamline_setup.py` prints settings
 if __name__ == '__main__':
-    unittest.main()
+    from scan.util.scan_settings import getScanSettings
+
+    print("Beam Line scan settings")
+    print("=======================")
+    for setting in getScanSettings().device_settings:
+        print(setting)
+    print("")
+    print("Scan Client")
+    print("===========")
+    print(scan_client)