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

feat: migrate topuptool python service -- best effort

Browse Source
This commit is contained in:
Benjamin Labrecque
2026-07-16 16:41:21 +02:00
parent 17078d934b
commit f07e2e8a0a
6 changed files with 440 additions and 60 deletions
Download Patch File Download Diff File Expand all files Collapse all files
services/topuptool/current/app/pyproject.toml
+1
View File
@@ -5,6 +5,7 @@ description = "AGEBD-TOPUPTOOL Service"
requires-python = "==3.10.*"
dependencies = [
"agebd",
"numpy>=2.2.6",
]
[dependency-groups]
services/topuptool/current/app/src/agebd_topuptool/main.py
+8 -4
View File
@@ -1,13 +1,13 @@
import typer
from agebd.enums import LogLevel
from agebd.runner import CallbackRunner
from agebd.utils import init_logging
from agebd_topuptool import PVs, Service
from agebd_topuptool.runner import Runner
# TODO: remove: cicd test 10
def main(
log_level: LogLevel = typer.Option(
LogLevel.INFO,
@@ -19,11 +19,15 @@ def main(
case_sensitive=False,
),
):
"""
Uses custom/old Runner because migrating to the agebd Runner would be quite
complex.
"""
init_logging(log_level)
service_name = "TOPUPTOOL"
pvs = PVs(service_name=service_name)
service = Service(name=service_name, pvs=pvs)
runner = CallbackRunner(service=service)
service = Service(pvs=pvs)
runner = Runner(svc=service)
runner.start()
services/topuptool/current/app/src/agebd_topuptool/runner.py
+209
View File
@@ -0,0 +1,209 @@
import traceback
from time import perf_counter, sleep
import numpy as np
from epics import poll
from agebd_topuptool.service import Service
class Runner:
def __init__(self, svc: Service) -> None:
self.svc = svc
def start(self):
svc = self.svc
pvs = self.svc.pvs
while not pvs.abort.get():
if pvs.onoff.get() == 1:
# add callback for triggering the loop
if not svc.CallbackActive:
pvs.status.put("running")
pvs.vac_all.add_callback(
callback=self.svc.PressureChangeCallback, index=1, svc=svc
)
svc.CallbackActive = 1
# reload control PVs
svc.update()
# check stop/abort conitions if injection is active
if svc.injection:
# ensure disabled scrubbing during injection
svc.stop_scrubbing()
# abort injection when limits exceeded
if svc.conditions_abort_injection:
# disable injection when upper limit reached
svc.stop_injection()
print("injection aborted")
if not svc.abort:
svc.abort = 1
svc.abort_t0 = perf_counter()
svc.set_status(svc.states["abort"], wait=60)
elif svc.conditions_stop_injection:
# disable injection when upper limit reached
svc.stop_injection()
print("injection stopped")
# start scrubbing if activated
if svc.scrubbi:
sleep(0.1)
svc.start_scrubbing()
svc.set_status(svc.states["decay + scrub"])
else:
svc.set_status(svc.states["decay - scrub"])
elif svc.info != svc.states["inject"]:
svc.set_status(svc.states["inject"])
# do nothing if master trigger missing
elif not svc.trig_master:
svc.set_status(svc.states["wait 4 master"])
# do nothing for 60s if injection was aborted
elif svc.abort:
dt = perf_counter() - svc.abort_t0
if dt > 60:
svc.abort = 0
else:
svc.set_status(svc.states["abort"], wait=60 - dt)
sleep(0.1)
# start acting when callback fired or 100 ms passed
elif svc.ChangeTrue or perf_counter() > svc.tl + 0.1:
# start time for next loop firing, if no vacuum changes
svc.tl = perf_counter()
if svc.conditions_start_injection and svc.info not in [
svc.states["inject"],
svc.states["warmup"],
]:
# disable injection to be sure
pvs.trig_guntxb.put(0)
pvs.trig_gunevt.put(0)
pvs.trig_booinj.put(0)
pvs.trig_boormp.put(0)
pvs.trig_booext.put(0)
pvs.trig_rininj.put(0)
# disable pingers for injection
svc.stop_scrubbing()
# make sure that injection bump and pinger not both fired
sleep(0.4)
svc.prepare_injection()
svc.tp = perf_counter()
# do top up analysis
svc.topup_dt = (svc.tp - svc.topup_t0) / 60
svc.topup_dI = svc.topup_I0 - pvs.cur.get()
pvs.topup_dt.put(svc.topup_dt)
pvs.topup_dI.put(svc.topup_dI)
if svc.topup_I0 > 0:
if svc.topup_dI / svc.topup_I0 < 1:
pvs.topup_tau.put(
(-svc.topup_dt / 60) / np.log(1 - svc.topup_dI / svc.topup_I0)
)
svc.set_status(svc.states["warmup"], wait=20)
elif svc.info == svc.states["warmup"]:
waited = perf_counter() - svc.tp
if waited > 20:
# disable pingers for injection
svc.stop_scrubbing()
sleep(0.5)
svc.start_injection()
svc.set_status(svc.states["inject"])
else:
svc.set_status(svc.states["warmup"], wait=20 - waited)
elif svc.info in [svc.states["pause"], svc.states["wait 4 master"]]:
if svc.injection:
svc.set_status(svc.states["inject"])
else:
if pvs.pinger_v.get():
svc.set_status(svc.states["decay + scrub"])
else:
svc.set_status(svc.states["decay - scrub"])
# stop scrubbing when deactivated
elif not svc.scrubbi:
svc.stop_scrubbing()
# if scrubbing activated and lifetime drops below 1 hour --> stop scrubbing
elif svc.scrubbi and svc.tau < 1:
svc.stop_scrubbing()
print("Scrubbing aborted, tau < 1")
svc.ChangeTrue = 0
# inform watchdog(s)
pvs.running.put(0)
# wait for callback to fire
poll(evt=1.0e-5, iot=0.1)
else:
# WHILE PAUSED REBOOT OF THE SCRUBBING IOC SHOULD NOT DISTURB THE SERVICE
if svc.CallbackActive:
# remove callback
pvs.vac_all.remove_callback(index=1)
svc.CallbackActive = 0
# reload control PVs
svc.update()
# disable injection and pingers when first paused
if svc.info != svc.states["pause"]:
svc.stop_injection()
svc.stop_scrubbing()
svc.set_status(svc.states["pause"])
pvs.status.put("paused")
sleep(0.1)
# sleep and inform watchdog(s)
pvs.alive.put(0)
if pvs.abort.get():
# confirm clean abort
pvs.status.put("killed")
print("Got Killed by Master Service!")
pvs.abortreq.put(0)
# elif svc.crashed:
# # confirm crash
# pvs.status.put('crashed')
# print('Service crashed on exception!')
# print('----- svc.exception.__traceback__ -----')
# print(svc.exception.__traceback__)
# print()
# print('----- traceback.format_exc() -----')
# print(traceback.format_exc())
else:
# confirm unknown exit
pvs.status.put("unknown crash")
print("Service crashed without exception!")
print(traceback.format_exc())
services/topuptool/current/app/src/agebd_topuptool/service.py
+216 -53
View File
@@ -1,12 +1,12 @@
import time
from datetime import datetime
from time import perf_counter
from epics import dbr
import numpy as np
from epics import PV, dbr
from agebd.pv import LocalPVLink
from agebd.service.base import BaseService
from agebd.pv import LocalPVLink, get_pv_class
from agebd.service.pvs import BasePVs
from agebd.utils import printgetversion
from agebd.pv import get_pv_class
__version__ = printgetversion(__file__)
@@ -14,12 +14,58 @@ PV = get_pv_class()
class PVs(BasePVs):
# Option 1:
## Service specific PVs from dedicated IOC
my_pv1 = PV("AGEBD-TOPUPTOOL:AO") # TODO: is this a good PV to use?
## Control PVs
# top-up control mode
# topumod = PV('.VAL')
# top-up mode vacuum limits
vac_max = PV("AGEBD-TOPUPTOOL:CONTROL-VACUUM-MAX.VAL")
vac_min = PV("AGEBD-TOPUPTOOL:CONTROL-VACUUM-MIN.VAL")
# top-up mode current limits
cur_max = PV("AGEBD-TOPUPTOOL:CONTROL-CURRENT-MAX.VAL")
cur_min = PV("AGEBD-TOPUPTOOL:CONTROL-CURRENT-MIN.VAL")
# top-up mode periodic refill (using cur_max)
# topufrq = PV('.VAL')
## Service specific PVs from other IOCs
# ...
vac_dyn = PV("AGEBD-TOPUPTOOL:CONTROL-VACUUM-DYN.VAL")
vac_dyn_frac = PV("AGEBD-TOPUPTOOL:CONTROL-VACUUM-DYN-FRAC.VAL")
grd_max = PV("AGEBD-TOPUPTOOL:CONTROL-GUNGRID-MAX.VAL")
scrubbing = PV("AGEBD-TOPUPTOOL:CONTROL-SCRUB.VAL")
# Input PVs
vac_all = PV("ARIVA-VMAVE-ALL:PRESS-MAX.VAL")
vac_acc = PV("ARIVA-VMAVE:PRESS-MAX.VAL")
vac_fe = PV("ARIVA-VMAVE-FE:PRESS-MAX.VAL")
# cur = PV('AGEBD-DBPM3CURR:CURRENT-AVG.VAL')
cur = PV("ARS07-DPCT-0000:CURR.VAL")
injguard = PV("AGEBD-INJECTIONGUARD:CHECK-SUM.VAL")
injactive = PV("AGEBD-PARAMS:INJECTION-ACTIVE")
tau = PV("AGEBD-PARAMS:LIFETIME.VAL")
fpfb_adc = PV("ARSGE-CECL-FPFB:BEAM-DUMP-ABS-ADC-DATA")
fpfb_max = PV("ARSGE-CECL-FPFB:FPFB-BEAM-DUMP-THR")
# Regulated PVs
trig_master = PV("AGETI-CVME-MASTER-TMA:SR-Inj-Status-Sel.VAL")
trig_guntxb = PV("ALIRF-VME-A-GUN:CH1-MODE.VAL")
trig_gunevt = PV("AGETI-CVME-MASTER-TMA:Evt-LI-Gun-Ena-Sel.VAL")
trig_booinj = PV("AGETI-CVME-MASTER-TMA:Evt-BO-Inj-Ena-Sel.VAL")
trig_boormp = PV("AGETI-CVME-MASTER-TMA:Evt-Old-BO-HW-Ena-Sel.VAL")
trig_booext = PV(
"AGETI-CVME-MASTER-TMA:Evt-BO-Ext-Ena-Sel.VAL"
) # Alias AGETI-CVME-MASTER-TMA:Evt-10
trig_rininj = PV(
"AGETI-CVME-MASTER-TMA:Evt-SR-Inj-Ena-Sel.VAL"
) # Alias AGETI-CVME-MASTER-TMA:Evt-26
gun_grid = PV("ALIRF-GUN-GRID:V-SET.VAL")
pinger_v = PV("ARS01-MKIK-0320:CTRL-ON-OFF.VAL")
pinger_h = PV("ARS01-MKIK-0290:CTRL-ON-OFF.VAL")
topup_dt = PV("AGEBD-TOPUPTOOL:TOPUP-DELTAT.VAL")
topup_dI = PV("AGEBD-TOPUPTOOL:TOPUP-DELTAI.VAL")
topup_tau = PV("AGEBD-TOPUPTOOL:TOPUP-TAU.VAL")
def __init__(self, service_name: str, pv_factory=PV):
super().__init__(service_name, pv_factory)
@@ -32,62 +78,179 @@ class PVs(BasePVs):
# Option 2:
## Service specific PVs from dedicated IOC
self.my_pv2 = PV(
"AGEBD-TOPUPTOOL:BO"
) # TODO: is this a good PV to use?
self.my_pv2 = PV("AGEBD-TOPUPTOOL: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-TOPUPTOOL:CALLBACK", auto_monitor=dbr.DBE_VALUE
) # TODO: ok?
self.CallbackPV = PV("AGEBD-TOPUPTOOL:CALLBACK", auto_monitor=dbr.DBE_VALUE) # TODO: ok?
class Service(BaseService[PVs]):
def __init__(
self,
name: str,
pvs: PVs,
version: str = __version__,
sleep_interval: float = 0.1,
):
super().__init__(name, pvs, version, sleep_interval)
class Service:
## define some "static" class attributes, i.e., persistent over every class call v = vals()
# trigger for execution of loop (controlled by PV callback)
ChangeTrue = 0
# current state of callback
CallbackActive = 0
# start time for next loop firing, if no vacuum changes
tl = perf_counter()
# waittime before injection after preparation
tp = perf_counter()
# status for GUI message
status = "Service Restarted"
# injection abort status
abort = 0
# injection abort timer
abort_t0 = perf_counter()
# states of state machine
states = {
"abort": "Injection Abort",
"decay + scrub": "Decaying Beam, Scrub ON",
"decay - scrub": "Decaying Beam, Scrub OFF",
"inject": "Injecting",
"wait 4 master": "Waiting for Master Trigger",
"warmup": "Septum Warmup",
"pause": "Service Paused",
}
# TODO: What is this for?
self.path = "sls/bd/bin"
def __init__(self, pvs: PVs):
self.pvs = pvs
pvs.started.put(str(datetime.now())[:-7])
pvs.version.put(__version__)
# self.vac = np.max([pv.get() for pv in pvs.vac])
self.vac = pvs.vac_all.get()
self.vac_acc = pvs.vac_acc.get()
# self.vac = pvs.vac_fe.get()
self.vac_array = np.repeat(self.vac_acc, 10)
self.vac_dyn = self.vac * pvs.vac_dyn_frac.get()
pvs.vac_dyn.put(self.vac_dyn * 1e9)
self.topup_t0 = perf_counter()
self.topup_I0 = pvs.cur_min.get()
self.topup_dI = 0
self.topup_dt = 0
def update(self):
# 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)
#######################################################################################
#######################################################################################
pvs = self.pvs
# callback done
self.CallbackFired = 0
# (re-)load input PVs
self.injection = pvs.injactive.get()
self.vac = pvs.vac_all.get()
self.vac_acc = pvs.vac_acc.get()
self.cur = pvs.cur.get()
self.trig_master = pvs.trig_master.get()
self.injguard = pvs.injguard.get()
self.info = pvs.info.get().split(",")[0]
self.tau = pvs.tau.get()
#######################################################################################
#######################################################################################
# here goes all the code that your service
# is supposed to be running in any case, e.g.,
# (re-)load control PVs
self.vac_max = pvs.vac_max.get() * 1e-9
self.vac_dyn = np.max(self.vac_array[:5]) * pvs.vac_dyn_frac.get()
pvs.vac_dyn.put(self.vac_dyn * 1e9)
# 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")
# roll oldest value arr[0] to end, i.e., arr[-1]
self.vac_array = np.roll(self.vac_array, -2)
# replace oldest value with new value
self.vac_array[-2:] = self.vac_acc
# get some pvs/vals
self.val1 = self.pvs.my_pv1.get()
self.val2 = self.pvs.my_pv2.get()
self.vac_min = pvs.vac_min.get() * 1e-9
self.cur_max = pvs.cur_max.get()
self.cur_min = pvs.cur_min.get()
self.grd_max = pvs.grd_max.get()
self.scrubbi = pvs.scrubbing.get()
# maybe sleep to limit the update loop execution rate
time.sleep(1)
#######################################################################################
#######################################################################################
self.conditions_start_injection = ( # is any lower limit reached?
self.vac <= self.vac_min or self.cur <= self.cur_min
) and ( # is no upper limit exceeded ?
self.cur <= self.cur_max and self.vac <= self.vac_max and self.vac_acc <= self.vac_dyn
)
self.conditions_stop_injection = (
# is upper current limit reached/exceeded?
self.cur >= self.cur_max or self.vac >= self.vac_max
)
self.conditions_abort_injection = (
# is any upper limit reached/exceeded?
self.vac_acc >= self.vac_dyn
or np.max(pvs.fpfb_adc.get()) >= 0.95 * pvs.fpfb_max.get()
or self.cur > self.cur_max * 1.02
or self.injguard != 1
)
def stop_injection(self):
pvs = self.pvs
# disable gun trigger
pvs.trig_guntxb.put(0)
# stop all injection events
pvs.trig_gunevt.put(0)
pvs.trig_booinj.put(0)
pvs.trig_boormp.put(0)
pvs.trig_booext.put(0)
pvs.trig_rininj.put(0)
self.injection = 0
# prepare top-up analysis
self.topup_I0 = pvs.cur.get()
self.topup_t0 = perf_counter()
def start_injection(self):
pvs = self.pvs
# enable gun trigger
pvs.trig_guntxb.put(2)
def prepare_injection(self):
pvs = self.pvs
# ensure correct injector settings
pvs.gun_grid.put(self.grd_max)
# start all injection events
pvs.trig_gunevt.put(1)
pvs.trig_booinj.put(1)
pvs.trig_boormp.put(1)
pvs.trig_booext.put(1)
pvs.trig_rininj.put(1)
def start_scrubbing(self):
pvs = self.pvs
pvs.pinger_v.put(1)
pvs.pinger_h.put(1)
print("pingers switched ON")
def stop_scrubbing(self):
pvs = self.pvs
if pvs.pinger_v.get():
pvs.pinger_v.put(0)
print("Vertical pinger switched OFF")
if pvs.pinger_h.get():
pvs.pinger_h.put(0)
print("Horizontal pinger switched OFF")
def set_status(self, status, wait=0):
pvs = self.pvs
if self.info != status:
self.info = status
if wait == 0:
pvs.info.put(status)
if wait > 0:
pvs.info.put(status + ", Wait {:.0f}s".format(wait))
# define callback function to trigger loop execution on change of corresponding pv change
def PressureChangeCallback(self, pvname=None, value=None, char_value=None, **kw):
# trigger execution of loop with actual callback since execution of code here not reliable
self.ChangeTrue += 1
services/topuptool/current/app/tests/test_app.py
+1 -2
View File
@@ -4,6 +4,5 @@ from agebd_topuptool import PVs, Service
def test_app():
service_name = "DUMMY"
pvs = PVs(service_name=service_name)
service = Service(name=service_name, pvs=pvs)
service.CallbackFired = 1
service = Service(pvs=pvs)
service.update()
services/topuptool/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 = [