sls/hla_framework_bd
0
1
Fork 0
Code Issues 28 Pull Requests 2 Actions Packages Projects 1 Releases Wiki Activity

feat: migrate taupct python service

Browse Source
This commit is contained in:
Benjamin Labrecque
2026-07-16 16:17:17 +02:00
parent 07f68e6634
commit 70217d68ef
4 changed files with 124 additions and 40 deletions
Download Patch File Download Diff File Expand all files Collapse all files
services/taupct/current/app/pyproject.toml
+1
View File
@@ -5,6 +5,7 @@ description = "AGEBD-TAUPCT Service"
requires-python = "==3.10.*"
dependencies = [
"agebd",
"numpy>=2.2.6",
]
[dependency-groups]
services/taupct/current/app/src/agebd_taupct/main.py
+1 -1
View File
@@ -5,9 +5,9 @@ from agebd.runner import CallbackRunner
from agebd.utils import init_logging
from agebd_taupct import PVs, Service
# TODO: remove: cicd test 10
def main(
log_level: LogLevel = typer.Option(
LogLevel.INFO,
services/taupct/current/app/src/agebd_taupct/service.py
+117 -38
View File
@@ -1,12 +1,12 @@
import time
from time import sleep
import numpy as np
from epics import dbr
from agebd.pv import LocalPVLink
from agebd.pv import get_pv_class
from agebd.service.base import BaseService
from agebd.service.pvs import BasePVs
from agebd.utils import printgetversion
from agebd.pv import get_pv_class
__version__ = printgetversion(__file__)
@@ -14,36 +14,29 @@ PV = get_pv_class()
class PVs(BasePVs):
# Option 1:
## Service specific PVs from dedicated IOC
my_pv1 = PV("AGEBD-TAUPCT:AO") # TODO: is this a good PV to use?
NsamplesTAU = PV("AGEBD-TAUPCT:CONTROL-NSAMPLES-FIT-TAU.VAL")
NsamplesROC = PV("AGEBD-TAUPCT:CONTROL-NSAMPLES-FIT-ROC.VAL")
MinCurr = PV("AGEBD-TAUPCT:CONTROL-MIN-CURR.VAL")
# Output PVs
inj_tau = PV("AGEBD-TAUPCT:TAU.VAL")
inj_tau_err = PV("AGEBD-TAUPCT:TAU-ERR.VAL")
inj_rate = PV("AGEBD-TAUPCT:INJRATE.VAL")
inj_rate_err = PV("AGEBD-TAUPCT:INJRATE-ERR.VAL")
## Service specific PVs from other IOCs
# ...
inj_active = PV("AGEBD-PARAMS:INJECTION-ACTIVE.VAL")
dump_valve = PV("ARS01-VVPG-0210:PLC_STATUS.VAL")
def __init__(self, service_name: str, pv_factory=PV):
super().__init__(service_name, pv_factory)
PV = pv_factory
# TODO: can we do this cleaner? Maybe mechanism for
# settings initial values in dev mode (e.g. ini file)
if isinstance(self.onoff, LocalPVLink):
self.onoff.value = True
# Option 2:
## Service specific PVs from dedicated IOC
self.my_pv2 = PV(
"AGEBD-TAUPCT:BO"
) # TODO: is this a good PV to use?
## Service specific PVs from other IOCs
# ...
# TODO: class attribute names usually lowercase
# usually it is advisable to trigger the mainloop of a service on changes of certain PVs:
self.CallbackPV = PV(
"AGEBD-TAUPCT:CALLBACK", auto_monitor=dbr.DBE_VALUE
) # TODO: ok?
self.CallbackPV = PV("ARS07-CECL-DPCT0:CURR.VAL", auto_monitor=dbr.DBE_VALUE)
# CallbackPV = PV('ARS07-DPCT-0000:AVG-CURR.VAL', auto_monitor = dbr.DBE_VALUE)
# ARS07-DPCT-0000:CURR (5 Hz, i.e., dt = 0.2 s)
# ARS07-DPCT-0000:AVG-CURR (5 Hz, i.e., dt = 0.2 s)
class Service(BaseService[PVs]):
@@ -56,17 +49,105 @@ class Service(BaseService[PVs]):
):
super().__init__(name, pvs, version, sleep_interval)
# TODO: What is this for?
self.path = "sls/bd/bin"
self.reset()
def reset(self):
pvs = self.pvs
self.IterSinceReset = 0
self.NsamplesTAU = int(pvs.NsamplesTAU.get())
self.NsamplesROC = int(pvs.NsamplesROC.get())
NsamplesMAX = int(np.nanmax([self.NsamplesTAU, self.NsamplesROC]))
self.MinCurr = pvs.MinCurr.get()
self.t = np.empty(NsamplesMAX)
self.I = np.empty(NsamplesMAX)
self.inj_active = pvs.inj_active.get()
def update(self):
pvs = self.pvs
# when callback is fired, this code will be executed
if self.CallbackFired:
#######################################################################################
#######################################################################################
# here goes all the code that your service
# is supposed to be running when the callback is fired
time.sleep(1)
conditions_cb_invalid = (
not self.inj_active == pvs.inj_active.get() # <-- injection status changed
or self.CallbackValue < self.MinCurr # <-- stored beam
or not 0 <= self.CallbackValue <= 500 # <-- measurement realistic
)
# set values to zero and wait without beam
if not pvs.dump_valve.get() == 1:
pvs.inj_tau.put(np.nan)
pvs.inj_tau_err.put(0.0)
pvs.inj_rate.put(0.0)
pvs.inj_rate_err.put(0.0)
sleep(0.5)
elif conditions_cb_invalid:
self.reset()
print("Callback Invalid --> Reset ({:})".format(self.CallbackTimestamp))
else:
self.t[:-1] = self.t[1:]
self.I[:-1] = self.I[1:]
self.t[-1] = self.CallbackTimestamp
self.I[-1] = self.CallbackValue
self.IterSinceReset += 1
if (
self.IterSinceReset > self.NsamplesROC
and self.IterSinceReset > self.NsamplesTAU
):
t = self.t - self.t[0]
p, pcov = np.polyfit(t, self.I, 1, cov=True)
perr = np.sqrt(np.diag(pcov))
injrate = p[0] * 60
injrate_err = perr[0] * 60
pvs.inj_rate.put(injrate)
pvs.inj_rate_err.put(injrate_err)
# injection rate in mA/hour to minimize risk of log(x) with x < 0:
injrate = injrate * 60
injrate_err = injrate_err * 60
# if injrate/p[1] < 1:
if not self.inj_active and self.IterSinceReset > self.NsamplesTAU:
tau = 1 / np.log(1 - injrate / p[1]) # (-dt in hours) / ln(1 - dI/I0)
tau_err = injrate_err * (-1) / (
(injrate - p[1]) * np.log(1 - injrate / p[1]) ** 2
) + perr[1] * (injrate_err) / (
p[1] * (injrate - p[1]) * np.log(1 - injrate / p[1]) ** 2
)
pvs.inj_tau.put(tau)
pvs.inj_tau_err.put(tau_err)
elif self.IterSinceReset > self.NsamplesROC:
t = self.t[-self.NsamplesROC :]
I = self.I[-self.NsamplesROC :]
t = t - t[0]
p, pcov = np.polyfit(t, I, 1, cov=True)
perr = np.sqrt(np.diag(pcov))
injrate = p[0] * 60
injrate_err = perr[0] * 60
pvs.inj_rate.put(injrate)
pvs.inj_rate_err.put(injrate_err)
# injection rate in mA/hour to minimize risk of log(x) with x < 0:
injrate = injrate * 60
injrate_err = injrate_err * 60
#######################################################################################
#######################################################################################
@@ -77,17 +158,15 @@ class Service(BaseService[PVs]):
#######################################################################################
# here goes all the code that your service
# is supposed to be running in any case, e.g.,
# update some pvs/vals
my_pv1_random_val = str(time.clock_gettime_ns(0))[-1]
self.pvs.my_pv1.put(my_pv1_random_val)
# self.pvs.my_pv2.put("MY_PV2 value")
# get some pvs/vals
self.val1 = self.pvs.my_pv1.get()
self.val2 = self.pvs.my_pv2.get()
conditions_reset = (
not self.NsamplesROC == pvs.NsamplesROC.get()
or not self.NsamplesTAU == pvs.NsamplesTAU.get()
or not self.MinCurr == pvs.MinCurr.get()
)
if conditions_reset:
self.reset()
# maybe sleep to limit the update loop execution rate
time.sleep(1)
# sleep(1)
#######################################################################################
#######################################################################################
services/taupct/current/app/uv.lock
Generated
+5 -1
View File
@@ -30,6 +30,7 @@ version = "0.1.0"
source = { editable = "." }
dependencies = [
{ name = "agebd" },
{ name = "numpy" },
]
[package.dev-dependencies]
@@ -40,7 +41,10 @@ dev = [
]
[package.metadata]
requires-dist = [{ name = "agebd", editable = "../../../../packages/agebd" }]
requires-dist = [
{ name = "agebd", editable = "../../../../packages/agebd" },
{ name = "numpy", specifier = ">=2.2.6" },
]
[package.metadata.requires-dev]
dev = [