import argparse import binascii from email import header import textwrap import pika import json from furl import furl from eventstream.event_pb2 import Event from eventstream.data_pb2 import SensorModel from eventstream.status_pb2 import GatewayStatus, HostConnectionReport import google.protobuf.text_format import google.protobuf.json_format import requests from requests.auth import HTTPBasicAuth import mysql.connector import os from xml.dom import minidom from xml.sax.handler import ContentHandler import xml.sax from datetime import datetime from zoneinfo import ZoneInfo class SenceiveIOConsumer: def _internal_callback(self, stream_name, channel, method_frame, header_frame, body, array_sensores): #Get the stream offset stream_offset = header_frame.headers['x-stream-offset'] #Create a blank protocol buffers message of type Event event = Event() #Decode the message payload and populate the fields event.ParseFromString(body) if self.callback_fn(stream_name, stream_offset, event, array_sensores) == True: self.offsets[stream_name] = stream_offset channel.basic_ack(delivery_tag=method_frame.delivery_tag) def _make_callback(self, stream_name, array_sensores): # Set up a callback lambda for each stream which bakes in the # channel name and reference to self # Parameters to the lambda are dictated by the AMQP library # An alternative approach to distingushing streams when messages arrive # is to record the Consumer Tag returned from basic_consume; this tag # is supplied in the method_frame parameter # Creating the lambda is done in a separate function to avoid issues # related to late binding # https://docs.python-guide.org/writing/gotchas/#late-binding-closures def cb(channel, method_frame, header_frame, body): self._internal_callback(stream_name, channel, method_frame, header_frame, body, array_sensores) return cb def __init__(self, connection_string, callback_fn, stream_names, offsets_filename, start_at=None, array_sensores=None): parameters = pika.URLParameters(connection_string) self.connection = pika.BlockingConnection(parameters) self.channel = self.connection.channel() self.array_sensores = array_sensores self.channel.basic_qos( prefetch_count=500 #500 message prefetch limit ) self.callback_fn = callback_fn self.offsets_filename = offsets_filename self.offsets = {} # Consume from each of the available streams for n in stream_names: # The x-stream-offset argument controls where in the stream we # start reading from consume_args = None if start_at is not None: consume_args = {'x-stream-offset': start_at} elif n in self.offsets: consume_args = {'x-stream-offset': self.offsets[n] + 1} self.channel.basic_consume(n, self._make_callback(n,self.array_sensores), arguments=consume_args) def consume_until_keyboard_interrupt(self): try: self.channel.start_consuming() except KeyboardInterrupt: try: self.channel.stop_consuming() except pika.exceptions.AMQPError: pass def close(self): try: self.connection.close() except: pass def insert_mysql(sensor_id, id_lectura, tiempo, data): cadena_sql_general = "insert into lecturas (sensor_id,id_lectura,fecha,lectura) values('" + str(sensor_id) + "'," + str(id_lectura) + ",'" + tiempo + "'," + str(data) + ");" conexion_mysql = mysql.connector.connect(user="sincroIns", password="teknes", host="192.168.1.110", database="campo_bolueta", port="3306") cursor = conexion_mysql.cursor() cursor.execute(cadena_sql_general) conexion_mysql.commit() conexion_mysql.close() def print_message_min(stream_name, stream_offset, event, array_sensores): event_type = event.WhichOneof('event_data') or 'Unknown' # DATO if event_type == 'data_sample': datos_evento = event.data_sample tiempo = event.event_time.ToJsonString() mac_sensor_recibido = event.source_device_id nombre_canal = SensorModel.Name(datos_evento.sensor_model) data = datos_evento.data try: dt = datetime.strptime(tiempo, "%Y-%m-%dT%H:%M:%SZ") except: dt = datetime.strptime(tiempo, "%Y-%m-%dT%H:%M:%S.%fZ") dt = dt.replace(tzinfo=ZoneInfo("UTC")) dt_madrid = dt.astimezone(ZoneInfo("Europe/Madrid")) tiempo = dt_madrid.strftime("%Y-%m-%d %H:%M:%S") #print(str(event.event_time.ToJsonString()) + " " + str(event.source_device_id) + " " + str(SensorModel.Name(datos_evento.sensor_model)) + " " + str(datos_evento.data) + " " + str(datos_evento.units)) id_lectura = 0 for x in array_sensores: teknes_sensor = x.get("teknes") senceive_sensor_mac = x.get("senceive") if senceive_sensor_mac == mac_sensor_recibido: id_lectura = id_lectura+1 if x.get("teknes").find("DIST")!= -1: if nombre_canal == "OdsDistance": print(x.get("teknes")) sensor_id = x.get("teknes") data = data/1000 insert_mysql(sensor_id, id_lectura, tiempo, data) elif nombre_canal == "InclinometerHighResX": sensor_id = x.get("teknes").replace("DIST","CLIN") print(sensor_id + "_X") sensor_id = sensor_id + "_X" insert_mysql(sensor_id, id_lectura, tiempo, data) elif nombre_canal == "InclinometerHighResY": sensor_id = x.get("teknes").replace("DIST","CLIN") print(sensor_id + "_Y") sensor_id = sensor_id + "_Y" insert_mysql(sensor_id, id_lectura, tiempo, data) elif nombre_canal == "InclinometerHighResZ": sensor_id = x.get("teknes").replace("DIST","CLIN") print(sensor_id + "_Z") sensor_id = sensor_id + "_Z" insert_mysql(sensor_id, id_lectura, tiempo, data) else: if nombre_canal == "InclinometerHighResX": print(x.get("teknes") + "_X") sensor_id = x.get("teknes") + "_X" insert_mysql(sensor_id, id_lectura, tiempo, data) elif nombre_canal == "InclinometerHighResY": print(x.get("teknes") + "_Y") sensor_id = x.get("teknes") + "_Y" insert_mysql(sensor_id, id_lectura, tiempo, data) elif nombre_canal == "InclinometerHighResZ": print(x.get("teknes") + "_Z") sensor_id = x.get("teknes") + "_Z" insert_mysql(sensor_id, id_lectura, tiempo, data) elif event_type == 'gateway_status': e = event.gateway_status sensor_id = "Gateway_bateria" id_lectura = 1 tiempo = event.event_time.ToJsonString() data = e.main_battery_v try: dt = datetime.strptime(tiempo, "%Y-%m-%dT%H:%M:%SZ") except: dt = datetime.strptime(tiempo, "%Y-%m-%dT%H:%M:%S.%fZ") dt = dt.replace(tzinfo=ZoneInfo("UTC")) dt_madrid = dt.astimezone(ZoneInfo("Europe/Madrid")) tiempo = dt_madrid.strftime("%Y-%m-%d %H:%M:%S") print(sensor_id) insert_mysql(sensor_id, id_lectura, tiempo, data) return True def main(): username = 'senceive_io.6901cd1f16e0673062ec4721' password = 'p0h9BodsKe6WfozRdsMkqbN2tdU6drQDtnZGLQ9zDNz0dwnV' home_uri = 'https://senceive.io' start_at = None offsets_file = 'resume.json' # Cargar configuracion para el cambio de nombre de los sensores array_sensores = [] f = open("/home/teknes/senceive-teknes/sensores.txt", "r") for x in f: y = x.strip() dotpos = y.find(";") sensor = y[:dotpos] nombre_senceive = sensor sensor = y[dotpos+1:] nombre_teknes = sensor array_sensores.append({"teknes":nombre_teknes,"senceive":nombre_senceive}) f.close() # Log in to the system and retrieve the Consumer Account details, # including the available streams consumer_info_url = furl(home_uri) consumer_info_url.path = 'api/consumers/current' request = requests.get(consumer_info_url.url, auth=HTTPBasicAuth(username, password)) if request.status_code != 200: print('ERROR: Unable to retrieve Consumer Account Details, status={}'.format(request.status_code)) return account = request.json() # Insert the credentials into the connection string as the API will not # include the password in any connection string apart from the one issued # on first creation connectionString = account['connectionString'] cs_url = furl(connectionString) cs_url.password = password connectionString = cs_url.url # Display Consumer Account details including list of available streams print('Consumer Account: {0[_id]} ({0[name]})'.format(account)) webui_url = furl(home_uri) webui_url.path = 'deviceGroups/{0[deviceGroup]}/consumers/{0[_id]}'.format(account) print(' ' + webui_url.url) print('Broker Connection String: {}'.format(connectionString)) print() print('{:49} {:25} {:30}'.format('Stream Identifier', 'Tenant Name', 'Device Group Name')) for s in account['streams']: print('{0[stream]:49} {0[tenantName]:25} {0[deviceGroupName]:30}'.format(s)) print() callback = print_message_min stream_names = [s['stream'] for s in account['streams']] consumer = SenceiveIOConsumer( connectionString, callback, stream_names, offsets_file, start_at, array_sensores) consumer.consume_until_keyboard_interrupt() consumer.close() if __name__ == '__main__': main()