Hello Ubidots Community!
From now on, we will share some of the solutions we provide to our users as they may be useful to you as well.
In this case, we want to share a solution to calculate a weekly average of a variable’s data by filtering the values based on 2 conditions:
- The values are within a given hour range
- Values are greater than zero
To achieve this, the UbiFunction below determines the start and end timestamps of the current week, retrieves the variable data in this range, filters it by the 2 conditions mentioned above, and then averages it.
Please see and try the code below and feel free to leave us your comments and questions on this forum about it.
NOTE: you need to replace de following constants with your information (the specific data of your account):
-
devices: The list of devices from which you want to get data. -
VAR_LABEL: The label of the variable from wich you want to get de data -
TOKEN: Temporary and revocable keys to be used in your API requests. -
TIMEZONE: Your timezone or the timezone you prefer for the solution -
HOURS: The hours to filter the data.
#------------- REPLACE CONSTS WITH YOUR INFO --------
devices = ['YOUR-DEVICES-LABELS (LIST)']
VAR_LABEL = 'YOUR-VARIABLE-LABEL'
TOKEN = 'YOUR-UBIDOTS-TOKEN'
TIMEZONE = 'YOUR-TIMEZONE'
HOURS = ['SELECTED-HOURS-TO-FILTER (LIST)'] #0,1,2,3...
#----------------------------------------------------
CODE:
#------------- IMPORT LIBRARIES ----------------------
import requests
import time
import json
import urllib
import math
import pytz
from datetime import datetime as dt
from datetime import timedelta as td
#----------------------------------------------------
#------------- REPLACE CONSTS WITH YOUR INFO --------
devices = ['YOUR-DEVICES-LABELS (LIST)']
VAR_LABEL = 'YOUR-VARIABLE-LABEL'
TOKEN = 'YOUR-UBIDOTS-TOKEN'
TIMEZONE = 'YOUR-TIMEZONE' # As in pytz.all_timezones
HOURS = ['SELECTED-HOURS-TO-FILTER (LIST)'] #0,1,2,3...
#----------------------------------------------------
BASE_URL = "https://industrial.api.ubidots.com"
now = dt.now(tz=pytz.timezone(TIMEZONE))
headers= {"X-Auth-Token":TOKEN,"Content-Type":"application/json"}
#------------- MAIN FUNCTION ------------------------
def main(args):
init, final = get_week_timerange()
now = dt.now()
ts_ejecution_function = int(dt.timestamp(now)*1000)
print("Timestamp of the fuction ejecution: {}".format(ts_ejecution_function))
if ts_ejecution_function + 60000 >= final:
print("INFO HOURS {}".format(HOURS))
for device in devices:
values = retrieve_single_variable_values(TOKEN, device, VAR_LABEL, init, final)
payload = filter_hour_value(values, HOURS)
print("[INFO] Payload to send: {}".format(payload))
if payload != None:
response = update_device(device, payload, TOKEN)
print("[INFO] POST request result:")
print(response.status_code)
else:
print("[ERROR] Average calculation was not executed")
response = None
return {"status": "averages generated and sent"}
return {"status": "Ok"}
#----------------------------------------------------
#--------- UPDATE DEVICE WITH THE AVERAGE VALUE(POST TO UBIDOTS)-------------
def update_device(device, payload, token):
url = "{}/api/v1.6/devices/{}".format(BASE_URL, device)
headers = {"X-Auth-Token": token, "Content-Type": "application/json"}
response = create_request(url, headers, payload, attempts=5, request_type="post")
return response
#----------------------------------------------------
#--------- CREATE A REQUEST TO THE SERVER -----------
def create_request(url, headers, data, attempts, request_type):
request_func = getattr(requests, request_type)
kwargs = {"url": url, "headers": headers}
if request_type == "post":
kwargs["json"] = data
try:
req = request_func(**kwargs)
print("[INFO] Request result: {}".format(req.text))
status_code = req.status_code
time.sleep(1)
while status_code >= 400 and attempts < 5:
req = request_func(**kwargs)
print("[INFO] Request result: {}".format(req.text))
status_code = req.status_code
attempts += 1
time.sleep(1)
return req
except Exception as e:
print("[ERROR] There was an error with the request, details:")
print(e)
return None
#----------------------------------------------------
#--------- GET WEEK TIMESTAMP RANGE --------------------
def get_week_timerange():
now_shifted = now.replace(hour=0, minute=0, second=0, microsecond=0)
day_week = now.weekday()
delta = td(days=day_week)
delta2 = td(days=6-day_week, hours=23, minutes=59, seconds=59)
init_ts = int((now_shifted - delta).timestamp()*1000)
final_ts = int((now_shifted + delta2).timestamp()*1000)
print(init_ts, final_ts)
return init_ts, final_ts
#----------------------------------------------------
#--------- FILTER DATA BY HOURS RANGE ---------------
def filter_hour_value(data, HOURS):
result_list = list()
curr_timestamp = int(dt.timestamp(now)*1000)
list_values = list()
results = data['results']
print("[INFO] Weekly data (Qty): {}".format(len(results)))
result_list = [result for result in results if result['value'] > 0.0]
print("[INFO] Data greater than zero {}".format(len(result_list)))
for result in result_list:
timestamp = result['timestamp']
dt_object = dt.fromtimestamp(int(timestamp/1000), tz=pytz.timezone(TIMEZONE))
hora = dt_object.hour
if hora in HOURS:
list_values.append(float(result['value']))
print("[INFO] Filtered data (by hours range): {}".format(len(list_values)))
print("[INFO] Values", list_values)
try:
prom = round(sum(list_values)/len(list_values),2)
payload = {"prom":{"value":prom, "timestamp":curr_timestamp}}
return payload
except Exception as e:
print("[ERROR] Error with the request, details:")
print(e)
return None
#----------------------------------------------------
#--------- GET SINGLE VARIABLE VALUES ---------------
def retrieve_single_variable_values(token, device_label, variable_label, init, final):
url= "{}/api/v1.6/devices/{}/{}/values/".format(BASE_URL, device_label, variable_label)
headers= {"X-Auth-Token":token,"Content-Type":"application/json"}
params = {"start":init, "end":final, "page_size":3000}
url = url + "?" + urllib.parse.urlencode(params)
response = create_request(url, headers, data=None, attempts=5, request_type="get")
print("[INFO] GET request results {}".format(response.status_code))
data = response.json()
return data
#----------------------------------------------------