Beamline checks failed at" - f" {str(datetime.datetime.now())}: {''.join(self._check_msgs)}
" - ).add_tag(["BEC", "beam_check"]) - self.client.logbook.send_logbook_message(msg) - except Exception: - logger.warning("Failed to send update to SciLog.") - - while True: - self._beam_is_okay = True - self._check_msgs = self._run_beamline_checks() - if self._beam_is_okay: - break - self._print_beamline_checks() - time.sleep(1) - - try: - msg = bec.logbook.LogbookMessage() - msg.add_text( - "Operation resumed at" - f" {str(datetime.datetime.now())}.
" - ).add_tag(["BEC", "beam_check"]) - self.client.logbook.send_logbook_message(msg) - except Exception: - logger.warning("Failed to send update to SciLog.") - - def rt_off(self): - dev.rtx.enabled = False - dev.rty.enabled = False - - def rt_on(self): - dev.rtx.enabled = True - dev.rty.enabled = True - if dev.rtx.enabled == True: - print("rt is enabled") - else: - print("failed to enable rt") - - def feedback_enable_with_reset(self): - self.device_manager.devices.rtx.controller.feedback_enable_with_reset() - self.feedback_status() - - def feedback_enable_without_reset(self): - self.device_manager.devices.rtx.controller.feedback_enable_without_reset() - self.feedback_status() - - def feedback_disable(self): - self.device_manager.devices.rtx.controller.feedback_disable() - self.feedback_status() - - def feedback_disable_and_reset_angle(self): - self.device_manager.devices.rtx.controller.feedback_disable_and_even_reset_lamni_angle_interferometer() - self.feedback_status() - - def feedback_status(self): - if self.device_manager.devices.rtx.controller.feedback_is_running(): - print("The rt feedback is \x1b[92mrunning\x1b[0m.") - else: - print("The rt feedback is \x1b[91mNOT\x1b[0m running.") - - def show_interferometer_positions(self): - self.device_manager.devices.rtx.controller.show_interferometer_positions() - - def show_signal_strength(self): - self.device_manager.devices.rtx.controller.show_signal_strength_interferometer() - - def show_analog_signals(self): - return self.device_manager.devices.rtx.controller.show_analog_signals() - - def add_sample_database( - self, samplename, date, eaccount, scan_number, setup, sample_additional_info, user - ): - """Add a sample to the omny sample database. This also retrieves the tomo id.""" - subprocess.run( - "wget --user=omny --password=samples -q -O /tmp/currsamplesnr.txt" - f" 'https://omny.web.psi.ch/samples/newmeasurement.php?sample={samplename}&date={date}&eaccount={eaccount}&scannr={scan_number}&setup={setup}&additional={sample_additional_info}&user={user}'", - shell=True, - ) - with open("/tmp/currsamplesnr.txt") as tomo_number_file: - tomo_number = int(tomo_number_file.read()) - return tomo_number - - def _at_each_angle(self, angle: float) -> None: - self.tomo_scan_projection(angle) - self.tomo_reconstruct() - - ### XMCD ### - # 2 projections, 1 for each polarization state - # cp() - # self.tomo_scan_projection(angle) - # self.tomo_reconstruct() - # cm() - # self.tomo_scan_projection(angle) - # self.tomo_reconstruct() - - def _write_subtomo_to_scilog(self, subtomo_number): - """Write subtomo start information to scilog.""" - bec = builtins.__dict__.get("bec") - if self.tomo_id > 0: - tags = ["BEC_subtomo", self.sample_name, f"tomo_id_{self.tomo_id}"] - else: - tags = ["BEC_subtomo", self.sample_name] - self.write_to_scilog( - f"Starting subtomo: {subtomo_number}. First scan number: {bec.queue.next_scan_number}.", - tags, - ) - - def sub_tomo_scan(self, subtomo_number, start_angle=None): - """start a subtomo""" - self._write_subtomo_to_scilog(subtomo_number) - - if start_angle is None: - if subtomo_number == 1: - start_angle = 0 - elif subtomo_number == 2: - start_angle = self.tomo_angle_stepsize / 8.0 * 4 - elif subtomo_number == 3: - start_angle = self.tomo_angle_stepsize / 8.0 * 2 - elif subtomo_number == 4: - start_angle = self.tomo_angle_stepsize / 8.0 * 6 - elif subtomo_number == 5: - start_angle = self.tomo_angle_stepsize / 8.0 * 1 - elif subtomo_number == 6: - start_angle = self.tomo_angle_stepsize / 8.0 * 5 - elif subtomo_number == 7: - start_angle = self.tomo_angle_stepsize / 8.0 * 3 - elif subtomo_number == 8: - start_angle = self.tomo_angle_stepsize / 8.0 * 7 - - # _tomo_shift_angles (potential global variable) - _tomo_shift_angles = 0 - angle_end = start_angle + 360 - angles = np.linspace( - start_angle + _tomo_shift_angles, - angle_end, - num=int(360 / self.tomo_angle_stepsize) + 1, - endpoint=True, - ) - - # reverse even sub-tomograms - if not (subtomo_number % 2): - angles = np.flip(angles) - for angle in angles: - self.progress["tomo_type"] = "Equally spaced sub-tomograms" - self.progress["subtomo"] = subtomo_number - self.progress["subtomo_projection"] = np.where(angles == angle)[0][0] - self.progress["subtomo_total_projections"] = 360 / self.tomo_angle_stepsize - self.progress["projection"] = (subtomo_number - 1) * self.progress[ - "subtomo_total_projections" - ] + self.progress["subtomo_projection"] - self.progress["total_projections"] = 360 / self.tomo_angle_stepsize * 8 - self.progress["angle"] = angle - self._tomo_scan_at_angle(angle, subtomo_number) - - def _print_progress(self): - print("\x1b[95mProgress report:") - print(f"Tomo type: ....................... {self.progress['tomo_type']}") - print(f"Projection: ...................... {self.progress['projection']}") - print(f"Total projections expected ....... {self.progress['total_projections']}") - print(f"Angle: ........................... {self.progress['angle']}") - print(f"Current subtomo: ................. {self.progress['subtomo']}") - print(f"Current projection within subtomo: {self.progress['subtomo_projection']}\x1b[0m") - - def _tomo_scan_at_angle(self, angle, subtomo_number): - successful = False - error_caught = False - if 0 <= angle < 360.05: - print(f"Starting LamNI scan for angle {angle} in subtomo {subtomo_number}") - self._print_progress() - while not successful: - self._start_beam_check() - if not self.special_angles: - self._current_special_angles = [] - if self._current_special_angles: - next_special_angle = self._current_special_angles[0] - if np.isclose(angle, next_special_angle, atol=0.5): - self._current_special_angles.pop(0) - num_repeats = self.special_angle_repeats - else: - num_repeats = 1 - try: - start_scan_number = bec.queue.next_scan_number - for i in range(num_repeats): - self._at_each_angle(angle) - error_caught = False - except AlarmBase as exc: - if exc.alarm_type == "TimeoutError": - bec.queue.request_queue_reset() - time.sleep(2) - error_caught = True - else: - num_repeats = 1 - try: - start_scan_number = bec.queue.next_scan_number - for i in range(num_repeats): - self._at_each_angle(angle) - error_caught = False - except AlarmBase as exc: - if exc.alarm_type == "TimeoutError": - bec.queue.request_queue_reset() - time.sleep(2) - error_caught = True - else: - raise exc - - if self._was_beam_okay() and not error_caught: - successful = True - else: - self._wait_for_beamline_checks() - end_scan_number = bec.queue.next_scan_number - for scan_nr in range(start_scan_number, end_scan_number): - self._write_tomo_scan_number(scan_nr, angle, subtomo_number) - - if self._was_beam_okay() and not error_caught: - successful = True - else: - self._wait_for_beamline_checks() - - def _write_tomo_scan_number(self, scan_number: int, angle: float, subtomo_number: int) -> None: - tomo_scan_numbers_file = os.path.expanduser( - "~/Data10/specES1/dat-files/tomography_scannumbers.txt" - ) - with open(tomo_scan_numbers_file, "a+") as out_file: - # pylint: disable=undefined-variable - out_file.write( - f"{scan_number} {angle} {dev.lsamrot.read()['lsamrot']['value']:.3f} {self.tomo_id} {subtomo_number} {0} {'lamni'}\n" - ) - - def _golden(self, ii, howmany_sorted, maxangle=360, reverse=False): - """Return the ii-th golden ratio angle within sorted bunches of size howmany_sorted, - and its subtomo number. Operates over maxangle degrees (360 for LamNI).""" - golden = [] - for iji in range( - (ii - (ii % howmany_sorted)), (ii - (ii % howmany_sorted)) + howmany_sorted, 1 - ): - golden.append( - ((iji * maxangle * (1 + pow(5, 0.5)) / 2) * 1000 % (maxangle * 1000)) / 1000 - ) - golden.sort() - subtomo_number = int(ii / howmany_sorted) + 1 - if reverse and not subtomo_number % 2: - golden.reverse() - return (golden[ii % howmany_sorted], subtomo_number) - - def _golden_equally_spaced(self, ii, number_of_projections_per_subtomo, maxangle=360, reverse=True, verbose=False): - """Return angles for equally spaced tomography with sub-tomogram starting angles - shifted according to the golden ratio. Operates over maxangle degrees (360 for LamNI). - ii is the projection number starting at 0.""" - angular_step = maxangle / number_of_projections_per_subtomo - subtomo_number = int(((ii) * angular_step) / maxangle) + 1 - start_angle = self._golden(subtomo_number - 1, 1, angular_step)[0] - projection_number_of_subtomo = ( - ii - (subtomo_number - 1) * number_of_projections_per_subtomo - ) - - if reverse: - if subtomo_number % 2: - angle = start_angle + projection_number_of_subtomo * angular_step - else: - angle = ( - start_angle - + (number_of_projections_per_subtomo - 1) * angular_step - - projection_number_of_subtomo * angular_step - ) - else: - angle = start_angle + projection_number_of_subtomo * angular_step - - if verbose: - print( - f"Equally spaced golden ratio tomography.\n" - f"Angular step: {angular_step}\n" - f"Subtomo Number: {subtomo_number}\n" - f"Angle: {angle}" - ) - - return angle, subtomo_number - - def tomo_scan(self, subtomo_start=1, start_angle=None, projection_number=None): - """Start a tomo scan. - - Args: - subtomo_start (int): For tomo_type 1, the sub-tomogram number to start from. Defaults to 1. - start_angle (float, optional): Override the starting angle of the first sub-tomogram. Defaults to None. - projection_number (int, optional): For tomo_types 2 and 3, resume from this projection index. Defaults to None. - """ - bec = builtins.__dict__.get("bec") - scans = builtins.__dict__.get("scans") - self._current_special_angles = self.special_angles.copy() - - # Register a new tomo scan in the database and write the PDF report - # only when starting fresh (not resuming mid-scan) - if ( - (self.tomo_type == 1 and subtomo_start == 1 and start_angle is None) - or (self.tomo_type == 2 and projection_number is None) - or (self.tomo_type == 3 and projection_number is None) - ): - # pylint: disable=undefined-variable - self.tomo_id = self.add_sample_database( - self.sample_name, - str(datetime.date.today()), - bec.active_account.decode(), - bec.queue.next_scan_number, - "lamni", - "test additional info", - "BEC", - ) - self.write_pdf_report() - - with scans.dataset_id_on_hold: - if self.tomo_type == 1: - # 8 equally spaced sub-tomograms over 360 degrees - self.progress["tomo_type"] = "Equally spaced sub-tomograms" - for ii in range(subtomo_start, 9): - self.sub_tomo_scan(ii, start_angle=start_angle) - start_angle = None - - elif self.tomo_type == 2: - # Golden ratio tomography (sorted bunches) over 360 degrees - self.progress["tomo_type"] = "Golden ratio tomography" - previous_subtomo_number = -1 - ii = 0 if projection_number is None else projection_number - while True: - angle, subtomo_number = self._golden( - ii, self.golden_ratio_bunch_size, maxangle=360, reverse=True - ) - if previous_subtomo_number != subtomo_number: - self._write_subtomo_to_scilog(subtomo_number) - if ( - subtomo_number % 2 == 1 - and ii > 10 - and self.golden_projections_at_0_deg_for_damage_estimation == 1 - ): - self._tomo_scan_at_angle(0, subtomo_number) - previous_subtomo_number = subtomo_number - - self.progress["subtomo"] = subtomo_number - self.progress["projection"] = ii - self.progress["angle"] = angle - if self.golden_ratio_bunch_size > 0: - self.progress["subtomo_total_projections"] = self.golden_ratio_bunch_size - self.progress["subtomo_projection"] = ( - ii - (subtomo_number - 1) * self.golden_ratio_bunch_size - ) - else: - self.progress["subtomo_total_projections"] = 0 - self.progress["subtomo_projection"] = 0 - self.progress["total_projections"] = ( - self.golden_max_number_of_projections - if self.golden_max_number_of_projections > 0 - else 0 - ) - - self._tomo_scan_at_angle(angle, subtomo_number) - ii += 1 - if ( - self.golden_max_number_of_projections > 0 - and ii > self.golden_max_number_of_projections - ): - print( - f"Golden ratio tomography stopped automatically after the requested" - f" {self.golden_max_number_of_projections} projections." - ) - break - - elif self.tomo_type == 3: - # Equally spaced tomography with golden ratio starting angles over 360 degrees - self.progress["tomo_type"] = "Equally spaced, golden ratio starting angles" - previous_subtomo_number = -1 - ii = 0 if projection_number is None else projection_number - while True: - angle, subtomo_number = self._golden_equally_spaced( - ii, int(360 / self.tomo_angle_stepsize), maxangle=360, reverse=True - ) - if previous_subtomo_number != subtomo_number: - self._write_subtomo_to_scilog(subtomo_number) - if ( - subtomo_number % 2 == 1 - and ii > 10 - and self.golden_projections_at_0_deg_for_damage_estimation == 1 - ): - self._tomo_scan_at_angle(0, subtomo_number) - previous_subtomo_number = subtomo_number - - self.progress["subtomo"] = subtomo_number - self.progress["projection"] = ii - self.progress["angle"] = angle - self.progress["subtomo_total_projections"] = 360 / self.tomo_angle_stepsize - self.progress["subtomo_projection"] = ( - ii - (subtomo_number - 1) * self.progress["subtomo_total_projections"] - ) - self.progress["total_projections"] = ( - self.golden_max_number_of_projections - if self.golden_max_number_of_projections > 0 - else 0 - ) - - self._tomo_scan_at_angle(angle, subtomo_number) - ii += 1 - if ( - self.golden_max_number_of_projections > 0 - and ii > self.golden_max_number_of_projections - ): - print( - f"Golden ratio tomography stopped automatically after the requested" - f" {self.golden_max_number_of_projections} projections." - ) - break - else: - raise ValueError(f"Unknown tomo_type: {self.tomo_type}.") - - def tomo_parameters(self): - """print and update the tomo parameters""" - print("Current settings:") - print(f"Counting time")).add_tag(
- ["BEC", "tomo_parameters", f"dataset_id_{dataset_id}", "LamNI", self.sample_name]
- )
- self.client.logbook.send_logbook_message(msg)
-
-
-class MagLamNI(LamNI):
- def sub_tomo_scan(self, subtomo_number, start_angle=None):
- super().sub_tomo_scan(subtomo_number, start_angle)
- # self.rotate_slowly(0)
-
- def rotate_slowly(self, angle, step_size=20):
- current_angle = dev.lsamrot.read(cached=True)["value"]
- steps = int(np.ceil(np.abs(current_angle - angle) / step_size)) + 1
- for target_angle in np.linspace(current_angle, angle, steps, endpoint=True):
- umv(dev.lsamrot, target_angle)
- scans.lamni_move_to_scan_center(
- self.align.tomo_fovx_offset, self.align.tomo_fovy_offset, target_angle
- )
-
- def _at_each_angle(self, angle: float) -> None:
- if "lamni_at_each_angle" in builtins.__dict__:
- lamni_at_each_angle(self, angle)
- return
-
- self.tomo_scan_projection(angle)
- self.tomo_reconstruct()
-
- # # cm()
- # # umv(dev.ppth,15.1762) #11.567 keV
- # for ii in range(2):
- # self.tomo_scan_projection(angle)
- # self.tomo_reconstruct()
- # # cp()
- # # umv(dev.ppth,15.1827) #11.567 keV
- # for ii in range(2):
- # self.tomo_scan_projection(angle)
- # self.tomo_reconstruct()
-
-
-class DataDrivenLamNI(LamNI):
- def __init__(self, client):
- super().__init__(client)
- self.tomo_data = {}
-
- def tomo_scan(
- self,
- subtomo_start=1,
- start_index=None,
- fname="~/Data10/data_driven_config/datadriven_params.h5",
- ):
- """start a tomo scan"""
- bec = builtins.__dict__.get("bec")
- scans = builtins.__dict__.get("scans")
-
- fname = os.path.expanduser(fname)
-
- if not os.path.exists(fname):
- raise FileNotFoundError(f"Could not find datadriven params file in {fname}.")
- content = f"Loading tomo parameters from {fname}."
- logger.warning(content)
- tags = ["Data_driven_file", "BEC"]
- msg = bec.logbook.LogbookMessage()
- msg.add_text(content).add_tag(tags)
- self.client.logbook.send_logbook_message(msg)
- self._update_tomo_data_from_file(fname)
-
- self._current_special_angles = self.special_angles.copy()
-
- if subtomo_start == 1 and start_index is None:
- # pylint: disable=undefined-variable
- self.tomo_id = self.add_sample_database(
- self.sample_name,
- str(datetime.date.today()),
- bec.active_account.decode(),
- bec.queue.next_scan_number,
- "lamni",
- "test additional info",
- "BEC",
- )
- self.write_pdf_report()
- with scans.dataset_id_on_hold:
- self.sub_tomo_data_driven(start_index)
-
- def sub_tomo_scan(self):
- raise NotImplementedError(
- "Cannot run sub_tomo_scan with data-driven LamNI. Please use"
- " lamni.tomo_scan(subtomo_start=