Skip to content

szarp
szarp
Source
Select Git revision
...
Target
Select Git revision
 
Commits (2)
Hide whitespace changes
Inline Side-by-side
Showing with 126 additions and 24 deletions
debian/control
View file @ 2821ef30
......@@ -6,7 +6,7 @@ Maintainer: Newterm Developers <devs@newterm.pl>
Section: misc
Standards-Version: 3.6.1
X-Python-Version: >= 2.6
Build-Depends: debhelper (>= 5.0.38), ccache, cmake (>=3.2), libwxgtk3.0-dev, wx-common, libwxbase3.0-dev, libxml2-dev, bison, flex, imagemagick, jadetex, gettext, xsltproc, libxslt1-dev, libnewt-dev, libssl-dev | libssl1.0-dev, libpam-dev | libpam0g-dev, libcurl3-dev | libcurl4-gnutls-dev, perl, librsync-dev, libgtk2.0-dev, libsqlite3-dev, libldap2-dev (>= 2.3.5), python-setuptools (>= 0.6b3-1), libzip-dev, openjade, python-pybabel | python-babel, libicu-dev, libxt-dev, libc-ares-dev, libxmlrpc-epi-dev, libluajit-5.1-dev | liblua5.1-0-dev, libftgl-dev, libboost-dev (>=1.55), libboost-system-dev (>=1.55), libboost-thread-dev (>=1.55), libboost-program-options-dev (>=1.55), libboost-regex-dev (>=1.55), libboost-filesystem-dev (>=1.55), libboost-python-dev (>=1.55), libboost-locale-dev (>=1.55), libboost-signals-dev (>=1.55), libasio-dev | libasio1.55-dev, libstdc++-dev, libxpm-dev, libevent-dev(>=2.0), docbook-dsssl, docbook, rsync, dh-python, konwert, pyqt4-dev-tools, libtool, python-all-dev, python-dev, pyqt4-dev-tools, qt4-linguist-tools, libzmq3-dev, libcppunit-dev, protobuf-compiler, python-zmq, libprotobuf-dev, python-sip, python-lxml, python-protobuf, texlive-generic-recommended, libsnap7-dev, libsystemd-dev
Build-Depends: debhelper (>= 5.0.38), ccache, cmake (>=3.2), libwxgtk3.0-dev, wx-common, libwxbase3.0-dev, libxml2-dev, bison, flex, imagemagick, jadetex, gettext, xsltproc, libxslt1-dev, libnewt-dev, libssl-dev | libssl1.0-dev, libpam-dev | libpam0g-dev, libcurl3-dev | libcurl4-gnutls-dev, perl, librsync-dev, libgtk2.0-dev, libsqlite3-dev, libldap2-dev (>= 2.3.5), python-setuptools (>= 0.6b3-1), libzip-dev, openjade, python-pybabel | python-babel, libicu-dev, libxt-dev, libc-ares-dev, libxmlrpc-epi-dev, libluajit-5.1-dev | liblua5.1-0-dev, libftgl-dev, libboost-dev (>=1.55), libboost-system-dev (>=1.55), libboost-thread-dev (>=1.55), libboost-program-options-dev (>=1.55), libboost-regex-dev (>=1.55), libboost-filesystem-dev (>=1.55), libboost-python-dev (>=1.55), libboost-locale-dev (>=1.55), libboost-signals-dev (>=1.55), libasio-dev | libasio1.55-dev, libstdc++-dev, libxpm-dev, libevent-dev(>=2.0), docbook-dsssl, docbook, rsync, dh-python, konwert, pyqt4-dev-tools, libtool, python-all-dev, python-dev, pyqt4-dev-tools, qt4-linguist-tools, libzmq3-dev, libcppunit-dev, protobuf-compiler, python-zmq, libprotobuf-dev, python-sip, python-lxml, python-protobuf, python3-coverage, texlive-generic-recommended, libsnap7-dev, libsystemd-dev
Package: szarp-base
Section: misc
......
python/CMakeLists.txt
View file @ 2821ef30
add_subdirectory(meaner4)
add_subdirectory(meaner5)
add_subdirectory(parhub)
add_subdirectory(protobuf)
add_subdirectory(pysz4)
add_subdirectory(pysz5)
add_subdirectory(pyszbase)
add_subdirectory(pytimer)
add_subdirectory(utils)
python/meaner5/meaner5.py
View file @ 2821ef30
......@@ -17,12 +17,22 @@ class Meaner5:
self.time_source = time_source
self.heartbeat_param_name = "Status/Meaner3/program_uruchomiony"
self.terminated = False
self.probe_save_time_shift = self.saving_interval * 0.5
self.max_future_shift = self.saving_interval * 2
assert(self.probe_save_time_shift > 0) # if equal to zero, daemons will send us probes at the same time that we would like to store data
assert(self.probe_save_time_shift < self.saving_interval) # with a small redesign we could use larger values too
def calculate_next_save_time(self, now):
def calculate_first_probe_time(self, now):
next_save_time = math.trunc(now) + self.saving_interval
next_save_time_aligned = next_save_time - (next_save_time % self.saving_interval)
return next_save_time_aligned
def calculate_next_probe_time(self, last_probe_time):
return last_probe_time + self.saving_interval
def calculate_save_time(self, probe_time):
return probe_time + self.probe_save_time_shift
def read_socket(self):
try:
while True:
......@@ -30,17 +40,31 @@ class Meaner5:
params_values = paramsvalues_pb2.ParamsValues.FromString(msg)
for param_value in params_values.param_values:
index = param_value.param_no
if index not in self.msgs.keys():
self.logger.error("Cannot store param with index: %d, no param name", index)
continue
self.msgs[index].append(param_value)
self.logger.debug("Param '%s' value received: %d with time: %d" % (self.param_names_ordered[index], param_value.value_before_prec, param_value.time))
except zmq.ZMQError as e:
if e.errno != zmq.EAGAIN:
raise
def msg_list_to_param_value(self, msg_list):
def msg_list_to_param_value(self, msg_list, probe_timestamp):
numeric_value = 0
count = 0
prec = None
is_nodata = True
unused_msgs = []
for msg in msg_list:
if msg.time <= probe_timestamp - self.saving_interval:
self.logger.debug("Ignoring msg, because it's too old, sth is malfunctioning...")
continue
if msg.time > probe_timestamp + self.max_future_shift:
self.logger.debug("Ignoring msg, because it's too far from the future, sth is malfunctioning...")
continue
if msg.time > probe_timestamp:
unused_msgs.append(msg)
continue
if msg.is_nan:
continue
numeric_value = numeric_value + msg.value_before_prec
......@@ -51,26 +75,36 @@ class Meaner5:
else:
if msg.prec != prec:
self.logger.error("Precision changed during time step, cannot continue")
return self.base_handler.ParamValue()
return (self.base_handler.ParamValue(), unused_msgs)
if is_nodata:
return self.base_handler.ParamValue()
return self.base_handler.ParamValue(value_before_prec=int(numeric_value / count), prec=prec, is_no_data=False)
return (self.base_handler.ParamValue(), unused_msgs)
return (self.base_handler.ParamValue(value_before_prec=int(numeric_value / count), prec=prec, is_no_data=False), unused_msgs)
def db_store(self, timestamp):
def db_store(self, probe_timestamp):
for index in self.msgs.keys():
value = self.msg_list_to_param_value(self.msgs[index])
value, unused_msgs = self.msg_list_to_param_value(self.msgs[index], probe_timestamp)
self.msgs[index] = unused_msgs
param_name = self.param_names_ordered[index]
self.db.write_value(param_name, timestamp, value)
self.db.write_value(self.heartbeat_param_name, timestamp, self.base_handler.ParamValue(value_before_prec=1, prec=0, is_no_data=False))
self.db.write_value(param_name, probe_timestamp, value)
self.logger.debug("Param '%s' value store: %d with time: %d" % (param_name, value.value_before_prec, probe_timestamp))
self.db.write_value(self.heartbeat_param_name, probe_timestamp, self.base_handler.ParamValue(value_before_prec=1, prec=0, is_no_data=False))
def loop(self):
now = self.time_source.time()
next_probe_time = self.calculate_first_probe_time(now)
while not interrupt_caught:
self.clear_msgs()
next_save_time = self.calculate_save_time(next_probe_time)
now = self.time_source.time()
next_save_time = self.calculate_next_save_time(now)
if next_save_time < now:
self.logger.error("next_save_time (%d) < now (%d), shifting next_probe_time" % (next_save_time, t))
next_probe_time = self.calculate_first_probe_time(now)
next_save_time = self.calculate_save_time(next_probe_time)
wait_ms = (next_save_time - now) * 1000
wait_ms_step = wait_ms / 10.0
while self.time_source.time() < next_save_time:
t = now
while t < next_save_time:
t = self.time_source.time()
if interrupt_caught:
break
try:
......@@ -82,7 +116,8 @@ class Meaner5:
self.read_socket()
if interrupt_caught:
break
self.db_store(next_save_time)
self.db_store(next_probe_time)
next_probe_time = self.calculate_next_probe_time(next_probe_time)
def clear_msgs(self):
for index in range(len(self.param_names_ordered)):
......
python/meaner5/meaner5dmn.py
View file @ 2821ef30
#!/usr/bin/python2
#!/usr/bin/python3
import fcntl, sys, os, struct
import time
......@@ -12,7 +12,7 @@ from simpleparamsreader import SimpleParamsReader
"""
Meaner5 daemon.
To run manually: env PYTHONPATH="${PYTHONPATH}:.:../protobuf:../pysz5:../../utils/probes_server" python ./meaner5dmn.py
To run manually: env PYTHONPATH="${PYTHONPATH}:.:../protobuf:../pysz5:../../utils/probes_server" python3 ./meaner5dmn.py
"""
daemon_name = "meaner5"
......
python/meaner5/run_tests.sh
View file @ 2821ef30
......@@ -2,4 +2,5 @@
# ugly way of calling python tests, it should be integrated somehow with python pybuild tests launched by dh_auto_test
env PYTHONPATH="${PYTHONPATH}:.:../protobuf:../pysz5" /usr/bin/python2 -m unittest discover -s ./unittests/
env PYTHONPATH="${PYTHONPATH}:.:../protobuf:../pysz5" python3-coverage run --source . --omit "unittests/*" -m unittest discover -s ./unittests/
python3-coverage report -m
python/meaner5/unittests/test_meaner5.py
View file @ 2821ef30
......@@ -15,26 +15,64 @@ def run_server():
context = zmq.Context()
socket = context.socket(zmq.PUB)
socket.bind(parhub_uri)
for i in range(0, 200):
for i in range(0, 2000):
values = paramsvalues_pb2.ParamsValues()
value = values.param_values.add()
value.param_no = 1
value.time = 0
value.time = 1
value.value_before_prec = 5
value.prec = 0
value.is_nan = False
value = values.param_values.add()
value.param_no = 2
value.time = 0
value.time = 1
value.value_before_prec = 6
value.prec = 1
value.is_nan = False
value = values.param_values.add()
value.param_no = 3
value.time = 0
value.time = 1
value.value_before_prec = 7
value.prec = i % 2 # varying prec should result in nodata
value.is_nan = False
value = values.param_values.add()
value.param_no = 100
value.time = 1
value.value_before_prec = 5
value.prec = 0
value.is_nan = False
value = values.param_values.add()
value.param_no = 1
value.time = 20
value.value_before_prec = 5
value.prec = 0
value.is_nan = True
value = values.param_values.add()
value.param_no = 2
value.time = 20
value.value_before_prec = 16
value.prec = 1
value.is_nan = False
value = values.param_values.add()
value.param_no = 3
value.time = 20
value.value_before_prec = 17
value.prec = 2
value.is_nan = False
value = values.param_values.add()
value.param_no = 1
value.time = 60 # this one should trigger adding to unused_msgs
value.value_before_prec = 50
value.prec = 0
value.is_nan = False
value = values.param_values.add()
value.param_no = 3
value.time = 100 # this one should trigger discarding too far from the future msgs
value.value_before_prec = 17
value.prec = 2
value.is_nan = False
socket.send(values.SerializeToString())
sleep(0.01)
......@@ -85,8 +123,8 @@ class SimpleTest(unittest.TestCase):
db_handler = TestDbHandler(db)
param_names_ordered = ["zero", "first", "second", "third"]
ipk_handler = TestIpkHandler(param_names_ordered)
time_handler = TimeHandler([0, 0, 0.1, 0.2, 0.3, 1])
meaner = meaner5.Meaner5(base_name="test", parhub_uri=parhub_uri, saving_interval=1, logger=logger, base_handler=db_handler, ipk_handler=ipk_handler, time_source=time_handler)
time_handler = TimeHandler([0, 0, 1, 2, 3, 10, 15, 20, 21, 25, 50, 65]) # time=50 should trigger shifting next probe save time
meaner = meaner5.Meaner5(base_name="test", parhub_uri=parhub_uri, saving_interval=10, logger=logger, base_handler=db_handler, ipk_handler=ipk_handler, time_source=time_handler)
server = Process(target=run_server)
server.start()
try:
......@@ -96,10 +134,10 @@ class SimpleTest(unittest.TestCase):
server.terminate() # TODO less violent termination, that would close zmq gracefully?
param_names = [heartbeat_param_name] + param_names_ordered
self.assertEqual(set(param_names), set(db.params.keys()))
expected_store_no = 1
expected_store_no = 3
for key in db.params.keys():
self.assertEqual(expected_store_no, len(db.params[key]))
expected_timestamp = 1
expected_timestamp = 10
for param_name in param_names:
self.assertEqual(expected_timestamp, db.params[param_name][0][0])
self.assertEqual(True, db.params["zero"][0][1].is_no_data)
......@@ -113,3 +151,29 @@ class SimpleTest(unittest.TestCase):
self.assertEqual(0, db.params["first"][0][1].prec)
self.assertEqual(1, db.params["second"][0][1].prec)
self.assertEqual(0, db.params[heartbeat_param_name][0][1].prec)
expected_timestamp = 20
for param_name in param_names:
self.assertEqual(expected_timestamp, db.params[param_name][1][0])
self.assertEqual(True, db.params["zero"][1][1].is_no_data)
self.assertEqual(True, db.params["first"][1][1].is_no_data)
self.assertEqual(False, db.params["second"][1][1].is_no_data)
self.assertEqual(False, db.params["third"][1][1].is_no_data)
self.assertEqual(False, db.params[heartbeat_param_name][1][1].is_no_data)
self.assertEqual(16, db.params["second"][1][1].value_before_prec)
self.assertEqual(17, db.params["third"][1][1].value_before_prec)
self.assertEqual(1, db.params[heartbeat_param_name][1][1].value_before_prec)
self.assertEqual(1, db.params["second"][1][1].prec)
self.assertEqual(2, db.params["third"][1][1].prec)
self.assertEqual(0, db.params[heartbeat_param_name][1][1].prec)
expected_timestamp = 60
for param_name in param_names:
self.assertEqual(expected_timestamp, db.params[param_name][2][0])
self.assertEqual(True, db.params["zero"][2][1].is_no_data)
self.assertEqual(False, db.params["first"][2][1].is_no_data)
self.assertEqual(True, db.params["second"][2][1].is_no_data)
self.assertEqual(True, db.params["third"][2][1].is_no_data)
self.assertEqual(False, db.params[heartbeat_param_name][2][1].is_no_data)
self.assertEqual(50, db.params["first"][2][1].value_before_prec)
self.assertEqual(1, db.params[heartbeat_param_name][2][1].value_before_prec)
self.assertEqual(0, db.params["first"][2][1].prec)
self.assertEqual(0, db.params[heartbeat_param_name][2][1].prec)