This repository has been archived by the owner on Nov 29, 2018. It is now read-only.
-
Notifications
You must be signed in to change notification settings - Fork 6
/
85-robot.py
241 lines (191 loc) · 8.09 KB
/
85-robot.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
import asyncio
from ophyd import Device, EpicsSignal, EpicsSignalRO
from ophyd import Component as C
from ophyd.utils import set_and_wait
from bluesky import Msg
from bluesky.plans import count, list_scan
from bluesky.plan_stubs import abs_set, open_run, close_run
from bluesky.utils import single_gen
from bluesky.preprocessors import subs_wrapper, pchain
from bluesky.callbacks import LiveTable
from bluesky import Msg
class Robot(Device):
sample_number = Cpt(EpicsSignal, 'ID:Tgt-SP')
load_cmd = Cpt(EpicsSignal, 'Cmd:Load-Cmd.PROC')
unload_cmd = Cpt(EpicsSignal, 'Cmd:Unload-Cmd.PROC')
execute_cmd = Cpt(EpicsSignal, 'Cmd:Exec-Cmd')
status = Cpt(EpicsSignal, 'Sts-Sts')
current_sample_number = Cpt(EpicsSignalRO, 'Addr:CurrSmpl-I')
# Map sample types to their load position and measurement position.
TH_POS = {'capillary': {'load': None, 'measure': None},
'plate': {'load': 0, 'measure': 90},
None: {'load': None, 'measure': None}, }
REL_MOVES = ['plate']
DIFF_POS = {'capilary': (1, 2), }
def __init__(self, *args, theta, diff=None, **kwargs):
"""
Parameters
----------
theta : motor
diff : motor, optional [not currently used]
"""
self.theta = theta
self._current_sample_geometry = None
super().__init__(*args, **kwargs)
def _poll_until_idle(self):
time.sleep(3) # give it plenty of time to start
while self.status.get() != 'Idle':
time.sleep(.1)
def _poll_until_sample_cleared(self):
while self.current_sample_number.get() != 0:
time.sleep(.1)
def load_sample(self, sample_number, sample_geometry=None):
# If no sample is loaded, current_sample_number=0
# is reported by the robot.
if self.current_sample_number.get() != 0:
raise RuntimeError(
"Sample %d is already loaded." % self.current_sample_number.get())
# Rotate theta into loading position if necessary (e.g. flat plate mode).
load_pos = self.TH_POS[sample_geometry]['load']
if load_pos is not None:
if sample_geometry not in self.REL_MOVES:
print('Moving theta to load position')
self.theta.move(load_pos, wait=True)
# Loading the sample is a three-step procedure:
# Set sample_number; issue load_cmd; issue execute_cmd.
set_and_wait(self.sample_number, sample_number)
set_and_wait(self.load_cmd, 1)
self.execute_cmd.put(1)
print('Loading...')
self._poll_until_idle()
# Rotate theta into measurement position if necessary (e.g. flat plate mode).
measure_pos = self.TH_POS[sample_geometry]['measure']
if measure_pos is not None:
print('Moving theta to measure position')
if sample_geometry not in self.REL_MOVES:
self.theta.move(measure_pos, wait=True)
else:
pos = self.theta.get()
self.theta.move(pos + measure_pos, wait=True)
# Stash the current sample geometry for reference when we unload.
self._current_sample_geometry = sample_geometry
def unload_sample(self):
if self.current_sample_number.get() == 0:
# there is nothing to do
return
# Rotate theta into loading position if necessary (e.g. flat plate mode).
load_pos = self.TH_POS[self._current_sample_geometry]['load']
measure_pos = self.TH_POS[self._current_sample_geometry]['measure']
if load_pos is not None:
print('Moving theta to measure position')
if self._current_sample_geometry not in self.REL_MOVES:
self.theta.move(load_pos, wait=True)
else:
pos = self.theta.get()
self.theta.move(pos - measure_pos, wait=True)
if load_pos is not None:
print('Moving theta to unload position')
self.theta.move(load_pos, wait=True)
set_and_wait(self.unload_cmd, 1)
self.execute_cmd.put(1)
print('Unloading...')
self._poll_until_idle()
self._poll_until_sample_cleared()
self._current_sample_geometry = None
def stop(self):
self.theta.stop()
super().stop()
# Define custom commands for sample loading/unloading.
@asyncio.coroutine
def _load_sample(msg):
msg.obj.load_sample(*msg.args, **msg.kwargs)
@asyncio.coroutine
def _unload_sample(msg):
msg.obj.unload_sample(*msg.args, **msg.kwargs)
# Register these custom command with the RunEngine.
RE.register_command('load_sample', _load_sample)
RE.register_command('unload_sample', _unload_sample)
# For convenience, define short plans the use these custom commands.
def load_sample(position, geometry=None):
# TODO: I think this can be simpler.
return (
yield from single_gen(Msg('load_sample', robot, position, geometry)))
def unload_sample():
# TODO: I think this can be simpler.
return (yield from single_gen(Msg('unload_sample', robot)))
# These are usable bluesky plans.
def robot_wrapper(plan, sample):
"""Wrap a plan in load/unload messages.
Parameters
----------
plan : a bluesky plan
sample : dict
must contain 'position'; optionally also 'geometry'
Example
-------
>>> plan = count([pe1c])
>>> new_plan = robot_wrapper(plan, {'position': 1})
"""
yield from load_sample(sample['position'], sample.get('geometry', None))
yield from plan
yield from unload_sample()
def ct(sample, exposure):
"""
Capture how many exposures are needed to get a total exposure
of the given value, and sum those into one file before saving.
"""
pe1c.images_per_set.put(1)
pe1c.number_of_sets.put(1)
plan = subs_wrapper(count([pe1c], num=1), LiveTable([]))
# plan = robot_wrapper(plan, sample)
yield from plan
def tseries(sample, exposure, num):
"""
Capture how ever many exposures are needed to get a total exposure
of the given value, and divide those into files of 'num' exposures
each, summed.
"""
if pe1c.cam.acquire_time.get() != 0.1:
raise RuntimeError("We expect pe1c.cam.acquire_time to be 0.1")
pe1c.images_per_set.put(num)
pe1c.number_of_sets.put(exposure // num)
plan = subs_wrapper(count([pe1c], num=1), LiveTable([]))
i # plan = robot_wrapper(plan, sample)
yield from plan
def Tramp(sample, exposure, start, stop, step):
Tpoints = np.arange(start, stop, step)
plan = subs_wrapper(list_scan([pec1], cs700, points), LiveTable([cs700]))
# plan = robot_wrapper(plan, sample)
yield from plan
# Define list of sample info.
samples = [{'position': 1, 'geometry': 'capillary', 'sample_name': 'stuff'},
{'position': 2, 'geometry': 'plate', 'sample_name': 'other_stuff'}]
def example():
for sample in samples:
# Define a normal plan, from bluesky or xpdacq.
plan = bp.count([], md={'sample_name': sample['sample_name']})
# Modify the plan by stacking robot instructions before and after.
# It also needs the sample, from which it extracts the numerical
# sample 'position' (required) # and 'geometry' (optional).
# Recognized geometries are None (default), 'capillary', and 'plate'.
yield from robot_wrapper(plan, sample)
def excel_example(filename, geometry=None):
"""
Example: RE(excel_example('/XF28IDC/XF28ID2/pe2_data/xpdUser/Import/example-with-dan.xlsx'))
"""
import pandas as pd
f = pd.ExcelFile(filename)
sheet = f.parse()
for _, row in sheet.iloc[1:].iterrows():
name, phase, bg_name, position = row[:4]
position = int(position)
sample = {'sample_name': name,
'phase_info': phase,
'bg_name': bg_name,
'position': position,
'geometry': geometry}
plan = bp.count([], md=sample)
yield from robot_wrapper(plan, sample)
# master_plan = pchain(plan(sample) for sample in samples)
robot = Robot('XF:28IDC-ES:1{SM}', name='robot', theta=th)
# old RobotPositioner code is .ipython/profile_2015_collection/startup/robot.py