tligui_y/slic
0
0
Fork 0
forked from slic/slic
Code Pull Requests Actions Wiki Activity

added BaseChecker for defining the Checker API, added Checker parent class for the collecting/analysis logic, adjusted PVChecker and ValueChecker to make use of the former

Browse Source
This commit is contained in:
augustin_s
2020-04-24 12:20:34 +00:00
parent d6409aff3e
commit d408db57f5
5 changed files with 119 additions and 136 deletions
Download Patch File Download Diff File Expand all files Collapse all files
slic/checkers/__init__.py
+1
View File
@@ -1,4 +1,5 @@
from .pvchecker import PVChecker
from .valuechecker import ValueChecker
slic/checkers/basechecker.py
+15
View File
@@ -0,0 +1,15 @@
from abc import ABC, abstractmethod
class BaseChecker(ABC):
@abstractmethod
def get_ready(self):
raise NotImplementedError
@abstractmethod
def is_happy(self):
raise NotImplementedError
slic/checkers/checker.py
+99
View File
@@ -0,0 +1,99 @@
from abc import abstractmethod
from time import sleep
from .basechecker import BaseChecker
from .utils import within, within_fraction, fraction_to_percentage
class Checker(BaseChecker):
@abstractmethod
def __init__(self, vmin, vmax, wait_time, required_fraction):
self.vmin = vmin
self.vmax = vmax
self.wait_time = wait_time
self.required_fraction = required_fraction
self.data = []
def check(self):
val = self.current()
return within(val, self.vmin, self.vmax)
@abstractmethod
def current(self):
raise NotImplementedError
def sleep(self):
sleep(self.wait_time)
def clear_and_start_counting(self):
self.clear()
self.start_counting()
def clear(self):
self.data.clear()
@abstractmethod
def start_counting(self):
raise NotImplementedError
def stop_counting_and_analyze(self):
self.stop_counting()
self.analyze()
@abstractmethod
def stop_counting(self):
raise NotImplementedError
def analyze(self):
vmin = self.vmin
vmax = self.vmax
required_fraction = self.required_fraction
fraction = within_fraction(self.data, vmin, vmax)
result = (fraction >= required_fraction)
status = "happy" if result else "unhappy"
percentage = fraction_to_percentage(fraction)
required_percentage = fraction_to_percentage(required_fraction)
msg = "Checker {}: {}% within limits [{}, {}), required was {}%.".format(status, percentage, vmin, vmax, required_percentage)
print(msg)
return result
def get_ready(self):
time_start = time()
checker_ever_unhappy = False
while not self.long_check():
checker_ever_unhappy = True
delta_t = time() - time_start
print(f"Checker is unhappy, waiting for OK conditions since {delta_t:5.1f} seconds.")
if checker_ever_unhappy:
delta_t = time() - time_start
print(f"Checker was unhappy, waited for {delta_t:5.1f} seconds.")
self.clear_and_start_counting()
def is_happy(self):
return self.stop_counting_and_analyze()
def long_check(self):
self.clear_and_start_counting()
self.sleep()
state = self.stop_counting_and_analyze()
return state
slic/checkers/pvchecker.py
+2 -67
View File
@@ -1,10 +1,9 @@
from epics import PV
from time import sleep
from .utils import within, within_fraction, fraction_to_percentage
from .checker import Checker
class PVChecker:
class PVChecker(Checker):
def __init__(self, channel, vmin, vmax, wait_time, required_fraction):
self.channel = channel
@@ -17,24 +16,9 @@ class PVChecker:
self.data = []
def check(self):
val = self.current()
return within(val, self.vmin, self.vmax)
def current(self):
return self.pv.get()
def sleep(self):
sleep(self.wait_time)
def clear_and_start_counting(self):
self.clear()
self.start_counting()
def clear(self):
self.data.clear()
def start_counting(self):
def collect(value=None, **kwargs):
@@ -44,57 +28,8 @@ class PVChecker:
self.pv.add_callback(callback=collect)
def stop_counting_and_analyze(self):
self.stop_counting()
self.analyze()
def stop_counting(self):
self.pv.clear_callbacks()
def analyze(self):
vmin = self.vmin
vmax = self.vmax
required_fraction = self.required_fraction
fraction = within_fraction(self.data, vmin, vmax)
result = (fraction >= required_fraction)
status = "happy" if result else "unhappy"
percentage = fraction_to_percentage(fraction)
required_percentage = fraction_to_percentage(required_fraction)
msg = "Checker {}: {}% within limits [{}, {}), required was {}%.".format(status, percentage, vmin, vmax, required_percentage)
print(msg)
return result
def get_ready(self):
time_start = time()
checker_ever_unhappy = False
while not self.long_check():
checker_ever_unhappy = True
delta_t = time() - time_start
print(f"Checker is unhappy, waiting for OK conditions since {delta_t:5.1f} seconds.")
if checker_ever_unhappy:
delta_t = time() - time_start
print(f"Checker was unhappy, waited for {delta_t:5.1f} seconds.")
self.clear_and_start_counting()
def is_happy(self):
return self.stop_counting_and_analyze()
def long_check(self):
self.clear_and_start_counting()
self.sleep()
state = self.stop_counting_and_analyze()
return state
slic/checkers/valuechecker.py
+2 -69
View File
@@ -1,11 +1,8 @@
from time import sleep
from .utils import within, within_fraction, fraction_to_percentage
from .checker import Checker
from slic.runners import LoopRunner
class ValueChecker:
class ValueChecker(Checker):
def __init__(self, get_value, vmin, vmax, wait_time, required_fraction):
self.get_value = get_value
@@ -18,24 +15,9 @@ class ValueChecker:
self.runner = None
def check(self):
val = self.current()
return within(val, self.vmin, self.vmax)
def current(self):
return self.get_value()
def sleep(self):
sleep(self.wait_time)
def clear_and_start_counting(self):
self.clear()
self.start_counting()
def clear(self):
self.data.clear()
def start_counting(self):
if self.runner:
@@ -48,10 +30,6 @@ class ValueChecker:
self.runner = LoopRunner(collect, self.wait_time)
def stop_counting_and_analyze(self):
self.stop_counting()
self.analyze()
def stop_counting(self):
if not self.runner:
return # can only stop something, if we started it
@@ -60,49 +38,4 @@ class ValueChecker:
self.runner = None
def analyze(self):
vmin = self.vmin
vmax = self.vmax
required_fraction = self.required_fraction
fraction = within_fraction(self.data, vmin, vmax)
result = (fraction >= required_fraction)
status = "happy" if result else "unhappy"
percentage = fraction_to_percentage(fraction)
required_percentage = fraction_to_percentage(required_fraction)
msg = "Checker {}: {}% within limits [{}, {}), required was {}%.".format(status, percentage, vmin, vmax, required_percentage)
print(msg)
return result
def get_ready(self):
time_start = time()
checker_ever_unhappy = False
while not self.long_check():
checker_ever_unhappy = True
delta_t = time() - time_start
print(f"Checker is unhappy, waiting for OK conditions since {delta_t:5.1f} seconds.")
if checker_ever_unhappy:
delta_t = time() - time_start
print(f"Checker was unhappy, waited for {delta_t:5.1f} seconds.")
self.clear_and_start_counting()
def is_happy(self):
return self.stop_counting_and_analyze()
def long_check(self):
self.clear_and_start_counting()
self.sleep()
state = self.stop_counting_and_analyze()
return state