refactor: extract base runner class
This commit is contained in:
Benjamin Labrecque
@@ -9,107 +9,109 @@ from agebd.service.base import BaseService
|
||||
logger = logging.getLogger(__name__)
|
||||
|
||||
|
||||
class CallbackRunner:
|
||||
class _BaseRunner:
|
||||
"""
|
||||
Shared control loop: poll abort/onoff PVs, dispatch to the active or
|
||||
paused tick, ping the alive watchdog, and stop cleanly on exception.
|
||||
"""
|
||||
|
||||
def __init__(self, service: BaseService):
|
||||
self.svc = service
|
||||
self.pvs = service.pvs
|
||||
|
||||
def start(self):
|
||||
logger.debug(f"Starting {type(self).__name__} for service: {self.svc.name}")
|
||||
while not self.pvs.abort.get():
|
||||
try:
|
||||
# if service is not paused
|
||||
if self.pvs.onoff.get():
|
||||
self._tick_active()
|
||||
# if service is paused
|
||||
else:
|
||||
self._tick_paused()
|
||||
|
||||
# inform watchdog of service alive state
|
||||
self.pvs.alive.put(0)
|
||||
|
||||
except Exception as e:
|
||||
self.svc.exception = e
|
||||
self.svc.crashed = True
|
||||
|
||||
break
|
||||
|
||||
self._on_stop()
|
||||
self.svc.finalize_termination()
|
||||
|
||||
def _tick_active(self):
|
||||
raise NotImplementedError
|
||||
|
||||
def _tick_paused(self):
|
||||
raise NotImplementedError
|
||||
|
||||
def _on_stop(self):
|
||||
pass
|
||||
|
||||
|
||||
class CallbackRunner(_BaseRunner):
|
||||
"""
|
||||
Executes service updates asynchronously, triggered by EPICS PV events.
|
||||
"""
|
||||
|
||||
def __init__(self, service: BaseService):
|
||||
self.svc = service
|
||||
self.pvs = service.pvs
|
||||
self._callback_active = False
|
||||
def _tick_active(self):
|
||||
# add callback for triggering the loop if not active
|
||||
if not self.svc.CallbackActive:
|
||||
self.svc.transition_to_running()
|
||||
self.pvs.CallbackPV.add_callback(callback=self.svc.CallbackFun, index=1)
|
||||
self.svc.CallbackActive = 1
|
||||
|
||||
def start(self):
|
||||
logger.debug(f"Starting CallbackRunner for service: {self.svc.name}")
|
||||
while not self.pvs.abort.get():
|
||||
try:
|
||||
# if service is not paused
|
||||
if self.pvs.onoff.get():
|
||||
# add callback for triggering the loop if not active
|
||||
if not self._callback_active:
|
||||
self.svc.transition_to_running()
|
||||
self.pvs.CallbackPV.add_callback(callback=self.svc.CallbackFun, index=1)
|
||||
self.svc.CallbackActive = 1
|
||||
# run mainloop of service
|
||||
self.svc.update()
|
||||
|
||||
# run mainloop of service
|
||||
self.svc.update()
|
||||
# inform watchdog of service run state
|
||||
self.pvs.running.put(0)
|
||||
|
||||
# inform watchdog of service run state
|
||||
self.pvs.running.put(0)
|
||||
# wait for callback to fire
|
||||
poll(evt=1.0e-5, iot=0.1)
|
||||
|
||||
# wait for callback to fire
|
||||
poll(evt=1.0e-5, iot=0.1)
|
||||
def _tick_paused(self):
|
||||
# remove callback for triggering the loop
|
||||
if self.svc.CallbackActive:
|
||||
self.svc.transition_to_paused()
|
||||
self.pvs.CallbackPV.remove_callback(index=1)
|
||||
self.svc.CallbackActive = 0
|
||||
self.svc.CallbackFired = 0
|
||||
|
||||
# if service is paused
|
||||
else:
|
||||
# remove callback for triggering the loop
|
||||
if self.svc.CallbackActive:
|
||||
self.svc.transition_to_paused()
|
||||
self.pvs.CallbackPV.remove_callback(index=1)
|
||||
self.svc.CallbackActive = 0
|
||||
self.svc.CallbackFired = 0
|
||||
|
||||
# wait until service is unpaused
|
||||
sleep(0.1)
|
||||
|
||||
# inform watchdog of service alive state
|
||||
self.pvs.alive.put(0)
|
||||
|
||||
except Exception as e:
|
||||
self.svc.exception = e
|
||||
self.svc.crashed = True
|
||||
|
||||
break
|
||||
# wait until service is unpaused
|
||||
sleep(0.1)
|
||||
|
||||
def _on_stop(self):
|
||||
# remove callback(s) from event(s)
|
||||
self.pvs.CallbackPV.remove_callback(index=1)
|
||||
self.svc.finalize_termination()
|
||||
|
||||
|
||||
class PeriodicRunner:
|
||||
class PeriodicRunner(_BaseRunner):
|
||||
"""
|
||||
Executes service updates at a fixed frequency.
|
||||
"""
|
||||
|
||||
def __init__(self, service: BaseService):
|
||||
self.svc = service
|
||||
self.pvs = service.pvs
|
||||
def _tick_active(self):
|
||||
if not self.svc.status == ServiceStatus.RUNNING:
|
||||
self.svc.transition_to_running()
|
||||
|
||||
def start(self):
|
||||
logger.debug(f"Starting PeriodicRunner for service: {self.svc.name}")
|
||||
while not self.pvs.abort.get():
|
||||
try:
|
||||
# if service is not paused
|
||||
if self.pvs.onoff.get():
|
||||
if not self.svc.status == ServiceStatus.RUNNING:
|
||||
self.svc.transition_to_running()
|
||||
# run mainloop of service
|
||||
self.svc.update()
|
||||
|
||||
# run mainloop of service
|
||||
self.svc.update()
|
||||
# inform watchdog of service run state
|
||||
self.pvs.running.put(0)
|
||||
|
||||
# inform watchdog of service run state
|
||||
self.pvs.running.put(0)
|
||||
# wait for timer
|
||||
sleep(self.svc.sleep_interval)
|
||||
|
||||
# wait for timer
|
||||
sleep(self.svc.sleep_interval)
|
||||
def _tick_paused(self):
|
||||
if not self.svc.status == ServiceStatus.PAUSED:
|
||||
self.svc.transition_to_paused()
|
||||
|
||||
# if service is paused
|
||||
else:
|
||||
if not self.svc.status == ServiceStatus.PAUSED:
|
||||
self.svc.transition_to_paused()
|
||||
self.svc.CallbackFired = 0
|
||||
|
||||
self.svc.CallbackFired = 0
|
||||
|
||||
# wait until service is unpaused
|
||||
sleep(0.1)
|
||||
|
||||
# inform watchdog of service alive state
|
||||
self.pvs.alive.put(0)
|
||||
|
||||
except Exception as e:
|
||||
self.svc.exception = e
|
||||
self.svc.crashed = True
|
||||
|
||||
break
|
||||
|
||||
self.svc.finalize_termination()
|
||||
# wait until service is unpaused
|
||||
sleep(0.1)
|
||||
|
||||
Reference in New Issue
Block a user