94 lines
2.0 KiB
Python
94 lines
2.0 KiB
Python
from time import sleep
|
|
from bsread import source
|
|
from bsvar import BSVar
|
|
from prodthread import ProdThread
|
|
|
|
|
|
class BSCache:
|
|
|
|
def __init__(self):
|
|
self.channels = {}
|
|
self.data = None
|
|
self.pt = ProdThread(self.run)
|
|
|
|
|
|
def __iter__(self):
|
|
return self
|
|
|
|
def __next__(self):
|
|
self.pt.start()
|
|
self.data = data = self.pt.queue.get()
|
|
return data
|
|
|
|
|
|
def run(self, queue, running):
|
|
channels = self.channels.keys()
|
|
configs = self.channels.values()
|
|
with source(channels=configs, receive_timeout=-1) as src:
|
|
while running.is_set():
|
|
msg = src.receive()
|
|
data = repack(channels, msg)
|
|
if data:
|
|
queue.put(data)
|
|
|
|
|
|
def stop(self):
|
|
self.pt.stop()
|
|
|
|
|
|
def get_var(self, name, modulo=None, offset=None):
|
|
if name is not "pid" and not name in self.channels:
|
|
cfg = make_channel_config(name, modulo, offset)
|
|
self.update_source(name, cfg)
|
|
return BSVar(name, self)
|
|
|
|
|
|
def update_source(self, name, cfg):
|
|
self.pt.stop()
|
|
print("add channel", name)
|
|
self.channels[name] = cfg
|
|
self.pt.start()
|
|
|
|
while self.data is None or name not in self.data:
|
|
print("dropping data that is missing new channel:", name)
|
|
next(self)
|
|
|
|
|
|
|
|
|
|
|
|
def make_channel_config(name, modulo, offset):
|
|
res = {}
|
|
if modulo is not None:
|
|
res["modulo"] = modulo
|
|
if offset is not None:
|
|
res["offset"] = offset
|
|
if not res:
|
|
return name
|
|
res["name"] = name
|
|
return res
|
|
|
|
|
|
|
|
|
|
|
|
def repack(channels, message):
|
|
data = message.data.data
|
|
pulse_id = message.data.pulse_id
|
|
|
|
res = {n: data[n].value for n in channels}
|
|
# res = {k: v for k, v in res.items() if v is not None}
|
|
|
|
#TODO: should this be a ValueError?
|
|
if any(v is None for v in res.values()):
|
|
return None
|
|
|
|
if not res:
|
|
return None
|
|
|
|
res["pid"] = pulse_id
|
|
return res
|
|
|
|
|
|
|