From 2ab449fe17c1f0688600a832aafb019771c49686 Mon Sep 17 00:00:00 2001 From: Vonka Jakub Date: Fri, 27 Jan 2023 16:15:47 +0100 Subject: [PATCH] Pulsed magnet pulse method added - temporarily in the cristallina.py --- cristallina.py | 61 ++++++++++++++++++++++++++++++++++++++++++-------- 1 file changed, 52 insertions(+), 9 deletions(-) diff --git a/cristallina.py b/cristallina.py index c3b0559..5207f26 100644 --- a/cristallina.py +++ b/cristallina.py @@ -1,16 +1,18 @@ #!/usr/bin/env python - - import logging - +from tracemalloc import start logger = logging.getLogger("slic") logger.setLevel(logging.INFO) logger.info("Loading started.") from time import sleep from datetime import datetime + import numpy as np +import os +os.chdir('/sf/cristallina/applications/slic/cristallina') + # from tqdm import trange from epics import PV @@ -28,12 +30,15 @@ from slic.devices.xoptics.pulsepicker import PulsePicker from slic.utils import devices, Marker, as_shortcut from slic.utils import Channels, Config, Elog, Screenshot, PV from slic.core.acquisition.fakeacquisition import FakeAcquisition +from slic.devices.timing.events import CTASequencer from channels import detectors, channels, pvs from spreadsheet import overview from channels_minimal import detectors_min, channels_min, pvs_min from pp_shutter import PP_Shutter + + # DEVICES ## example devices and adjustables from cool_motor import MyNewCoolThing @@ -48,6 +53,7 @@ from knife_edge import KnifeEdge attenuator = Attenuator("SAROP31-OATA150", description="Cristallina attenuator OATA150") front_end_attenuator = Attenuator("SARFE10-OATT053", description="Front end attenuator OATT053") +cta = CTASequencer("SAR-CCTA-ESC") def test_attenuator(): tfundamental = attenuator.get_transmission() @@ -90,22 +96,22 @@ kbVer = KBVer( instrument = "cristallina" # pgroup = "p19739" # commissioning March 2022 -- July 2022 -pgroup = "p20443" # commissioning Wavefront Sensor August 2022 (active) +# pgroup = "p20443" # commissioning Wavefront Sensor August 2022 (active) # pgroup = "p20558" # SwissMX commissioning 3 # pgroup = "p20557" # CrQ PMS commisioning 1 -# pgroup = "p20509" # CrQ commissoing DilSc1 (active) +# pgroup = "p20509" # CrQ commissoing DilSc1 # pgroup = "p20519" # beamline commissioning 2 - +pgroup = "p20841" # CrQ PMS commisioning 2 (Jan 2023) # pgroup = "p19150" # Scratch # pgroup = "p19152" # Scratch daq = SFAcquisition( instrument, pgroup, - default_channels=channels_min , - default_pvs=pvs_min, - default_detectors=detectors_min, + default_channels=channels , + default_pvs=pvs, + default_detectors=detectors, rate_multiplicator=1, ) @@ -122,3 +128,40 @@ gui = GUI(scan, show_goto=True, show_spec=True) logger.info(f"Running at {instrument} with pgroup {pgroup}.") logger.info("Loading finished.") + + +# For PMS pulsing. Temporarily placed here, should be a method of the pulsed magnet device once created. +import sys +sys.path.append('/sf/jungfrau/applications/daq_client') +import daq_client as sf_daq_client + +CTA_sequence_start_PID = PV("SAR-CCTA-ESC:seq0Ctrl-StartedAt-O") + +# Number of pulses to ask from the SFDAQ. This will depend on the CTA sequence. +npulses=10 + +def pulse(run_comment=None): + # Start the CTA sequence + cta.cta_client.start() + + # Wait until done + sleep(3) + + # Ask for the first pid in the sequence + start_PID=int(CTA_sequence_start_PID.get()) + + # Data retrieval does not allow empty list, so creating a new variable + if detectors == [] or detectors == None: + used_detectors = {} + else: + used_detectors = detectors + + # Ask the DAQ client to download the data + run_number = sf_daq_client.retrieve_data_from_buffer(pgroup=pgroup, user_tag=run_comment, + camera_channels=[], bsread_channels=channels, epics_channels=pvs, + detectors=used_detectors, + start_pulseid=start_PID, stop_pulseid=start_PID+npulses, + rate_multiplicator=1, + ) + print('\n') + print(f' Magnet pulse done.\n Run {run_number} created. \n First PID in the CTA sequence: {start_PID}\n')