-
Notifications
You must be signed in to change notification settings - Fork 0
/
BCSz_sync.py
1862 lines (1292 loc) · 76.7 KB
/
BCSz_sync.py
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317
318
319
320
321
322
323
324
325
326
327
328
329
330
331
332
333
334
335
336
337
338
339
340
341
342
343
344
345
346
347
348
349
350
351
352
353
354
355
356
357
358
359
360
361
362
363
364
365
366
367
368
369
370
371
372
373
374
375
376
377
378
379
380
381
382
383
384
385
386
387
388
389
390
391
392
393
394
395
396
397
398
399
400
401
402
403
404
405
406
407
408
409
410
411
412
413
414
415
416
417
418
419
420
421
422
423
424
425
426
427
428
429
430
431
432
433
434
435
436
437
438
439
440
441
442
443
444
445
446
447
448
449
450
451
452
453
454
455
456
457
458
459
460
461
462
463
464
465
466
467
468
469
470
471
472
473
474
475
476
477
478
479
480
481
482
483
484
485
486
487
488
489
490
491
492
493
494
495
496
497
498
499
500
501
502
503
504
505
506
507
508
509
510
511
512
513
514
515
516
517
518
519
520
521
522
523
524
525
526
527
528
529
530
531
532
533
534
535
536
537
538
539
540
541
542
543
544
545
546
547
548
549
550
551
552
553
554
555
556
557
558
559
560
561
562
563
564
565
566
567
568
569
570
571
572
573
574
575
576
577
578
579
580
581
582
583
584
585
586
587
588
589
590
591
592
593
594
595
596
597
598
599
600
601
602
603
604
605
606
607
608
609
610
611
612
613
614
615
616
617
618
619
620
621
622
623
624
625
626
627
628
629
630
631
632
633
634
635
636
637
638
639
640
641
642
643
644
645
646
647
648
649
650
651
652
653
654
655
656
657
658
659
660
661
662
663
664
665
666
667
668
669
670
671
672
673
674
675
676
677
678
679
680
681
682
683
684
685
686
687
688
689
690
691
692
693
694
695
696
697
698
699
700
701
702
703
704
705
706
707
708
709
710
711
712
713
714
715
716
717
718
719
720
721
722
723
724
725
726
727
728
729
730
731
732
733
734
735
736
737
738
739
740
741
742
743
744
745
746
747
748
749
750
751
752
753
754
755
756
757
758
759
760
761
762
763
764
765
766
767
768
769
770
771
772
773
774
775
776
777
778
779
780
781
782
783
784
785
786
787
788
789
790
791
792
793
794
795
796
797
798
799
800
801
802
803
804
805
806
807
808
809
810
811
812
813
814
815
816
817
818
819
820
821
822
823
824
825
826
827
828
829
830
831
832
833
834
835
836
837
838
839
840
841
842
843
844
845
846
847
848
849
850
851
852
853
854
855
856
857
858
859
860
861
862
863
864
865
866
867
868
869
870
871
872
873
874
875
876
877
878
879
880
881
882
883
884
885
886
887
888
889
890
891
892
893
894
895
896
897
898
899
900
901
902
903
904
905
906
907
908
909
910
911
912
913
914
915
916
917
918
919
920
921
922
923
924
925
926
927
928
929
930
931
932
933
934
935
936
937
938
939
940
941
942
943
944
945
946
947
948
949
950
951
952
953
954
955
956
957
958
959
960
961
962
963
964
965
966
967
968
969
970
971
972
973
974
975
976
977
978
979
980
981
982
983
984
985
986
987
988
989
990
991
992
993
994
995
996
997
998
999
1000
"""
A python client interface to the BCS API, using zmq and asyncio.
All API calls ultimately call bcs_request with specialized JSON. Calling bcs_request from client
application code is possible, but discouraged, and not supported, as the call signature may change in future versions.
Contact [email protected] with questions / requests.
"""
import sys
import asyncio
if sys.platform[:3] == 'win':
# zmq.asyncio does not support the default (proactor) event loop on windows.
# so set the event loop to one zmq supports
asyncio.set_event_loop_policy(asyncio.WindowsSelectorEventLoopPolicy())
import zmq
# import zmq.asyncio
import zmq.utils.z85
import json
import time
# helper classes
from enum import Flag # for MotorStatus
class MotorStatus(Flag):
HOME = 1
FORWARD_LIMIT = 2
REVERSE_LIMIT = 4
MOTOR_DIRECTION = 8
MOTOR_OFF = 16
MOVE_COMPLETE = 32
FOLLOWING_ERROR = 64
NOT_IN_DEAD_BAND = 128
FORWARD_SW_LIMIT = 256
REVERSE_SW_LIMIT = 512
MOTOR_DISABLED = 1024
RAW_MOTOR_DIRECTION = 2048
RAW_FORWARD_LIMIT = 4096
RAW_REVERSE_LIMIT = 8192
RAW_FORWARD_SW_LIMIT = 16384
RAW_REVERSE_SW_LIMIT = 32768
RAW_MOVE_COMPLETE = 65536
MOVE_LT_THRESHOLD = 131072
def is_set(self, flag):
return bool(self._value_ & flag._value_)
# helper functions
def bytes_from_blob(blob):
"""Deserializes binary data blobs. This implementation uses zmq's Base85 encoding, but future ones may not."""
blob_len = blob['length']
blob_str = blob['blob']
return zmq.utils.z85.decode(blob_str)[:blob_len]
_zmq_context = None
class BCSServer:
"""Represents a remote BCS endstation or beamline system, running the BCS zmq server.
Each endpoint returns a dictionary of results. In addition to endpoint-specific keys, the following
global keys are included.
* **success** (*bool*) - API endpoint completion status. Note that this does *not* indicate endpoint errors.
* **error_description** (*str*) - If the server failed to execute the endpoint, this should indicate why. There may be warnings or information here even if the endpoint completed successfully.
* **log** (*bool*) - True if the server logged this request.
"""
_zmq_socket = None
@staticmethod
def _get_server_public_key(addr, port):
clear_socket = _zmq_context.socket(zmq.REQ)
clear_socket.connect(f'tcp://{addr}:{port}')
clear_socket.send('public'.encode())
server_public = clear_socket.recv()
clear_socket.close()
return server_public
# def __init__(self, addr='127.0.0.1', port=5577):
def connect(self, addr='127.0.0.1', port=5577):
"""(formerly the Constructor) Supply the zmq address string, addr, to reach this endstation."""
global _zmq_context
# the first server object will create the global zmq context
if not _zmq_context:
# _zmq_context = zmq.asyncio.Context()
_zmq_context = zmq.Context()
self._zmq_socket = _zmq_context.socket(zmq.REQ)
(client_public_key, client_secret_key) = zmq.curve_keypair()
# server_public_key = asyncio.get_running_loop().run_until_complete(self._get_server_public_key(addr, port))
server_public_key = self._get_server_public_key(addr, port)
print(f'Server Public Key {server_public_key}')
self._zmq_socket.setsockopt(zmq.CURVE_SERVERKEY, server_public_key)
self._zmq_socket.setsockopt(zmq.CURVE_PUBLICKEY, client_public_key)
self._zmq_socket.setsockopt(zmq.CURVE_SECRETKEY, client_secret_key)
self._zmq_socket.connect(f'tcp://{addr}:{port + 1}')
def bcs_request(self, command_name, param_dict, debugging=False):
"""
The method responsible for direct communication to the BCS server
:param command_name: Name of the API endpoint
:type command_name: str
:param param_dict: Parameter dictionary
:type param_dict: dict
"""
if debugging:
print(f"API command {command_name} BEGIN.")
api_call_start = time.time()
param_dict['command'] = command_name
param_dict['_unused'] = '_unused'
if 'self' in param_dict:
del param_dict['self']
self._zmq_socket.send(json.dumps(param_dict).encode())
response_dict = json.loads(self._zmq_socket.recv())
response_dict['API_delta_t'] = time.time() - api_call_start
if debugging:
print(f"API command {command_name} END {response_dict['API_delta_t']} s.")
return response_dict
# end BCSz_header.py
def acquire_data(self, chans=[], time=0, counts=0) -> dict:
"""
Acquires for ``time`` seconds, or ``counts`` counts. whichever is non-zero. **Waits for the acquision to complete** and returns data for channels specified in ``chans``.
If both ``counts`` *and* ``time`` are non-zero, which parameter takes precedence is not defined.
:param chans: AI channel names to acquire. An empty array will return data for all AI channels on the server.
:type chans: list
:param time: If non-zero, the amount of time to acquire.
:type time: float
:param counts: If non-zero, the number of counts to acquire.
:type counts: int
:return: Dictionary of results, with the following key(s).
* **chans** (*list*) - The channel names that have data in the **data** array, in the same order.
* **not_found** (*list*) - The requested channels in ``chans`` that were not found on the host system, if any.
* **data** (*list*) - The acquired data, in the same order as the channel names in **chans**.
"""
return self.bcs_request('AcquireData', dict(locals()))
def at_preset(self, name="_none") -> dict:
"""
Checks if the associated motor is at the preset position. The associated motor and preset position are defined on the server.
:param name: Name of the preset (not the motor name).
:type name: str
:return: Dictionary of results, with the following key(s).
* **at_preset** (*bool*) - True if the target motor is at the preset position.
* **position** (*float*) - The preset position, in units of the target motor.
"""
return self.bcs_request('AtPreset', dict(locals()))
def at_trajectory(self, name="success") -> dict:
"""
Checks if all requirement been satisfied to be "At trajectory" (usually just means that each motor is at its trajectory goal).
:param name: Trajectory name (defined on the server).
:type name: str
:return: Dictionary of results, with the following key(s).
* **at_trajectory** (*bool*) - Are the motors "At trajectory?"
* **running** (*bool*) - Is the trajectory still running (are the motors still moving)?
"""
return self.bcs_request('AtTrajectory', dict(locals()))
def command_motor(self, commands=[], motors=[], goals=[]) -> dict:
"""
Command one or more motors. A large number of commands are available.
:param commands: Array of motor commands, one for each motor in ``motors``. Current valid commands are {None, Normal Move, Backlash Move, Velocity Move, Move to Home, Stop Motor, Set Position, Enable Motor, Disable Motor, Move to Index, Run Home Routine, Set Velocity, Set Acceleration, Set Deceleration, Enable and Move, Disable SW Limits, Enable SW Limits, Start Time Delay, Check Time Delay, Set Output Pulses, Backlash Jog, Normal Jog, Run Coord Program, Halt Coord Program, Gearing ON, Gearing OFF, Set Forward SW Limit, Set Reverse SW Limit, Revert Forward SW Limit, Revert Reverse SW Limit}.
:type commands: list
:param motors: Array of motor names
:type motors: list
:param goals: Goal (or associated value for the command) for each motor provided in ``motors``
:type goals: list
:return: Dictionary of results, with the following key(s).
* **timed_out** (*list*) - List of motors that timed-out, if any.
* **not_found** (*list*) - List of motors that were not found, if any.
"""
return self.bcs_request('CommandMotor', dict(locals()))
def current_scan_running(self) -> dict:
"""
Returns the currently running 'integrated scan' run, or an empty string if none is running.
:return: Dictionary of results, with the following key(s).
* **running_scan** (*str*) - The name of the currently running scan, or an empty string if none.
"""
return self.bcs_request('CurrentScanRunning', dict(locals()))
def disable_breakpoints(self, name="") -> dict:
"""
Disables the breakpoint output (output-on-position) of motor controller for the named motor. Or, 'takes the motor out of flying mode'.
:param name: Motor name
:type name: str
:return: Dictionary of results (no endpoint-specific keys).
"""
return self.bcs_request('DisableBreakpoints', dict(locals()))
def disable_motor(self, name="success") -> dict:
"""
Disables the named motor.
:param name: Motor name.
:type name: str
:return: Dictionary of results (no endpoint-specific keys).
"""
return self.bcs_request('DisableMotor', dict(locals()))
def enable_motor(self, name="success") -> dict:
"""
Enables the named motor.
:param name: Motor name.
:type name: str
:return: Dictionary of results (no endpoint-specific keys).
"""
return self.bcs_request('EnableMotor', dict(locals()))
def get_acquired(self, chans=[]) -> dict:
"""
Retrieve the average value from the most recent *single-shot* acquisition (see StartAcquire).
:param chans: AI channel name(s). An empty array returns data for all AI channels on the server.
:type chans: list
:return: Dictionary of results, with the following key(s).
* **chans** (*list*) - AI channel names that have data in the **data** array.
* **not_found** (*list*) - The requested channels in ``chans`` that were not found on the host system, if any.
* **data** (*list*) - The acquired data, in the same order as the channel names in **chans**.
"""
return self.bcs_request('GetAcquired', dict(locals()))
def get_acquired_array(self, chans=[]) -> dict:
"""
Retrieve the array acquired from the most recent *single-shot* acquisition (see StartAcquire).
:param chans: AI channel name(s). An empty array returns data for all AI channels on the server.
:type chans: list
:return: Dictionary of results, with the following key(s).
* **not_found** (*list*) - The requested channels in ``chans`` that were not found on the server, if any.
* **chans** (*list*) - Array of clusters/dictionaries. Each array element contains the channel (**chan**), **period**, **time**, and **data**. Data is an array of sampled values.
"""
return self.bcs_request('GetAcquiredArray', dict(locals()))
def get_acquire_status(self) -> dict:
"""
Read the current acquisition state of the AI subsystem
:return: Dictionary of results, with the following key(s).
* **acquiring** (*bool*) - True if an acquisition is in progress
* **started** (*bool*) - True if an acquisition has been started
* **reading** (*bool*) - True if acquisition data is currently being retrieved from an acquisition device
"""
return self.bcs_request('GetAcquireStatus', dict(locals()))
def get_di(self, chans=[]) -> dict:
"""
Get digital input (DI) channel value
:param chans: DI channel name(s). An empty array returns data for all channels on the server.
:type chans: list
:return: Dictionary of results, with the following key(s).
* **chans** (*list*) - DI channel names that have data in the **data** array.
* **not_found** (*list*) - The requested channels in ``chans`` that were not found on the host system, if any.
* **enabled** (*list*) - Boolean list, in the same order as **chans**, indicating if the channel is used (enabled).
* **data** (*list*) - Boolean value of the DI's, in the same order as the channel names in **chans**.
"""
return self.bcs_request('GetDI', dict(locals()))
def get_flying_positions(self, name="") -> dict:
"""
Retrieve the locations of motor **name** that will trigger acquisitions in a flying scan.
:param name: Motor name
:type name: str
:return: Dictionary of results, with the following key(s).
* **pulse_positions** (*list*) - Trigger (pulse) positions.
"""
return self.bcs_request('GetFlyingPositions', dict(locals()))
def get_folder_listing(self, path=".", pattern="*.txt", recurse=False) -> dict:
"""
Get lists of all files and folders in a location descended from "C:\\\\Beamline Controls\\\\BCS Setup Data"
:param path: Path to text file, relative to C:\\\\Beamline Controls\\\\BCS Setup Data. Use backslashes (\\\\) not forward slashes (/).
:type path: str
:param pattern: Pattern for files for which you want to search.
:type pattern: str
:param recurse: If True, list all descendent files and folders that match pattern.
:type recurse: bool
:return: Dictionary of results, with the following key(s).
* **files** (*list*) - Array of paths, relative to C:\\\\Beamline Controls\\\\BCS Setup Data.
* **folders** (*list*) - Array of paths, relative to C:\\\\Beamline Controls\\\\BCS Setup Data.
"""
return self.bcs_request('GetFolderListing', dict(locals()))
def get_freerun(self, chans=[]) -> dict:
"""
Get freerun AI data for one or more channels
:param chans: Array of channel names to get. An empty array retrieves all channels' data.
:type chans: list
:return: Dictionary of results, with the following key(s).
* **chans** (*list*) - The retrieved channel names, in the same order as **data**.
* **not_found** (*list*) - The channels requested in ``chans`` that were not found on the server.
* **data** (*list*) - The retrieved data, corresponding to the channels in **chans**.
"""
return self.bcs_request('GetFreerun', dict(locals()))
def get_freerun_array(self, chans=[]) -> dict:
"""
Retrieve most recent AI freerun data.
:param chans: AI channel name(s). An empty array returns data for all AI channels on the server.
:type chans: list
:return: Dictionary of results, with the following key(s).
* **not_found** (*list*) - The requested channels in ``chans`` that were not found on the server, if any.
* **chans** (*list*) - Array of clusters/dictionaries. Each array element contains the channel (**chan**), **data**, and **x_values**.
"""
return self.bcs_request('GetFreerunArray', dict(locals()))
def get_instrument_acquired1d(self, name="") -> dict:
"""
Retrieve data (1D array) from the most recent acquisition.
:param name: Instrument name
:type name: str
:return: Dictionary of results, with the following key(s).
* **x** (*list*) - 1D array of abscissae
* **y** (*list*) - 1D array of data
"""
return self.bcs_request('GetInstrumentAcquired1D', dict(locals()))
def get_instrument_acquired2d(self, name="") -> dict:
"""
Retrieve data (2D array) from the most recent acquisition.
:param name: Instrument name
:type name: str
:return: Dictionary of results, with the following key(s).
* **data** (*list*) - 2D array of u32.
"""
return self.bcs_request('GetInstrumentAcquired2D', dict(locals()))
def get_instrument_acquired3d(self, name="") -> dict:
"""
Retrieve data (3D array) from the most recent acquisition.
:param name: Instrument name
:type name: str
:return: Dictionary of results, with the following key(s).
* **data** (*list*) - 3D array of u32.
"""
return self.bcs_request('GetInstrumentAcquired3D', dict(locals()))
def get_instrument_acquisition_info(self, name="") -> dict:
"""
Retrieve miscellanaeous info about the instrument and acquisition: file_name, sensor temperature (if applicable), live time, and dead time.
:param name: Instrument name
:type name: str
:return: Dictionary of results, with the following key(s).
* **file_name** (*str*) - Fully qualified path name to latest data file.
* **temperature** (*float*) - The temperature of an internal element, such as a CCD, if the instrument supports it.
* **live_time** (*list*) - Live time for the acquistion.
* **dead_time** (*list*) - Dead time for the acquisition.
"""
return self.bcs_request('GetInstrumentAcquisitionInfo', dict(locals()))
def get_instrument_acquisition_status(self, name="") -> dict:
"""
Retrieve all available status bits from the instrument subsystem for the named instrument.
:param name: Instrument name
:type name: str
:return: Dictionary of results, with the following key(s).
* **scan_params_set_up** (*bool*) - Are the scan parameters set up?
* **acquiring** (*bool*) - Is the instrument acquiring data?
* **single_shot** (*bool*) - Unclear at the moment. TBD
* **aborted** (*bool*) - Is the acquisition aborted?
* **data_available** (*bool*) - Is data available?
* **ins_single_shot** (*bool*) - 'Instrument Single Shot': Also unclear. TBD.
"""
return self.bcs_request('GetInstrumentAcquisitionStatus', dict(locals()))
def get_instrument_count_rates(self, name="") -> dict:
"""
Retreive instrument count rates
:param name: Instrument name
:type name: str
:return: Dictionary of results, with the following key(s).
* **input_cr** (*list*) - Input count rate
* **output_cr** (*list*) - Output count rate
"""
return self.bcs_request('GetInstrumentCountRates', dict(locals()))
def get_instrument_driver_status(self, name="") -> dict:
"""
Returns the status of the BCS Instrument Driver for the named instrument.
:param name: Instrument name
:type name: str
:return: Dictionary of results, with the following key(s).
* **running** (*bool*) - True if the driver is running.
"""
return self.bcs_request('GetInstrumentDriverStatus', dict(locals()))
def get_motor(self, motors=[]) -> dict:
"""
Get information and status for the motors in ``motors``
:param motors: Array of motor names to retrieve.
:type motors: list
:return: Dictionary of results, with the following key(s).
* **not_found** (*list*) - The requested motors in ``motors`` that were not found on the host system, if any.
* **data** (*list*) - Array of clusters/dictionaries. Each array element contains the motor name (**motor**), **position**, **position_raw**, **goal**, **goal_raw**, **status**, and **time**.
**status** is a bit-field. TODO: helper function to test status
"""
return self.bcs_request('GetMotor', dict(locals()))
def get_motor_full(self, motors=[]) -> dict:
"""
Returns the complete state of the requested motors.
:param motors: Names of motors to query.
:type motors: list
:return: Dictionary of results, with the following key(s).
* **not_found** (*list*) - Names in **motors** that were not found, if any.
* **data** (*list*) - dictionary of dictionaries all relateing to state. Too many parameters to list atm.
"""
return self.bcs_request('GetMotorFull', dict(locals()))
def get_panel_image(self, name="", quality=0) -> dict:
"""
Send a current image of the LabVIEW panel, in jpg format.
:param name: Full path to panel
:type name: str
:param quality: JPEG quality 0-100
:type quality: int
:return: Dictionary of results, with the following key(s).
* **image_blob** (*dict*) - Ideally opaque blob for holding binary data. Helper functions should be available to clients to extract blob contents. Contains **length** and **blob**.
"""
return self.bcs_request('GetPanelImage', dict(locals()))
def get_state_variable(self, name="") -> dict:
"""
Get the value of the named BCS State Variable.
:param name: Name of the state variable to retrieve
:type name: str
:return: Dictionary of results, with the following key(s).
* **value** (*void*) - Value of the state variable
* **found** (*bool*) - True if requested name was found on the server
* **type** (*enum u16*) - 'String', 'Boolean', 'Integer', or 'Double'
* **name** (*str*) - Same as ``name``
"""
return self.bcs_request('GetStateVariable', dict(locals()))
def get_subsystem_status(self) -> dict:
"""
Returns the current status of all BCS subsystems on the server
:return: Dictionary of results, with the following key(s).
* **status** (*list*) - Array of dictionaryies of keys: **name** - Subsystem Name, and **status** - one of {Initializing, Running, Stopping, Stopped, Force Stop, Disabled, Bad Path, VI Broken Not in List}
"""
return self.bcs_request('GetSubsystemStatus', dict(locals()))
def get_text_file(self, path="You must have at least one input control in the Parameter cluster. Delete this (_unused) if you don't need it.") -> dict:
"""
Get contents of a text file (usually acquired data) from any file in a location descended from "C:\\\\Beamline Controls\\\\BCS Setup Data".
:param path: Path to text file, relative to C:\\\\Beamline Controls\\\\BCS Setup Data. Use backslashes (\\\\) not forward slashes (/).
:type path: str
:return: Dictionary of results, with the following key(s).
* **text** (*str*) - Text of the file. Platform-dependent end-of-line characters have been converted to line feed characters
"""
return self.bcs_request('GetTextFile', dict(locals()))
def get_video_image(self, name="", quality=0, type="default") -> dict:
"""
Return the most recent image from the named camera, in jpg format.
:param name: Video (camera) name
:type name: str
:param quality: JPEG quality 0-100
:type quality: int
:param type: One of [default, roi, or threshold]
:type type: enum u16
:return: Dictionary of results, with the following key(s).
* **image_blob** (*dict*) - Ideally opaque blob for holding binary data. Helper functions should be available to clients to extract blob contents. Contains **length** and **blob**.
"""
return self.bcs_request('GetVideoImage', dict(locals()))
def home_motor(self, motors=[]) -> dict:
"""
Home the motors in the input array, ``motors``.
:param motors: Names of the motors to home.
:type motors: list
:return: Dictionary of results, with the following key(s).
* **timed_out** (*list*) - True if the home operation timed out.
* **not_found** (*list*) - The motors in ``motors`` that are not found on the server, if any.
"""
return self.bcs_request('HomeMotor', dict(locals()))
def last_scan_run(self) -> dict:
"""
Returns the last 'integrated scan' run, or an empty string if none have run.
:return: Dictionary of results, with the following key(s).
* **last_scan** (*str*) - The name of the last run scan, or an empty string if none.
"""
return self.bcs_request('LastScanRun', dict(locals()))
def list_ais(self) -> dict:
"""
Return the list of all AI channels that are defined on the server.
:return: Dictionary of results, with the following key(s).
* **names** (*list*) - Llist of AI channels that are defined on the server. The list includes disabled and hidden channels.
* **displayed** (*list*) - The list of AI channel that are displayed (not set to 'hidden' in BCS).
* **disabled** (*list*) - Iindexes in **displayed** that are disabled.
"""
return self.bcs_request('ListAIs', dict(locals()))
def list_dios(self) -> dict:
"""
Retrieves the complete list of digital input channel names defined on the server.
:return: Dictionary of results, with the following key(s).
* **names** (*list*) - Liist of all digital input channel names on the server. The list includes disabled and hidden channels.
* **displayed** (*list*) - The list of channels that are displayed (not set to 'hidden' in BCS).
* **disabled** (*list*) - Iindexes in **displayed** that are disabled.
"""
return self.bcs_request('ListDIOs', dict(locals()))
def list_instruments(self) -> dict:
"""
Return the list of instruments that are defined on the server.
:return: Dictionary of results, with the following key(s).
* **names** (*list*) - The list of all instruments that are defined on the server, including disabled and hidden instruments.
* **displayed** (*list*) - The list of instruments that are displayed (not set to 'hidden' in BCS).
* **disabled** (*list*) - Iindexes in **displayed** that are disabled.
"""
return self.bcs_request('ListInstruments', dict(locals()))
def list_motors(self) -> dict:
"""
Return the list of motors that are defined on the server.
:return: Dictionary of results, with the following key(s).
* **names** (*list*) - The list of motors that are defined on the server. The list includes disabled and hidden motors.
* **displayed** (*list*) - The list of motors that are displayed (not set to 'hidden' in BCS).
* **disabled** (*list*) - Iindexes in **displayed** that are disabled.
"""
return self.bcs_request('ListMotors', dict(locals()))
def list_presets(self) -> dict:
"""
Return array of motor preset positions. Each array entry is a dictionary describing one preset. The dictionary keys are **Preset Name**, **Motor Name**, **Preset Position**, **Tolerance (+/-)**.
:return: Dictionary of results, with the following key(s).
* **presets** (*list*) - Array of dictionaries, one per preset, each with keys: **Preset Name**, **Motor Name**, **Preset Position**, **Tolerance (+/-)**.
"""
return self.bcs_request('ListPresets', dict(locals()))
def list_state_variables(self) -> dict:
"""
Get the complete list of state variable, and their types (Boolean, Integer, String, Double).
:return: Dictionary of results, with the following key(s).
* **names** (*list*) - Current list of state variable on the server.
* **types** (*list*) - Variable types (Boolean, Integer, String, Double), in the same order as **names**.
"""
return self.bcs_request('ListStateVariables', dict(locals()))
def list_trajectories(self) -> dict:
"""
List all trajectories.
:return: Dictionary of results, with the following key(s).
* **trajectories** (*list*) - Array of dictionaries, each element describing a trajectory.
"""
return self.bcs_request('ListTrajectories', dict(locals()))
def move_motor(self, motors=[], goals=[]) -> dict:
"""
Command one or more motors to begin moves to supplied goals.
:param motors: Array of motor names
:type motors: list
:param goals: Goal for each motor listed in ``motors``
:type goals: list
:return: Dictionary of results, with the following key(s).
* **timed_out** (*list*) - List of motors that timed-out, if any.
* **not_found** (*list*) - List of motors that were not found, if any.
"""
return self.bcs_request('MoveMotor', dict(locals()))
def move_to_preset(self, names=[]) -> dict:
"""
Move to preset positions. Takes a list of preset names and executes them (sends their motors to their respective positions).
:param names: Array of preset names to execute.
:type names: list
:return: Dictionary of results, with the following key(s).
* **not_found** (*list*) - List of names in ``names`` that were not found on the server, if any.
"""
return self.bcs_request('MoveToPreset', dict(locals()))
def move_to_trajectory(self, names=[]) -> dict:
"""
Move to trajectory positions. Takes a list of trajectory names and executes them (sends their motors to their respective positions).
:param names: Array of trajectory names to execute.
:type names: list
:return: Dictionary of results, with the following key(s).
* **not_found** (*list*) - List of names in ``names`` that were not found on the server, if any.
"""
return self.bcs_request('MoveToTrajectory', dict(locals()))
def scan_status(self) -> dict:
"""
Returns information about the 'integrated scan' system, namely the last scan (last_scan) run, the currently running scan, and the scanner status. An empty string indicates no scan.
:return: Dictionary of results, with the following key(s).
* **last_scan** (*str*) - The name of the last run scan, or an empty string if none.
* **running_scan** (*str*) - The name of the currently running scan, or an empty string if none.
* **scanner_status** (*str*) - The current status of the scanner. One of a large set of finite-state-machine states used by the scanner. The set also varies by scan. Empty string if the scanner has stopped running.
* **log_directory** (*str*) - Location of data file, maybe.
* **last_filename** (*str*) - Data file name
* **user_path** (*str*) - Or maybe this is the location of the data file
"""
return self.bcs_request('ScanStatus', dict(locals()))
def set_breakpoints(self, name="", x0=0, dx=0, n=0, breakpoints=[], counts_or_units=False) -> dict:
"""
Set Breakpoints for the named motor. Breakpoints are the positions at which the controller will generate acquisition triggers during a flying scan. Specify either {**x0**, **dx**, **n**} to generate a regular grid, or send an arbitrary list in **breakpoints**.
:param name: Motor
:type name: str
:param x0: Starting position.
:type x0: float
:param dx: Interval spacing.
:type dx: float
:param n: Number of breakpoints (triggers) to generate.
:type n: int
:param breakpoints: An array of arbitrary locations to on which to trigger data acquisition.
:type breakpoints: list
:param counts_or_units: Are the locations in motor units or motor counts (False == units)?
:type counts_or_units: bool
:return: Dictionary of results (no endpoint-specific keys).
"""
return self.bcs_request('SetBreakpoints', dict(locals()))
def set_do(self, chan="success", value=False) -> dict:
"""
Sets the digital output (DO) channel ``chan`` to ``value``
:param chan: Name of the channel to set.
:type chan: str
:param value: New value
:type value: bool
:return: Dictionary of results (no endpoint-specific keys).
"""
return self.bcs_request('SetDO', dict(locals()))
def set_motor_velocity(self, motors=[], velocities=[]) -> dict:
"""
Set motor speeds. This endpoint duplicates the functionality of CommandMotor, and may be removed from future releases.
:param motors: List of motors
:type motors: list
:param velocities: Speeds to set, in the same order as **motors**.
:type velocities: list
:return: Dictionary of results, with the following key(s).
* **timed_out** (*list*) -
* **not_found** (*list*) -
"""
return self.bcs_request('SetMotorVelocity', dict(locals()))
def set_state_variable(self, name="variable name", value=0) -> dict:
"""
Set the value of the named BCS State Variable.
:param name: Name of the state variable to set
:type name: str
:param value: New value of the state variable. Accepts string, numeric, and boolean values.
:type value: int
:return: Dictionary of results, with the following key(s).
* **found** (*bool*) - True if requested name was found on the server
"""
return self.bcs_request('SetStateVariable', dict(locals()))
def start_acquire(self, time=0, counts=0) -> dict:
"""
Start Acquisition for either ``time`` or ``counts``, whichever is non-zero. The acquisition is started, but the endpoint does not wait for the acquisition to complete.
:param time: Acquisition time.
:type time: float
:param counts: Acquisition counts.
:type counts: int
:return: Dictionary of results (no endpoint-specific keys).
"""
return self.bcs_request('StartAcquire', dict(locals()))
def start_flying_scan(self, name="", xi=0, xf=0, dx=0, speed=0) -> dict:
"""
Start a flying scan with the named motor. The scan is started. The enpoint does not wait for the scan to finish.
:param name: Motor name
:type name: str
:param xi: Starting point
:type xi: float
:param xf: Stopping point
:type xf: float
:param dx: Interval between triggers (pulses)
:type dx: float
:param speed: Motor speed
:type speed: float
:return: Dictionary of results (no endpoint-specific keys).
"""