Bernina/eco
0
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

fixes

Browse Source
This commit is contained in:
gac bernina
2026-06-26 15:22:07 +02:00
parent 32f33f9f9c
commit 41fc1a33db
6 changed files with 288 additions and 54 deletions
Download Patch File Download Diff File Expand all files Collapse all files
eco/bernina/bernina.py
+13 -13
View File
@@ -1318,13 +1318,13 @@ namespace.append_obj(
# module_name="eco.devices_general.cameras_swissfel",
# )
# namespace.append_obj(
# "OxygenSensor",
# "SARES20-CWAG-GPS01:ADC08",
# lazy=True,
# name="oxygen_sensor",
# module_name="eco.devices_general.sensors_ai",
# )
namespace.append_obj(
"OxygenSensor",
"SARES20-CWAG-GPS01:ADC08",
lazy=True,
name="oxygen_sensor",
module_name="eco.devices_general.sensors_ai",
)
# namespace.append_obj(
# "CameraBasler",
@@ -1400,12 +1400,12 @@ namespace.append_obj(
)
# namespace.append_obj(
# "IncouplingCleanBernina",
# lazy=True,
# name="clic",
# module_name="eco.loptics.bernina_laser",
# )
namespace.append_obj(
"IncouplingCleanBernina",
lazy=True,
name="clic",
module_name="eco.loptics.bernina_laser",
)
# namespace.append_obj(
# "MidIR",
# lazy=True,
eco/devices_general/motors.py
+195 -4
View File
@@ -10,6 +10,7 @@ from ..aliases import Alias
from ..elements.adjustable import (
AdjustableError,
AdjustableFS,
AdjustableGetSet,
AdjustableMemory,
spec_convenience,
ValueInRange,
@@ -883,7 +884,7 @@ class SmarActOpenLoopRecord(Assembly):
self._append(
AdjustableFS,
f"/sf/bernina/code/gac-bernina/eco_cnf_bernina/reference_values/smaract_openloop_{name}_position.json",
name="position",
name="_position",
default_value=0,
is_setting=True,
is_display=True,
@@ -907,7 +908,7 @@ class SmarActOpenLoopRecord(Assembly):
def move(self, value, check=True, wait=False):
value = int(value) * self.direction()
pos = int(self.position() * self.direction())
pos = int(self._position() * self.direction())
target_rel = value - pos
if check:
if self.limit_low:
@@ -924,7 +925,7 @@ class SmarActOpenLoopRecord(Assembly):
f":MST{self.channel-1},{target_rel},{int(self.voltage()*40.95)},{int(self.frequency())}"
)
if res[-1] == "0":
self.position.mv(value)
self._position.mv(value)
else:
raise Exception(
f"Motion of SmarAct Motor {self.name} failed with error code {res}"
@@ -938,7 +939,197 @@ class SmarActOpenLoopRecord(Assembly):
self.limit_high.set_target_value(limit_high)
def get_current_value(self):
return self.position.get_current_value()
return self._position.get_current_value()
def set_current_value(self, value):
self._position(value)
def set_target_value(self, value, hold=False, check=True, **kwargs):
return Changer(
target=value,
parent=self,
changer=self.move,
hold=hold,
stopper=self.stop,
)
# return string with motor value as variable representation
def __str__(self):
# """ return short info for the current motor"""
s = f"{self.name}"
# s += f"\t@ {colorama.Style.BRIGHT}{self.get_current_value():1.6g}{colorama.Style.RESET_ALL} stat: {self.status_flag().name}"
s += f"\t@ {colorama.Style.BRIGHT}{self.get_current_value():1.6g}{colorama.Style.RESET_ALL}"
# # s += "\tuser limits (low,high) : {:1.6g},{:1.6g}\n".format(*self.get_limits())
s += f"\n{colorama.Style.DIM}low limit {colorama.Style.RESET_ALL}"
s += ValueInRange(*self.get_limits()).get_str(self.get_current_value())
s += f" {colorama.Style.DIM}high limit{colorama.Style.RESET_ALL}"
# # s += "\tuser limits (low,high) : {:1.6g},{1.6g}".format(self.get_limits())
return s
def __repr__(self):
print(str(self))
return object.__repr__(self)
@spec_convenience
# @get_from_archive
@value_property
@tweak_option
class SmarActOpenLoopRecordMCS2(Assembly):
def __init__(self, pvname=None, channel=None, name=None):
super().__init__(name=name)
self.pvname = pvname
self.channel = channel
self._append(
AdjustablePv,
self.pvname.split(":")[0] + f":MOT_{self.channel}.DESC",
name="description",
is_setting=False,
is_display=False,
)
self._append(
AdjustablePv,
self.pvname + ".AOUT",
name="_com_set",
is_setting=False,
is_display=False,
)
self._append(
AdjustablePv,
self.pvname + ".TINP",
name="_com_get",
is_setting=False,
is_display=False,
)
self._append(
AdjustableFS,
f"/sf/bernina/code/gac-bernina/eco_cnf_bernina/reference_values/smaract_openloop_{name}_limit_high.json",
default_value=-1e6,
name="limit_high",
is_setting=True,
)
self._append(
AdjustableFS,
f"/sf/bernina/code/gac-bernina/eco_cnf_bernina/reference_values/smaract_openloop_{name}_limit_low.json",
default_value=1e6,
name="limit_low",
is_setting=True,
)
self._append(
AdjustableFS,
f"/sf/bernina/code/gac-bernina/eco_cnf_bernina/reference_values/smaract_openloop_{name}_voltage.json",
name="voltage",
default_value=25,
is_setting=True,
is_display=True,
)
self._append(
AdjustableFS,
f"/sf/bernina/code/gac-bernina/eco_cnf_bernina/reference_values/smaract_openloop_{name}_frequency.json",
name="frequency",
default_value=250,
is_setting=True,
is_display=True,
)
self._append(
AdjustableFS,
f"/sf/bernina/code/gac-bernina/eco_cnf_bernina/reference_values/smaract_openloop_{name}_position.json",
name="_position",
default_value=0,
is_setting=True,
is_display=True,
)
self._append(
AdjustableFS,
f"/sf/bernina/code/gac-bernina/eco_cnf_bernina/reference_values/smaract_openloop_{name}_direction.json",
name="direction",
default_value=1,
is_setting=True,
is_display=True,
)
def get_motion_mode():
return int(self.eval_command(f":CHAN{self.channel-1}:MMOD?"))
def set_motion_mode(value):
self.eval_command(f":CHAN{self.channel-1}:MMOD {value}")
self._append(
AdjustableGetSet,
get_motion_mode,
set_motion_mode,
set_returns_changer=False,
precision=0,
check_interval=None,
cache_get_seconds=None,
unit=None,
name="motion_mode",
)
if not self.motion_mode() == 4:
self.motion_mode.mv_elog(
4,
f"Setting SmarAct Channel {self.channel} on {self.pvname} to Open Loop (4). To go back to closed loop, set it to (0)",
)
def eval_command(self, cmd):
self._com_set(cmd)
sleep(0.2)
return self._com_get()
def stop(self):
self._com_set(f":STOP{self.channel-1}")
def move(self, value, check=True, wait=True):
value = int(value) * self.direction()
pos = int(self._position() * self.direction())
target_rel = value - pos
if check:
if self.limit_low:
if value < self.limit_low():
raise Exception(
f"Target value of {self.name} is smaller than limit value!"
)
if self.limit_high:
if self.limit_high() < value:
raise Exception(
f"Target value of {self.name} is higher than limit value!"
)
self.eval_command(f":MOVE{self.channel-1} {target_rel}")
if wait:
while not self.get_move_done():
sleep(0.1)
self._position.mv(value)
def get_limits(self):
return (self.limit_low(), self.limit_high())
def get_move_done(self):
stat = self.get_status()
if int(stat[-1]) == 9:
return False
else:
return True
def get_status(self):
stat = self.eval_command(f":CHAN{self.channel-1}:STAT?")
if len(stat) == 0:
for n in range(10):
stat = self.eval_command(f":CHAN{self.channel-1}:STAT?")
if len(stat) > 0:
break
return stat
def set_limits(self, limit_low, limit_high):
self.limit_low.set_target_value(limit_low)
self.limit_high.set_target_value(limit_high)
def get_current_value(self):
return self._position.get_current_value()
def set_current_value(self, value):
self._position(value)
def set_target_value(self, value, hold=False, check=True, **kwargs):
return Changer(
eco/devices_general/sensors_ai.py
+14 -16
View File
@@ -7,28 +7,26 @@ class OxygenSensor(AnalogInput):
super().__init__(pvname, name=name)
self.unit.set_target_value("%")
def set_no_oxygen(self, val_curr=None):
if not val_curr:
val_curr = self.raw.get_current_value()
def set_no_oxygen(self, raw_val=None):
if not raw_val:
raw_val = self.raw.get_current_value()
slo = self.linear_calibration_slope.get_current_value()
off = self.linear_calibration_offset.get_current_value()
off_n = -slo * val_curr
af = (100 - off) / slo
# slo_new = slo*((100-val_curr)/(100-off))
slo_n = 100 / af
r0 = raw_val
r100 = (100 - off) / slo
slo_n = 100 / (r100 - r0)
off_n = -slo_n * r0
self.linear_calibration_offset(off_n)
self.linear_calibration_slope(slo_n)
def set_full_oxygen(self, val_curr=None):
if not val_curr:
val_curr = self.raw.get_current_value()
af = val_curr
def set_full_oxygen(self, raw_val=None):
if not raw_val:
raw_val = self.raw.get_current_value()
slo = self.linear_calibration_slope.get_current_value()
off = self.linear_calibration_offset.get_current_value()
az = -off / slo
slo_n = 100 / (af - az)
off_n = -slo_n * az
r0 = -(off / slo)
r100 = raw_val
slo_n = 100 / (r100 - r0)
off_n = -slo_n * r0
self.linear_calibration_offset(off_n)
self.linear_calibration_slope(slo_n)
eco/endstations/bernina_rixs.py
+40 -19
View File
@@ -303,7 +303,6 @@ class DetectorStages(Assembly):
f"Initialization of epics motor {name}: {pvname}:{pvmot} failed, replaced by dummy!"
)
class RIXS(Assembly):
def __init__(
self,
@@ -379,13 +378,10 @@ class RIXS(Assembly):
pvname=pvname,
)
# self.append_analyzer(
# pos=2,
# analyzer="Si844",
# hkl=(8, 4, 4),
# name="ana2_laser",
# pvname=pvname,
# )
self.append_multi_analyzer_motion(
analyzer_list=[self.ana_right, self.ana_center, self.ana_left],
name="energy_all_analyzers"
)
self._append(
Jungfrau,
@@ -426,20 +422,45 @@ class RIXS(Assembly):
is_display="recursive",
)
def append_multi_analyzer_motion(self, analyzer_list = [], name=None):
detector_motors = [self.det.t_hor, self.det.t_ver, self.det.rot]
analyzer_motors = [motor for ana in analyzer_list for motor in [ana.__dict__["om"], ana.__dict__["t_hor"]]]
def motor_pos_from_energy_multy_analyzers(energy):
motor_positions = np.array([ana.motor_pos_from_energy(energy) for ana in analyzer_list])
detector_positions = motor_positions[:,2:]
if np.sum(np.std(detector_positions, axis=0)) > 0.1:
raise (f"Conflicting detector target positions in multi analyzer motion: {[[ana.name, det_pos] for ana, det_pos in zip(analyzer_list, detector_positions)]}, check analyzer configuration")
else:
detector_position = np.mean(detector_positions, axis=0)
analyzer_positions = np.concat(motor_positions[:,0:2])
return np.concat([detector_position, analyzer_positions])
def energy_from_motor_pos_multy_analyzers(*args, **kwargs):
energies = np.array([[ana.energy()] for ana in analyzer_list])
if None in energies:
return None
elif np.std(energies) > 0.1:
return None
else:
return np.mean(energies)
self._append(
AdjustableVirtual,
detector_motors+analyzer_motors,
energy_from_motor_pos_multy_analyzers,
motor_pos_from_energy_multy_analyzers,
is_setting=False,
is_display=True,
name=name,
unit="eV",
)
def gui(self, guiType="xdm"):
"""Adjustable convention"""
cmd = ["caqtdm", "-macro"]
cmd += ["P=SARES22-RIXS", "ESB_RIXS_motors.ui"]
return self._run_cmd(" ".join(cmd))
# def get_adjustable_positions_str(self):
# ostr = "*****GPS motor positions******\n"
# for tkey, item in self.__dict__.items():
# if hasattr(item, "get_current_value"):
# pos = item.get_current_value()
# ostr += " " + tkey.ljust(17) + " : % 14g\n" % pos
# return ostr
# def __repr__(self):
# return self.get_adjustable_positions_str()
eco/endstations/bernina_sample_environments.py
+2 -2
View File
@@ -13,7 +13,7 @@ from ..devices_general.motors import (
MotorRecord,
SmaractRecord,
ThorlabsPiezoRecord,
SmarActOpenLoopRecord,
SmarActOpenLoopRecordMCS2,
)
from ..epics.adjustable import AdjustablePv
import numpy as np
@@ -1657,7 +1657,7 @@ class GrazingIncidenceLowTemperatureChamber(Assembly):
for name, config in self.motor_configuration_openloop.items():
self._append(
SmarActOpenLoopRecord,
SmarActOpenLoopRecordMCS2,
pvname=config["id"],
name=name,
channel=config["channel"],
eco/loptics/bernina_laser.py
+24
View File
@@ -1350,6 +1350,30 @@ class LaserBernina(Assembly):
name="delay_twin",
is_setting=True,
)
self._append(
SmaractRecord,
"SLAAR21-LMTS-SMAR1:MOT_5",
name="delaystage_ftir",
is_setting=True,
)
self._append(
SmaractRecord,
"SLAAR21-LMTS-SMAR1:MOT_4",
name="dfg_rot",
is_setting=True,
)
self._append(
SmaractRecord,
"SLAAR21-LMTS-SMAR1:MOT_3",
name="dfg_pos",
is_setting=True,
)
self._append(
DelayTime,
self.delaystage_ftir,
name="delay_ftir",
is_setting=True,
)
self._append(
DelayTime,