chore: change default PVs in template python app
Deploy bin / deploy (push) Successful in 1s
Deploy agebd python package / deploy (push) Successful in 1s

This commit is contained in:
Benjamin Labrecque
2026-07-08 11:53:32 +02:00
parent 051ed3311b
commit 39f38d6bb9
3 changed files with 24 additions and 28 deletions
+17 -25
View File
@@ -1,10 +1,10 @@
from time import perf_counter, sleep, time
import numpy as np
from epics import PV
from time import time, sleep, perf_counter
class PVLink:
def __init__(self, pvname, **kwargs):
self.link = self.connect(pvname, **kwargs)
@@ -14,21 +14,19 @@ class PVLink:
@property
def pvname(self):
return self.link.pvname
@property
def connected(self):
return self.link.connected
def poll(self, **kwargs):
self.link.poll(**kwargs)
def add_callback(self, **kwargs):
self.link.add_callback(**kwargs)
def remove_callback(self, **kwargs):
self.link.remove_callback(**kwargs)
@@ -41,21 +39,19 @@ class PVLink:
while wait:
timeout = perf_counter() - t0 > 10
wait = not link.connected and not timeout
link.poll(evt=1.e-5, iot=0.1)
link.poll(evt=1.0e-5, iot=0.1)
if timeout:
print('failed to establish link to {:}'.format(pvname))
print("failed to establish link to {:}".format(pvname))
return link
def reconnect(self, **kwargs):
t0 = perf_counter()
print('{:} is not connected, attempting to reconnect for 60s ...'.format(self.pvname))
print("{:} is not connected, attempting to reconnect for 60s ...".format(self.pvname))
while not self.connected and (perf_counter() - t0) < 60:
self.link.disconnect()
sleep(5)
@@ -65,45 +61,40 @@ class PVLink:
sleep(1)
if not self.connected:
print('{:} remains not connected, reconnect failed ...'.format(self.pvname))
print("{:} remains not connected, reconnect failed ...".format(self.pvname))
self.timestamp = time()
self.value = np.nan
else:
print('{:} is connected again...'.format(self.pvname))
print("{:} is connected again...".format(self.pvname))
def get(self, **kwargs):
if not self.connected:
self.reconnect()
try:
self.value = self.link.get(timeout=0.1, **kwargs)
self.timestamp = self.link.timestamp
except:
print('{:} has an issue, cannot get'.format(self.pvname))
print("{:} has an issue, cannot get".format(self.pvname))
self.timestamp = time()
self.value = np.nan
return self.value
def put(self, val, **kwargs):
if not self.connected:
self.reconnect()
try:
ret = self.link.put(val, **kwargs)
return ret
except:
print('{:} has an issue, cannot put {:}'.format(self.pvname, val))
print("{:} has an issue, cannot put {:}".format(self.pvname, val))
return False
@@ -111,11 +102,11 @@ class DevPVLink(PVLink):
"""
Uses DEV PVs
"""
def __init__(self, pvname, **kwargs):
dev_pvname = self._get_pvname_dev(pvname)
super().__init__(dev_pvname, **kwargs)
def _get_pvname_dev(self, pvname):
device, record = pvname.split(":")
return f"{device}-DEV:{record}"
@@ -125,6 +116,7 @@ class LocalPVLink:
"""
A test-only subclass. It skips network reads and writes entirely.
"""
# TODO: should we dyanmically append "-LOCAL" to PV name?
def __init__(self, pvname, initial_value=None, **kwargs):
self.pvname = pvname
@@ -133,7 +125,7 @@ class LocalPVLink:
self.timestamp = time()
self._callbacks = {}
def poll(self, **kwargs):
pass
@@ -18,7 +18,9 @@ PV = get_pv_class()
class PVs(BasePVs):
# Option 1:
## Service specific PVs from dedicated IOC
my_pv1 = PV("MY_PV1")
my_pv1 = PV(
"AGEBD-CPCL-{{ service_name_upper }}-DEV:AO"
) # TODO: is this a good PV to use?
## Service specific PVs from other IOCs
# ...
@@ -34,7 +36,9 @@ class PVs(BasePVs):
# Option 2:
## Service specific PVs from dedicated IOC
self.my_pv2 = PV("MY_PV2")
self.my_pv2 = PV(
"AGEBD-CPCL-{{ service_name_upper }}-DEV:BO"
) # TODO: is this a good PV to use?
## Service specific PVs from other IOCs
# ...
@@ -1,4 +1,4 @@
file {{ service_name_upper }}.template {
pattern { DEVICE }
{ AGEBD-{{ service_name_upper }} }
{ AGEBD-{{ service_name_upper }}{{ '{{ agebd_env }}' }} }
}