-
Notifications
You must be signed in to change notification settings - Fork 2
/
Copy pathbeam_stability_scan.py
executable file
·152 lines (115 loc) · 6.11 KB
/
beam_stability_scan.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
#!/usr/bin/env python
# -*- coding: utf-8 -*-
'''
Beam stability scan.
'''
import os
import time
import gevent
from xray_experiment import xray_experiment
from monitor import xray_camera, analyzer
class beam_stability_scan(xray_experiment):
specific_parameter_fields =[{'name': 'observation_period', 'type': 'float', 'description': 'Intended observation period in s'},
{'name': 'default_slit_gap', 'type': 'float', 'description': 'Default slits gap in mm'}]
def __init__(self,
name_pattern,
directory,
observation_period=300.,
default_slit_gap=4.,
transmission=None,
diagnostic=True,
analysis=None,
conclusion=None,
simulation=None,
display=False,
extract=False):
if hasattr(self, 'parameter_fields'):
self.parameter_fields += beam_stability_scan.specific_parameter_fields
else:
self.parameter_fields = beam_stability_scan.specific_parameter_fields[:]
xray_experiment.__init__(self,
name_pattern,
directory,
transmission=transmission,
diagnostic=diagnostic,
analysis=analysis,
conclusion=conclusion,
simulation=simulation)
self.description = 'Beam stability scan, Proxima 2A, SOLEIL, %s' % (time.ctime(self.timestamp),)
self.observation_period = observation_period
self.default_slit_gap = default_slit_gap
self.extract = extract
self.display = display
self.monitor_device = None
self.total_expected_exposure_time = self.observation_period
self.total_expected_wedges = 1
def set_up_monitor(self):
self.monitor_device = xray_camera()
self.auxiliary_monitor_device = analyzer(continuous_monitor_name='focus_monitor')
self.monitors_dictionary['analyzer'] = self.auxiliary_monitor_device
self.monitor_names += ['analyzer']
self.monitors += [self.auxiliary_monitor_device]
def prepare(self):
self.set_up_monitor()
self.check_directory(self.directory)
self.write_destination_namepattern(self.directory, self.name_pattern)
initial_settings_a = []
if self.simulation != True:
initial_settings_a.append(gevent.spawn(self.goniometer.set_transfer_phase, wait=False))
initial_settings_a.append(gevent.spawn(self.set_photon_energy, self.photon_energy, wait=True))
for k in self.get_clean_slits():
initial_settings_a.append(gevent.spawn(getattr(getattr(self, 'slits%d' % k), 'set_horizontal_gap'), self.default_slit_gap))
initial_settings_a.append(gevent.spawn(getattr(getattr(self, 'slits%d' % k), 'set_vertical_gap'), self.default_slit_gap))
if self.safety_shutter.closed():
initial_settings_a.append(gevent.spawn(self.safety_shutter.open))
gevent.joinall(initial_settings_a)
initial_settings_b = []
for k in self.get_clean_slits():
initial_settings_b.append(gevent.spawn(getattr(getattr(self, 'slits%d' % k), 'set_horizontal_position'), 0))
initial_settings_b.append(gevent.spawn(getattr(getattr(self, 'slits%d' % k), 'set_vertical_position'), 0))
initial_settings_b.append(gevent.spawn(self.monitor_device.insert))
gevent.joinall(initial_settings_b)
def run(self):
print('in run')
print('self.fast_shutter', self.fast_shutter)
self.fast_shutter.open()
gevent.sleep(self.observation_period)
self.fast_shutter.close()
def clean(self):
self.collect_parameters()
self.save_parameters()
self.save_results()
self.save_log()
final_settings_a = []
for k in self.get_clean_slits(): # [1, 2, 3, 5, 6]:
final_settings_a.append(gevent.spawn(getattr(getattr(self, 'slits%d' % k), 'set_horizontal_gap'), self.default_slit_gap))
final_settings_a.append(gevent.spawn(getattr(getattr(self, 'slits%d' % k), 'set_vertical_gap'), self.default_slit_gap))
gevent.joinall(final_settings_a)
final_settings_b = []
for k in self.get_clean_slits(): #[1, 2, 3, 5, 6]:
final_settings_b.append(gevent.spawn(getattr(getattr(self, 'slits%d' % k), 'set_horizontal_position'), 0))
final_settings_b.append(gevent.spawn(getattr(getattr(self, 'slits%d' % k), 'set_vertical_position'), 0))
gevent.joinall(final_settings_b)
if self.extract:
final_settings_b.append(gevent.spawn(self.monitor_device.extract))
gevent.joinall(final_settings_b)
def main():
import optparse
usage = '''Program will execute a slit scan
./slit_scan.py <options>
'''
parser = optparse.OptionParser(usage=usage)
parser.add_option('-d', '--directory', type=str, default='/tmp/slit_scan', help='Directory to store the results (default=%default)')
parser.add_option('-n', '--name_pattern', type=str, default='slit_scan', help='name_pattern')
parser.add_option('-s', '--observation_period', type=float, default=300, help='observation period')
parser.add_option('-D', '--display', action='store_true', help='display plot')
parser.add_option('-E', '--extract', action='store_true', help='Extract the calibrated diode after the scan')
options, args = parser.parse_args()
print('options', options)
print('args', args)
filename = os.path.join(options.directory, options.name_pattern) + '_parameters.pickle'
stability = beam_stability_scan(**vars(options))
if not os.path.isfile(filename):
stability.execute()
if __name__ == '__main__':
main()