bec/csaxs_bec
0
0
Fork 2
Code Issues 17 Pull Requests 2 Actions Packages Projects Releases Wiki Activity

feat(xbpms): add diag signal and normalize diff signals

Browse Source
This commit is contained in:
wakonig_k
2025-07-18 12:00:11 +02:00
parent a4c1b89395
commit e3077f6d2f
1 changed files with 44 additions and 14 deletions
Download Patch File Download Diff File Expand all files Collapse all files
csaxs_bec/devices/epics/xbpms.py
+44 -14
View File
@@ -1,5 +1,6 @@
from ophyd import Signal, Device, EpicsSignalRO
from ophyd import Component as Cpt from ophyd import Component as Cpt
from ophyd import Device, EpicsSignalRO, Signal
class SumSignal(Signal): class SumSignal(Signal):
def __init__(self, *args, **kwargs): def __init__(self, *args, **kwargs):
@@ -18,15 +19,23 @@ class SumSignal(Signal):
return val1 + val2 + val3 + val4 return val1 + val2 + val3 + val4
def describe(self): def describe(self):
source = [
self.parent.current1.pvname,
self.parent.current2.pvname,
self.parent.current3.pvname,
self.parent.current4.pvname,
]
source = " / ".join(source)
desc = { desc = {
"shape": [], "shape": [],
"dtype": "number", "dtype": "number",
"source": "PV: {} / {}".format(self.parent.current1.pvname, self.parent.current1.pvname), "source": f"PV: {source}",
"units": "", "units": "",
"precision": self.parent.current1.precision, "precision": self.parent.current1.precision,
} }
return desc return desc
class DiffXYSignal(Signal): class DiffXYSignal(Signal):
def __init__(self, sum1, sum2, *args, **kwargs): def __init__(self, sum1, sum2, *args, **kwargs):
self.sum1 = sum1 self.sum1 = sum1
@@ -46,33 +55,54 @@ class DiffXYSignal(Signal):
for signal in self.sum2: for signal in self.sum2:
summed_2 += getattr(self.parent, signal).get() summed_2 += getattr(self.parent, signal).get()
return summed_1 - summed_2 _sum = summed_1 + summed_2
if _sum == 0:
return 0.0
return (summed_1 - summed_2) / _sum
def describe(self): def describe(self):
source = [getattr(self.parent, signal).pvname for signal in self.sum1 + self.sum2]
source = " / ".join(source)
desc = { desc = {
"shape": [], "shape": [],
"dtype": "number", "dtype": "number",
"source": "PV: {} / {}".format(self.parent.current1.pvname, self.parent.current1.pvname), "source": f"PV: {source}",
"units": "", "units": "",
"precision": self.parent.current1.precision, "precision": self.parent.current1.precision,
} }
return desc return desc
class BPMDevice(Device): class BPMDevice(Device):
current1 = Cpt(EpicsSignalRO, ":Current1:MeanValue_RBV", kind="normal") current1 = Cpt(EpicsSignalRO, ":Current1:MeanValue_RBV", kind="normal", doc="Current 1")
current2 = Cpt(EpicsSignalRO, ":Current2:MeanValue_RBV", kind="normal") current2 = Cpt(EpicsSignalRO, ":Current2:MeanValue_RBV", kind="normal", doc="Current 2")
current3 = Cpt(EpicsSignalRO, ":Current3:MeanValue_RBV", kind="normal") current3 = Cpt(EpicsSignalRO, ":Current3:MeanValue_RBV", kind="normal", doc="Current 3")
current4 = Cpt(EpicsSignalRO, ":Current4:MeanValue_RBV", kind="normal") current4 = Cpt(EpicsSignalRO, ":Current4:MeanValue_RBV", kind="normal", doc="Current 4")
sum = Cpt(SumSignal, kind="hinted") sum = Cpt(SumSignal, kind="hinted", doc="Sum of all currents")
x = Cpt(DiffXYSignal, sum1=["current1", "current2"], sum2=["current3", "current4"]) x = Cpt(
y = Cpt(DiffXYSignal, sum1=["current1", "current3"], sum2=["current2", "current4"]) DiffXYSignal,
sum1=["current1", "current2"],
sum2=["current3", "current4"],
doc="X difference signal",
)
y = Cpt(
DiffXYSignal,
sum1=["current1", "current3"],
sum2=["current2", "current4"],
doc="Y difference signal",
)
diag = Cpt(
DiffXYSignal,
sum1=["current1", "current4"],
sum2=["current2", "current3"],
doc="Diagonal difference signal",
)
def __init__(self, prefix="", *args, **kwargs): def __init__(self, prefix="", *args, **kwargs):
super().__init__(*args, prefix=prefix, **kwargs) super().__init__(*args, prefix=prefix, **kwargs)
if __name__ == "__main__": if __name__ == "__main__":
dev = BPMDevice(name="bpm", prefix='X12SA-FE-XBPM1') dev = BPMDevice(name="bpm", prefix="X12SA-FE-XBPM1")
dev.wait_for_connection() dev.wait_for_connection()
print(dev.read()) print(dev.read())