linse/seweb
0
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

add dummy webserver

Browse Source
This commit is contained in:
zolliker
2024-10-23 16:35:59 +02:00
parent dc887e68d5
commit 7d536c187b
2 changed files with 187 additions and 0 deletions
Download Patch File Download Diff File Expand all files Collapse all files

9
dummy-webserver Executable file
View File

@ -0,0 +1,9 @@
#!/usr/bin/env python3
import sys
import pathlib
sys.path.insert(0, str((pathlib.Path(__file__) / '..').resolve()))
import webserver
from dummy import SecopDummyInstrument
webserver.instrument = webserver.main(SecopDummyInstrument)

178
dummy.py Normal file
View File

@ -0,0 +1,178 @@
import time
import math
import io
from colors import assign_colors_to_curves
from secop import SecopInstrument, SecopClient
from base import get_abs_time
class DummyGraph:
def __init__(self):
self.blocks = []
self.phase = {}
self.end_time = 0
phase = 0
for i in range(5):
curves = []
for j in range(5 - i):
name = f'curve{i}{j}'
curves.append({'name': name, 'label': name.title(), 'color': str(j+1)})
self.phase[name] = phase
phase += 15
unit = 'ABCDEFG'[i]
self.blocks.append({'tag': unit, 'unit': unit, 'curves': curves})
def dummy_fun(self, var, t):
return math.sin(math.radians(t % 3600 / 10 - self.phase[var])) + 1.1
def get_curves(self, variables, start, end):
result = {}
step = 5 * max(1, (end - start) // 1000)
for i, var in enumerate(variables):
result[var] = [(t, self.dummy_fun(var, t)) for t in range(start - start % step, end + 1, step)]
time.sleep(0.5)
return result
def w_graph(self, variables, time="-1800,0", interval=None):
"""
Gets the curves given by variables in the time range "time", spaced by "interval" if given (binning/resolution)
Called when the route /graph is reached.
Parameters :
variables (str) : a comma separataed value string of variable names (influx names) to retrieve
time (str) : a commma separated value string (range) of seconds. They are treated as relative from now
if they are lesser than one year.
interval (str) : the interval (resolution) of the values to get (string in milliseconds)
Returns :
{"type":"graph-draw", "graph":{(str):[[(int),(float)]]}} : a dictionnary with its "graph-draw" type
(so it can be processed by the client), and a "graph" dictionnary with the variable names as key,
and an array of points as a tuple (timestamp, y-value as float)
"""
time = [float(t) for t in time.split(',')]
start, end, now = get_abs_time(time + [0])
start, end, now = int(start), int(end), int(now)
#self.livemode = self.ACTUAL if end+10 >= now else self.HISTORICAL
return dict(type='graph-draw', graph=self.get_curves(variables.split(','), start, end))
def w_gettime(self, time):
"""
Gets the server time for the give time.
Called when the route /gettime is reached.
Parameters :
time (str="-1800,0") : the given point in time represented by a string, which is a comma separated unix
timestamp values list (in seconds). They are treated as relative from now if they are lesser than one year.
Returns :
{"type":"time", "time":(int)} : a dictionnary with its "time" type (so the data can be processed by the
client) and the server unix timestamp in seconds corresponding to the time asked by the client
"""
time = [float(t) for t in time.split(',')]
return dict(type='time', time=get_abs_time(time))
def w_getvars(self, time, userconfiguration = None):
"""
Gets the curve names available at a given point in time, with a possible user configuration on the client side.
Called when the route /getvars is reached.
Parameters :
time (str) : the given point in time represented by a string, which is a unix timestamp in seconds.
It is treated as relative from now if it is lesser than one year.
userconfiguration (str|None) : the JSON string representing the user configuration
Returns :
{"type":"var_list", "device":(str), "blocks":[{"tag":(str),"unit":(str), "curves":
[{"name":(str), "label":(str), "color":(str), "original_color":(str)}]}]}:
a dictionnary with its "var_list" type (so the data can be processed by the client), the device that
was currently set at that time, and the available curves with the name of the internal variable,
the color to display for this curve, its original color in SEA, grouped by their tag (which is a
category or unit if absent) and their unit (in "blocks")
"""
time = [float(t) for t in time.split(',')]
assign_colors_to_curves(self.blocks)
result = dict(type='var_list')
result['blocks'] = self.blocks
result['device'] = 'dummy'
return result
def w_updategraph(self):
"""
Sets the current visualisation mode to LIVE if not in HISTORICAL mode.
Called when the route /updategraph is reached.
Returns :
{"type":"accept-graph", "live": bool} : a dict with its "accept-graph" type and a "live"
value telling if the server could change its visualization mode to live
"""
return dict(type='accept-graph', live=True)
def w_export(self, variables, time, nan, interval):
"""
Returns the bytes of a dataframe with the curves given by variables in the time range "time"
Called when the route /export is reached.
Parameters :
variables (str) : a comma separataed value string of variable names (influx names) to retrieve
time (str) : a commma separated value string (range) of seconds.
nan (str) : the representation for NaN values in the TSV
interval (str) : the interval (resolution) of the values to get (string in seconds)
Returns :
io.BytesIO : an BytesIO object containing the dataframe to retrieve
"""
mem = io.BytesIO()
return mem
def graphpoll(self):
"""
Polls the last known values for all the available variables, and returns only those whose polled values
are more recent than the most recent displayed one.
Every plain minute, all the variables are returned with a point having their last known value at the current
timestamp to synchronize all the curves on the GUI.
Returns :
{"type":"graph-update", "time":(int), "graph":{(str):[[(int),(float)]]}} | None :
a dictionnary with its "graph-update" type
(so it can be processed by the client), and a "graph" dictionnary with the variable names as key,
and an array of points, which are an array containing the timestamp
as their first value, and the y-value in float as their second one
"""
now, = get_abs_time([0])
if not self.end_time:
self.end_time = now
return None
result = self.get_curves(self.phase, self.end_time, now)
for variable, values in list(result.items()):
# removes points older than the last known point
# (queries are in seconds and might return points already displayed)
while values and values[0][0] < self.end_time:
values.pop(0)
if not values or values[-1][0] > self.end_time:
del result[variable]
self.end_time = now
if len(result) > 0:
return dict(type='graph-update', time=now, graph=result)
return None
class SecopDummyClient(SecopClient, DummyGraph):
def __init__(self, instrument):
SecopClient.__init__(self, instrument)
DummyGraph.__init__(self)
def poll(self):
messages = super().poll()
msg = self.graphpoll()
if msg:
messages.append(msg)
return messages
class SecopDummyInstrument(SecopInstrument):
def new_client(self):
return self.register(SecopDummyClient(self))