slic/cristallina
0
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Mobile JJ slits added

Browse Source
This commit is contained in:
vonka_j
2023-05-09 18:22:18 +02:00
parent 24a2a465c5
commit 9c59d52d5c
2 changed files with 56 additions and 51 deletions
Download Patch File Download Diff File Expand all files Collapse all files

87
cristallina.py
View File

@ -1,22 +1,35 @@
#!/usr/bin/env python
import sys
import os
os.chdir('/sf/cristallina/applications/slic/cristallina')
import re
# setup logging
import logging
from tracemalloc import start
logger = logging.getLogger("slic")
logger.setLevel(logging.INFO)
logger.info("Loading started.")
# create file handler which logs
fh = logging.FileHandler('/sf/cristallina/applications/slic/cristallina/log/cristallina.log')
fh.setLevel(logging.DEBUG)
logger.addHandler(fh)
try:
fh = logging.FileHandler('/sf/cristallina/applications/slic/cristallina/log/cristallina.log')
fh.setLevel(logging.DEBUG)
logger.addHandler(fh)
except PermissionError as e:
logger.warning("Cannot write log file.")
logger.warning(e)
from time import sleep
import sys
import os
os.chdir('/sf/cristallina/applications/slic/cristallina')
from tracemalloc import start
from epics import PV
from stand_client import Client
stand_client = Client(host='saresc-vcons-02.psi.ch', port=9090)
from alignment_laser import AlignmentLaser
from slic.gui import GUI
from slic.core.adjustable import Adjustable, PVAdjustable, DummyAdjustable
@ -39,7 +52,6 @@ from channels_minimal import detectors_min, channels_min, pvs_min
from pp_shutter import PP_Shutter
# DEVICES
dummy = DummyAdjustable(units="au")
@ -73,11 +85,12 @@ def test_attenuator():
try:
assert tfundamental > 0
except TypeError:
print("No transmission value reported from {attenuator.ID}")
#test_attenuator()
logger.warning("No transmission value reported from {attenuator.ID}")
test_attenuator()
## Undulator
# Undulators
import undulator
undulators = undulator.Undulators()
@ -85,7 +98,7 @@ undulators = undulator.Undulators()
shutter = PP_Shutter("SARES30-LTIM01-EVR0:RearUniv0-Ena-SP", name="Cristallina pulse picker shutter") # Shutter buttton when using the pulse picker
pulsepicker = PulsePicker("SAROP31-OPPI151","SARES30-LTIM01-EVR0:Pul3", name="Cristallina X-ray pulse picker OPPI151")
# Alignmnet laser
# Alignment laser
alaser = AlignmentLaser("SAROP31-OLAS147:MOTOR_1", name="Cristallina alignment laser OLAS147")
## Slits
@ -94,6 +107,7 @@ from slic.devices.xoptics import slits
## Smaract & attocube stages
from smaract_device_def import smaract_Juraj, smaract_mini_XYZ
from attocube_device_def import attocube
from jj_device_def import jjslits
# KBs
from slic.devices.xoptics.kb import KBHor,KBVer
@ -139,50 +153,21 @@ daq = SFAcquisition(
# scan = Scanner(scan_info_dir=f"/sf/{instrument}/data/{pgroup}/res/scan_info", default_acquisitions=[daq], condition=None)
# Run the scan only when gas monitor value larger than 10uJ (and smaller than 2000uJ):
check_intensity_gas_monitor = PVCondition("SARFE10-PBPG050:PHOTON-ENERGY-PER-PULSE-US", vmin=0.1, vmax=2000, wait_time=0.5, required_fraction=0.8) # required fraction defines ammount of data recorded to save the step and move on to the next one
# required fraction defines ammount of data recorded to save the step and move on to the next one
check_intensity_gas_monitor = PVCondition("SARFE10-PBPG050:PHOTON-ENERGY-PER-PULSE-US", vmin=10, vmax=2000, wait_time=0.5, required_fraction=0.8)
scan = Scanner(default_acquisitions=[daq],condition = check_intensity_gas_monitor)
scan = Scanner(default_acquisitions=[daq], condition = check_intensity_gas_monitor)
gui = GUI(scan, show_goto=True, show_spec=True)
logger.info(f"Running at {instrument} with pgroup {pgroup}.")
logger.info("Loading finished.")
def simple_acquisition(filename):
task = daq.acquire(filename)
fname = task.filenames[0]
pattern = r"run([0-9]{4})"
match = re.search(pattern, fname)
run_number = int(match.groups()[0])
# 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
stand_client.add_row(run_number=run_number)
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 = 100
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
used_detectors = {"JF16T03V01":{}}
# TODO: change to SFAcquisition.retrieve(...) from slic
# 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=bs_channels, epics_channels=pvs,
detectors=used_detectors,
start_pulseid=start_PID-10, stop_pulseid=start_PID+210,
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')

20
jj_device_def.py Normal file
View File

@ -0,0 +1,20 @@
from slic.core.device import Device, SimpleDevice
from slic.devices.general.motor import Motor
class SlitUnitCenterWidthJJ(Device):
def __init__(self, ID, motor_naming="MOTOR",**kwargs):
super().__init__(ID, **kwargs)
self.x = SimpleDevice(ID + "-X",
center = Motor(ID + ":"+motor_naming+"_X"),
width = Motor(ID + ":"+motor_naming+"_W")
)
self.y = SimpleDevice(ID + "-Y",
center = Motor(ID + ":"+motor_naming+"_Y"),
width = Motor(ID + ":"+motor_naming+"_H")
)
jjslits = SlitUnitCenterWidthJJ("SARES30-MOBI2", motor_naming="MOT")