86 lines
2.7 KiB
Python
Executable File
86 lines
2.7 KiB
Python
Executable File
#!/usr/bin/python3
|
|
|
|
import sys
|
|
import os
|
|
import pika
|
|
import logging
|
|
import csv
|
|
import re
|
|
|
|
from asebat.timefmt import timestamp_fmt as ts
|
|
from asebat.timefmt import date_refmt as df
|
|
from asebat.config import set_config as setting
|
|
|
|
|
|
class sqlraw():
|
|
def __init__(self, db, table):
|
|
self.db = db
|
|
self.table = table
|
|
self.fdata = 1
|
|
self.sql = (
|
|
"INSERT IGNORE INTO " + db + "." + table +
|
|
" (`UnitName`,`ToolName`,`eventDT`,`BatteryLevel`,`Temperature`,`NodeNum`,"
|
|
+ "`Val0`,`Val1`,`Val2`,`Val3`,`Val4`,`Val5`,`Val6`,`Val7`," +
|
|
"`Val8`,`Val9`,`ValA`,`ValB`,`ValC`,`ValD`,`ValE`,`ValF`) VALUES ")
|
|
|
|
def add_data(self, values):
|
|
if not self.fdata:
|
|
self.sql += ","
|
|
self.fdata = 0
|
|
self.sql += "(" + values + "),"
|
|
|
|
|
|
def callback(ch, method, properties, body): #body è di tipo byte
|
|
logging.info("PID {0:>5} >> Read message {1}".format(os.getpid(), body.decode("utf-8")))
|
|
msg = body.decode("utf-8").split(";")
|
|
sql = sqlraw('ase', 'rawdata')
|
|
commonData = '"{0}","{1}"'.format(msg[1],msg[2])
|
|
with open(msg[6], "r") as csvfile:
|
|
lines = csvfile.read().splitlines()
|
|
for line in lines:
|
|
fields = line.split(";|;")
|
|
if (md := re.match(
|
|
r"^(\d\d\d\d\/\d\d\/\d\d\s\d\d:\d\d:\d\d);(.+);(.+)$",
|
|
fields[0]
|
|
)):
|
|
rowData = ',"{0}",{1},{2}'.format(md.group(1), md.group(2), md.group(3))
|
|
nodeNum = 0
|
|
for field in fields[1:]:
|
|
if field == fields[0]:
|
|
continue
|
|
else:
|
|
nodeNum += 1
|
|
vals = field.split(";")
|
|
sql.add_data(commonData + rowData + ',{0},'.format(nodeNum) + ', '.join('"{0}"'.format(d) for d in vals))
|
|
print(sql.sql)
|
|
|
|
ch.basic_ack(delivery_tag=method.delivery_tag)
|
|
|
|
|
|
def main():
|
|
cfg = setting.config()
|
|
|
|
logging.basicConfig(
|
|
format="%(asctime)s %(message)s",
|
|
filename="/var/log/" + cfg.elablog,
|
|
level=logging.INFO,
|
|
)
|
|
|
|
parameters = pika.URLParameters('amqp://' + cfg.mquser + ':' + cfg.mqpass +
|
|
'@' + cfg.mqhost + ':5672/%2F')
|
|
connection = pika.BlockingConnection(parameters)
|
|
channel = connection.channel()
|
|
channel.queue_declare(queue=cfg.csv_queue, durable=True)
|
|
|
|
channel.basic_qos(prefetch_count=1)
|
|
channel.basic_consume(queue=cfg.csv_queue, on_message_callback=callback)
|
|
try:
|
|
channel.start_consuming()
|
|
except KeyboardInterrupt:
|
|
logging.info("PID {0:>5} >> Info: {1}.".format(
|
|
os.getpid(), "Shutdown requested...exiting"))
|
|
|
|
|
|
if __name__ == "__main__":
|
|
main()
|