71 lines
2.6 KiB
Python
71 lines
2.6 KiB
Python
# Enter your python code.
|
|
import json, time
|
|
from datetime import datetime as dt
|
|
from common.Logger import logger
|
|
from quickfaas.remotebus import publish
|
|
|
|
def chunk_payload(payload, chunk_size=20, is_attributes_payload=False):
|
|
if is_attributes_payload:
|
|
# For attributes payload, chunk the controllers
|
|
controllers = list(payload.items())
|
|
for i in range(0, len(controllers), chunk_size):
|
|
yield dict(controllers[i:i + chunk_size])
|
|
else:
|
|
# For data payload, chunk the values within each controller
|
|
for controller, data in payload.items():
|
|
for entry in data:
|
|
ts = entry['ts']
|
|
values = entry['values']
|
|
chunked_values = list(values.items())
|
|
for i in range(0, len(chunked_values), chunk_size):
|
|
yield {
|
|
controller:[{
|
|
"ts": ts,
|
|
"values": dict(chunked_values[i:i + chunk_size])
|
|
}]
|
|
}
|
|
|
|
def sendData(message):
|
|
#logger.debug(message)
|
|
# Extract measures and group by ctrlName
|
|
grouped_data = {}
|
|
grouped_attributes = {}
|
|
now = (round(dt.timestamp(dt.now())/600)*600)*1000
|
|
for measure in message['measures']:
|
|
ctrlName = " ".join(measure['ctrlName'].split("_"))
|
|
name = measure['name']
|
|
value = measure['value']
|
|
health = measure['health']
|
|
#Add controller for telemetry if it doesn't exist
|
|
if ctrlName not in grouped_data:
|
|
grouped_data[ctrlName] = {}
|
|
#Add controller for attributes if it doesn't exist
|
|
if ctrlName not in grouped_attributes:
|
|
grouped_attributes[ctrlName] = {}
|
|
#Add data to temp payload if datapoint health is good
|
|
if health:
|
|
grouped_data[ctrlName][name] = value
|
|
grouped_attributes[ctrlName]["latestReportTime"] = now
|
|
#print(grouped_data)
|
|
# Transform the grouped data to desired structure
|
|
payload = {}
|
|
|
|
for key, value in grouped_data.items():
|
|
if value:
|
|
payload[key] = [{"ts": now ,"values": value}]
|
|
attributes_payload = {}
|
|
for key, value in grouped_attributes.items():
|
|
if value:
|
|
attributes_payload[key] = value
|
|
|
|
|
|
|
|
#logger.debug(payload)
|
|
for chunk in chunk_payload(payload=payload):
|
|
publish(__topic__, json.dumps(chunk), __qos__)
|
|
time.sleep(2)
|
|
|
|
for chunk in chunk_payload(payload=attributes_payload, is_attributes_payload=True):
|
|
publish("v1/gateway/attributes", json.dumps(attributes_payload), __qos__)
|
|
time.sleep(2)
|
|
|