257 lines
10 KiB
Python
257 lines
10 KiB
Python
# Enter your python code.
|
|
import json, os, time, shutil
|
|
from datetime import datetime as dt
|
|
from common.Logger import logger
|
|
from quickfaas.remotebus import publish
|
|
from quickfaas.global_dict import get as get_params
|
|
from quickfaas.global_dict import _set_global_args
|
|
|
|
|
|
|
|
def chunk_payload(payload, chunk_size=20):
|
|
if "values" in payload:
|
|
# Original format: {"ts": ..., "values": {...}}
|
|
chunked_values = list(payload["values"].items())
|
|
for i in range(0, len(chunked_values), chunk_size):
|
|
yield {
|
|
"ts": payload["ts"],
|
|
"values": dict(chunked_values[i:i+chunk_size])
|
|
}
|
|
else:
|
|
# New format: {"key1": "value1", "key2": "value2"}
|
|
chunked_keys = list(payload.keys())
|
|
for i in range(0, len(chunked_keys), chunk_size):
|
|
yield {k: payload[k] for k in chunked_keys[i:i+chunk_size]}
|
|
|
|
|
|
def sendData(message):
|
|
#logger.debug(message)
|
|
payload = {"ts": (round(dt.timestamp(dt.now())/600)*600)*1000, "values": {}}
|
|
attributes_payload = {}
|
|
resetPayload = {"ts": "", "values": {}}
|
|
dayReset, weekReset, monthReset, yearReset = False, False, False, False
|
|
for measure in message["measures"]:
|
|
try:
|
|
logger.debug(measure)
|
|
if measure["health"] == 1:
|
|
if "_spt" in measure["name"]:
|
|
attributes_payload[measure["name"]] = measure["value"]
|
|
elif measure in ["totalizer_1"]:
|
|
payload["values"]["day_volume"], dayReset = totalizeDay(measure["value"])
|
|
payload["values"]["week_volume"], weekReset = totalizeWeek(measure["value"])
|
|
payload["values"]["month_volume"], monthReset = totalizeMonth(measure["value"])
|
|
payload["values"]["year_volume"], yearReset = totalizeYear(measure["value"])
|
|
payload["values"][measure["name"]] = measure["value"]
|
|
except Exception as e:
|
|
logger.error(e)
|
|
|
|
|
|
for chunk in chunk_payload(payload=payload):
|
|
publish(__topic__, json.dumps(chunk), __qos__, cloud_name="FK Transfer")
|
|
time.sleep(2)
|
|
|
|
attributes_payload["latestReportTime"] = (round(dt.timestamp(dt.now())/600)*600)*1000
|
|
for chunk in chunk_payload(payload=attributes_payload):
|
|
publish("v1/devices/me/attributes", json.dumps(chunk), __qos__, cloud_name="FK Transfer")
|
|
time.sleep(2)
|
|
|
|
if dayReset:
|
|
resetPayload["values"]["yesterday_volume"] = payload["values"]["day_volume"]
|
|
resetPayload["values"]["day_volume"] = 0
|
|
if weekReset:
|
|
resetPayload["values"]["last_week_volume"] = payload["values"]["week_volume"]
|
|
resetPayload["values"]["week_volume"] = 0
|
|
if monthReset:
|
|
resetPayload["values"]["last_month_volume"] = payload["values"]["month_volume"]
|
|
resetPayload["values"]["month_volume"] = 0
|
|
if yearReset:
|
|
resetPayload["values"]["last_year_volume"] = payload["values"]["year_volume"]
|
|
resetPayload["values"]["year_volume"] = 0
|
|
|
|
if resetPayload["values"]:
|
|
resetPayload["ts"] = 1 + (round(dt.timestamp(dt.now())/600)*600)*1000
|
|
publish(__topic__, json.dumps(resetPayload), __qos__, cloud_name="FK Transfer")
|
|
|
|
|
|
def initialize_totalizers():
|
|
return {
|
|
"day": 0,
|
|
"week": 0,
|
|
"month": 0,
|
|
"year": 0,
|
|
"lifetime": 0,
|
|
"dayHolding": 0,
|
|
"weekHolding": 0,
|
|
"monthHolding": 0,
|
|
"yearHolding": 0
|
|
}
|
|
|
|
def getTotalizers(file_path="/var/user/files/totalizers.json"):
|
|
"""
|
|
Retrieves totalizer data from a JSON file.
|
|
|
|
:param file_path: Path to the JSON file storing totalizer data.
|
|
:return: Dictionary containing totalizer values.
|
|
"""
|
|
try:
|
|
with open(file_path, "r") as t:
|
|
totalizers = json.load(t)
|
|
if not totalizers or not isinstance(totalizers, dict):
|
|
logger.info("Invalid data format in the file. Initializing totalizers.")
|
|
totalizers = initialize_totalizers()
|
|
except FileNotFoundError:
|
|
logger.info("File not found. Initializing totalizers.")
|
|
totalizers = initialize_totalizers()
|
|
except json.JSONDecodeError:
|
|
timestamp = dt.now().strftime("%Y%m%d_%H%M%S")
|
|
# Split the file path and insert the timestamp before the extension
|
|
file_name, file_extension = os.path.splitext(file_path)
|
|
backup_file_path = f"{file_name}_{timestamp}{file_extension}"
|
|
shutil.copyfile(file_path, backup_file_path)
|
|
logger.error(f"Error decoding JSON. A backup of the file is created at {backup_file_path}. Initializing totalizers.")
|
|
totalizers = initialize_totalizers()
|
|
return totalizers
|
|
|
|
def saveTotalizers(totalizers, file_path="/var/user/files/totalizers.json"):
|
|
"""
|
|
Saves totalizer data to a JSON file.
|
|
|
|
:param totalizers: Dictionary containing totalizer values to be saved.
|
|
:param file_path: Path to the JSON file where totalizer data will be saved.
|
|
"""
|
|
try:
|
|
with open(file_path, "w") as t:
|
|
json.dump(totalizers, t)
|
|
except (IOError, OSError, json.JSONEncodeError) as e:
|
|
logger.error(f"Error saving totalizers to {file_path}: {e}")
|
|
raise # Optionally re-raise the exception if it should be handled by the caller
|
|
|
|
|
|
def totalizeDay(lifetime, max_retries=3, retry_delay=2):
|
|
"""
|
|
Update and save daily totalizers based on the lifetime value.
|
|
|
|
:param lifetime: The current lifetime total.
|
|
:param max_retries: Maximum number of save attempts.
|
|
:param retry_delay: Delay in seconds between retries.
|
|
:return: A tuple containing the calculated value and a boolean indicating if a reset occurred, or (None, False) if save fails.
|
|
"""
|
|
totalizers = getTotalizers()
|
|
now = dt.fromtimestamp(round(dt.timestamp(dt.now())/600)*600)
|
|
reset = False
|
|
value = lifetime - totalizers["dayHolding"]
|
|
|
|
if not int(now.strftime("%d")) == int(totalizers["day"]):
|
|
totalizers["dayHolding"] = lifetime
|
|
totalizers["day"] = int(now.strftime("%d"))
|
|
|
|
for attempt in range(max_retries):
|
|
try:
|
|
saveTotalizers(totalizers)
|
|
reset = True
|
|
return (value, reset)
|
|
except Exception as e:
|
|
logger.error(f"Attempt {attempt + 1} failed to save totalizers: {e}")
|
|
if attempt < max_retries - 1:
|
|
time.sleep(retry_delay)
|
|
else:
|
|
logger.error("All attempts to save totalizers failed.")
|
|
return (None, False)
|
|
|
|
return (value, reset)
|
|
|
|
def totalizeWeek(lifetime, max_retries=3, retry_delay=2):
|
|
"""
|
|
Update and save weekly totalizers based on the lifetime value.
|
|
|
|
:param lifetime: The current lifetime total.
|
|
:param max_retries: Maximum number of save attempts.
|
|
:param retry_delay: Delay in seconds between retries.
|
|
:return: A tuple containing the calculated value and a boolean indicating if a reset occurred, or (None, False) if save fails.
|
|
"""
|
|
totalizers = getTotalizers()
|
|
now = dt.fromtimestamp(round(dt.timestamp(dt.now())/600)*600)
|
|
reset = False
|
|
value = lifetime - totalizers["weekHolding"]
|
|
if (not now.strftime("%U") == totalizers["week"] and now.strftime("%a") == "Sun") or totalizers["week"] == 0:
|
|
totalizers["weekHolding"] = lifetime
|
|
totalizers["week"] = now.strftime("%U")
|
|
|
|
for attempt in range(max_retries):
|
|
try:
|
|
saveTotalizers(totalizers)
|
|
reset = True
|
|
return (value, reset)
|
|
except Exception as e:
|
|
logger.error(f"Attempt {attempt + 1} failed to save totalizers: {e}")
|
|
if attempt < max_retries - 1:
|
|
time.sleep(retry_delay)
|
|
else:
|
|
logger.error("All attempts to save totalizers failed.")
|
|
return (None, False)
|
|
return (value, reset)
|
|
|
|
def totalizeMonth(lifetime, max_retries=3, retry_delay=2):
|
|
"""
|
|
Update and save monthly totalizers based on the lifetime value.
|
|
|
|
:param lifetime: The current lifetime total.
|
|
:param max_retries: Maximum number of save attempts.
|
|
:param retry_delay: Delay in seconds between retries.
|
|
:return: A tuple containing the calculated value and a boolean indicating if a reset occurred, or (None, False) if save fails.
|
|
"""
|
|
totalizers = getTotalizers()
|
|
now = dt.fromtimestamp(round(dt.timestamp(dt.now())/600)*600)
|
|
reset = False
|
|
value = lifetime - totalizers["monthHolding"]
|
|
if not int(now.strftime("%m")) == int(totalizers["month"]):
|
|
totalizers["monthHolding"] = lifetime
|
|
totalizers["month"] = now.strftime("%m")
|
|
|
|
for attempt in range(max_retries):
|
|
try:
|
|
saveTotalizers(totalizers)
|
|
reset = True
|
|
return (value, reset)
|
|
except Exception as e:
|
|
logger.error(f"Attempt {attempt + 1} failed to save totalizers: {e}")
|
|
if attempt < max_retries - 1:
|
|
time.sleep(retry_delay)
|
|
else:
|
|
logger.error("All attempts to save totalizers failed.")
|
|
return (None, False)
|
|
|
|
return (value,reset)
|
|
|
|
def totalizeYear(lifetime, max_retries=3, retry_delay=2):
|
|
"""
|
|
Update and save yearly totalizers based on the lifetime value.
|
|
|
|
:param lifetime: The current lifetime total.
|
|
:param max_retries: Maximum number of save attempts.
|
|
:param retry_delay: Delay in seconds between retries.
|
|
:return: A tuple containing the calculated value and a boolean indicating if a reset occurred, or (None, False) if save fails.
|
|
"""
|
|
totalizers = getTotalizers()
|
|
now = dt.fromtimestamp(round(dt.timestamp(dt.now())/600)*600)
|
|
reset = False
|
|
value = lifetime - totalizers["yearHolding"]
|
|
if not int(now.strftime("%Y")) == int(totalizers["year"]):
|
|
totalizers["yearHolding"] = lifetime
|
|
totalizers["year"] = now.strftime("%Y")
|
|
|
|
for attempt in range(max_retries):
|
|
try:
|
|
saveTotalizers(totalizers)
|
|
reset = True
|
|
return (value, reset)
|
|
except Exception as e:
|
|
logger.error(f"Attempt {attempt + 1} failed to save totalizers: {e}")
|
|
if attempt < max_retries - 1:
|
|
time.sleep(retry_delay)
|
|
else:
|
|
logger.error("All attempts to save totalizers failed.")
|
|
return (None, False)
|
|
return (value, reset)
|
|
|