368 lines
14 KiB
Python
368 lines
14 KiB
Python
import requests
|
|
import time
|
|
from functools import partial
|
|
from typing import Any
|
|
import json
|
|
from tqdm import tqdm
|
|
import os
|
|
from settings import USERNAME,PASSWORD,BASE_URL
|
|
|
|
def delay(func,sleep_time=0.1):
|
|
"""
|
|
decorator that delays any function
|
|
|
|
@partial(delay,sleep_time=1)
|
|
def func(*args,**kwargs):
|
|
..
|
|
return ..
|
|
"""
|
|
|
|
def wrapper(*args,**kwargs):
|
|
time.sleep(sleep_time)
|
|
return func(*args,**kwargs)
|
|
return wrapper
|
|
|
|
|
|
class DataScraper():
|
|
def __init__(self,username:str,password:str,base_url:str) -> None:
|
|
self.base_url = base_url
|
|
self.cookie,self.session_expires = self.login(username,password)["response"]
|
|
|
|
def response_formatting(self,response:Any,successfull:bool) -> dict:
|
|
return {"response":response,"successfull":successfull}
|
|
|
|
@partial(delay,sleep_time=0.1)
|
|
def refresh_session(self) -> None:
|
|
"""
|
|
refreshes the expire date of the current token (so no need to update cookie, as the token will be valid for longer)
|
|
"""
|
|
requests.post(f"{self.base_url}/refresh",cookies=self.cookie)
|
|
|
|
|
|
@partial(delay,sleep_time=0.1)
|
|
def login(self,username:str,password:str) -> dict:
|
|
"""
|
|
signs in and returns the cookie (with the sesson-token).
|
|
"""
|
|
json_obj:dict = {"username":username,"password":password}
|
|
|
|
response = requests.post(f"{self.base_url}/login",json=json_obj)
|
|
|
|
if(not response.ok):
|
|
print(f"response:{response}")
|
|
print(f"response.content:{response.content}")
|
|
print(f"response.cookies:{response.cookies}")
|
|
exit("couldn't successfully login")
|
|
|
|
session_expires = json.loads(response.content.decode("utf-8"))["expire"]
|
|
return self.response_formatting((response.cookies,session_expires),True)
|
|
|
|
@partial(delay,sleep_time=0.1)
|
|
def get_all_sensor_names(self) -> tuple[list[str],bool]:
|
|
response = requests.get(f"{self.base_url}/api/sensor/getallnames",cookies=self.cookie)
|
|
|
|
if(response.ok):
|
|
#only return the names
|
|
return self.response_formatting([elem["name"] for elem in json.loads(response.content.decode("utf-8"))["names"]],True)
|
|
else:
|
|
return self.response_formatting(response.content,False)
|
|
|
|
|
|
@partial(delay,sleep_time=0.1)
|
|
def get_data(self,name:str,history_length:int,bin_size:int=1) -> dict:
|
|
"""
|
|
History length in days
|
|
Bin size is used to reduce the amount of data that is send, it would bin the data in bins of n minutes and only send over the average within the bin.
|
|
"""
|
|
|
|
json_obj = {"name":name,"binsize":bin_size,"length":history_length}
|
|
|
|
response = requests.get(f"{self.base_url}/api/sensor/getdata",params=json_obj,cookies=self.cookie)
|
|
|
|
if(response.ok):
|
|
return self.response_formatting(response.content, True)
|
|
else:
|
|
return self.response_formatting(response.content, False)
|
|
|
|
@partial(delay,sleep_time=0.1)
|
|
def get_info_sensor(self,name:str) -> dict:
|
|
"""
|
|
returns information related to the sensor
|
|
"createdAt", moment the sensor is created in db
|
|
"fw_version", #firmware version
|
|
"imei", #unique indentifier
|
|
"is_active", #if the sensor is active or not
|
|
"latitude", #where it's placed geographically
|
|
"longitude",
|
|
"name", #unique name
|
|
"notes",
|
|
"project_id", #id of the project it belongs to
|
|
"status", #status code
|
|
"type", #type sensor BS1000 of BS1400, etc..
|
|
"updatedAt" #last update of content in the database
|
|
"""
|
|
|
|
json_obj = {"name":name}
|
|
|
|
response = requests.get(f"{self.base_url}/api/sensor/get",params=json_obj,cookies=self.cookie)
|
|
|
|
if(response.ok):
|
|
return self.response_formatting(response.content,True)
|
|
else:
|
|
return self.response_formatting(response.content,False)
|
|
|
|
@partial(delay,sleep_time=0.1)
|
|
def get_all_sensors(self):
|
|
"""
|
|
Retrieves a list of all sensors in the database. (the information retrieved is information from the sensor)
|
|
"""
|
|
|
|
response = requests.get(f"{self.base_url}/api/sensor/getall",cookies=self.cookie)
|
|
|
|
if(response.ok):
|
|
return self.response_formatting(response.content,True)
|
|
else:
|
|
return self.response_formatting(response.content,False)
|
|
|
|
@partial(delay,sleep_time=0.1)
|
|
def get_predictions(self,name:str):
|
|
"""
|
|
name: sensor name
|
|
|
|
Fetches predictions from the database based on optional query parameters for sensor name, time start, and time end. If parameters are not provided, it retrieves all predictions.
|
|
"""
|
|
json_obj = {"sensorname":name}
|
|
|
|
response = requests.get(f"{self.base_url}/api/learning/prediction/get",params=json_obj,cookies=self.cookie)
|
|
|
|
if(response.ok):
|
|
return self.response_formatting(response.content,True)
|
|
else:
|
|
return self.response_formatting(response.content,False)
|
|
|
|
@partial(delay,sleep_time=0.1)
|
|
def get_labels(self,name:str):
|
|
"""
|
|
name: sensorname
|
|
|
|
Retrieves classification labels for a specific sensor based on the sensorname query parameter.
|
|
|
|
"""
|
|
json_obj = {"sensorname":name}
|
|
response = requests.get(f"{self.base_url}/api/classification/labels/get",params=json_obj,cookies=self.cookie)
|
|
|
|
if(response.ok):
|
|
return self.response_formatting(response.content,True)
|
|
else:
|
|
return self.response_formatting(response.content,False)
|
|
|
|
@partial(delay,sleep_time=0.1)
|
|
def get_all_sensor_IMEIs(self) -> list[str]:
|
|
response = requests.get(f"{self.base_url}/api/sensor/getall",cookies=self.cookie)
|
|
|
|
if(response.ok):
|
|
|
|
sensor_dict = json.loads(response.content.decode("utf-8"))
|
|
|
|
|
|
EMEIs = [sensor_dict[key]["imei"] for key in sensor_dict.keys()]
|
|
|
|
return self.response_formatting(EMEIs,True)
|
|
else:
|
|
return self.response_formatting(response.content,False)
|
|
|
|
@partial(delay,sleep_time=0.1)
|
|
def getLabeled_data(self,sensor_name:str,history_length:int=365,bin_size:int=1):
|
|
json_obj = {"name":sensor_name,"binsize":bin_size,"length":history_length}
|
|
|
|
response = requests.get(f"{self.base_url}/api/sensor/getlabeleddata",cookies=self.cookie,params=json_obj)
|
|
if(response.ok):
|
|
return self.response_formatting(response.content,True)
|
|
else:
|
|
return self.response_formatting(response.content,False)
|
|
|
|
class Info_storage():
|
|
def __init__(self,scraper:DataScraper,data_folder_name:str = "data") -> None:
|
|
self.scraper = scraper
|
|
self.sensor_IMEIs = [] #list of imeis
|
|
self.sensor_info = [] #list of {imei:name,sensor_info:sensor_info}
|
|
self.sensor_names = [] #list of names
|
|
self.sensor_data = [] #list of {imei:sensor_data}
|
|
self.sensor_predictions = [] #list of {name:sensor_predictions}
|
|
self.sensor_classification_labels = [] #list of {name:sensor_prediction_labels}
|
|
self.sensor_data_labeled = [] #list of {sensor_name:sensor_data}
|
|
|
|
self.data_folder_name = data_folder_name
|
|
self.data_folder_path = os.path.join(os.curdir,self.data_folder_name)
|
|
|
|
|
|
def write_to_file(self,file_name:str,jsonifiable_data:Any) -> None:
|
|
"""
|
|
writes jsonifiable data to a file
|
|
"""
|
|
with open(os.path.join(self.data_folder_path,file_name),"wt") as f:
|
|
try:
|
|
obj = json.dumps(jsonifiable_data) #str(json.loads(str(jsonifiable_data).strip("'<>() ").replace('\'', '\"').replace('\0', '').replace("True","true").replace("False","false")))
|
|
f.write(obj)
|
|
|
|
except Exception as e:
|
|
print(e)
|
|
print(f"CANT SAVE: {self.sensor_names}")
|
|
|
|
|
|
def read_from_file(self,file_name:str) -> Any:
|
|
sensor_names_path = os.path.join(self.data_folder_path,file_name)
|
|
with open(sensor_names_path) as f:
|
|
return json.loads(f.readline().replace("'",'"'))
|
|
|
|
def retrieve(self,use_imei:bool=True) -> None:
|
|
"""
|
|
new retrieve data from server
|
|
"""
|
|
#get sensor names
|
|
self.sensor_names = self.scraper.get_all_sensor_names()["response"]
|
|
self.sensor_IMEIs = self.scraper.get_all_sensor_IMEIs()["response"]
|
|
|
|
#get sensor info
|
|
sensor_list = self.sensor_names
|
|
if(use_imei):
|
|
sensor_list = self.sensor_IMEIs
|
|
|
|
|
|
for index,name in enumerate(tqdm(sensor_list,ascii=True,desc="get_sensor_info")):
|
|
resp = self.scraper.get_info_sensor(name)
|
|
if(resp["successfull"]):
|
|
sensor_info = json.loads(resp["response"].decode("utf-8"))
|
|
self.sensor_info.append({"name":name,"sensor_info":sensor_info})
|
|
else:
|
|
print(f"name: {name}, unsuccessfull: {resp}")
|
|
|
|
if(index % 100 == 0): #refresh session if it takes long
|
|
self.scraper.refresh_session()
|
|
|
|
#sensor data
|
|
for index,name in enumerate(tqdm(sensor_list,ascii=True,desc="sensor_data")):
|
|
resp = self.scraper.get_data(name,history_length=364,bin_size=1)
|
|
|
|
if(resp["successfull"]):
|
|
sensor_data = json.loads(resp["response"].decode("utf-8"))
|
|
self.sensor_data.append({"name":name,"sensor_data":sensor_data})
|
|
else:
|
|
print(f"unsuccessfull: {resp}")
|
|
|
|
if(index % 100 == 0):
|
|
self.scraper.refresh_session()
|
|
|
|
#predictions
|
|
for index,name in enumerate(tqdm(self.sensor_names,ascii=True,desc="predictions")):
|
|
resp = self.scraper.get_predictions(name)
|
|
if(resp["successfull"]):
|
|
sensor_pred = json.loads(resp["response"].decode("utf-8"))
|
|
self.sensor_predictions.append({"name":name,"sensor_pred":sensor_pred})
|
|
else:
|
|
print(f"unsuccessfull: {resp}")
|
|
|
|
if(index % 100 == 0):
|
|
self.scraper.refresh_session()
|
|
|
|
|
|
#classification labels
|
|
for index,name in enumerate(tqdm(self.sensor_names,ascii=True,desc="classification_labels")):
|
|
resp = self.scraper.get_predictions(name)
|
|
if(resp["successfull"]):
|
|
sensor_classification_labels = json.loads(resp["response"].decode("utf-8"))
|
|
self.sensor_classification_labels.append({"name":name,"sensor_classification_labels":sensor_classification_labels})
|
|
else:
|
|
print(f"unsuccessfull: {resp}")
|
|
|
|
if(index % 100 == 0):
|
|
self.scraper.refresh_session()
|
|
|
|
for index,name in enumerate(tqdm(self.sensor_names,ascii=True,desc="labeled data")):
|
|
resp = self.scraper.getLabeled_data(name)
|
|
if(resp["successfull"]):
|
|
labeled_data = json.loads(resp["response"].decode("utf-8"))
|
|
self.sensor_data_labeled.append({"name":name,"labeled_data":labeled_data})
|
|
else:
|
|
print(f"unsuccessfull: {resp}")
|
|
|
|
if(index % 100 == 0):
|
|
self.scraper.refresh_session()
|
|
|
|
def save(self) -> None:
|
|
"""
|
|
saves all of the data to files
|
|
"""
|
|
if(not os.path.isdir(self.data_folder_name)):
|
|
os.makedirs(self.data_folder_path)
|
|
|
|
#store sensor names
|
|
self.write_to_file("sensor_names.txt",self.sensor_names)
|
|
|
|
#store sensor EMEI's
|
|
self.write_to_file("sensor_EMEIs.txt",self.sensor_IMEIs)
|
|
|
|
#store sensor info
|
|
self.write_to_file("sensor_info.txt",self.sensor_info)
|
|
|
|
#store sensor data
|
|
self.write_to_file("sensor_data.txt",self.sensor_data)
|
|
|
|
#store sensor predictions
|
|
self.write_to_file("sensor_predictions.txt",self.sensor_predictions)
|
|
|
|
#store sensor classification labels
|
|
self.write_to_file("sensor_classification_labels.txt",self.sensor_classification_labels)
|
|
|
|
#store labeled data
|
|
self.write_to_file("labeled_data.txt",self.sensor_data_labeled)
|
|
|
|
|
|
def load(self) -> None:
|
|
"""
|
|
load data from local
|
|
"""
|
|
#load sensor names
|
|
self.sensor_names = self.read_from_file("sensor_names.txt")
|
|
|
|
#store sensor EMEI's
|
|
self.sensor_IMEIs = self.read_from_file("sensor_EMEIs.txt")
|
|
|
|
#load sensor info
|
|
self.sensor_info = self.read_from_file("sensor_info.txt")
|
|
|
|
#load sensor data
|
|
self.sensor_data = self.read_from_file("sensor_data.txt")
|
|
|
|
#load sensor predictions
|
|
self.sensor_predictions = self.read_from_file("sensor_predictions.txt")
|
|
|
|
#load sensor classification labels
|
|
self.sensor_classification_labels = self.read_from_file("sensor_classification_labels.txt")
|
|
|
|
#load labeled data
|
|
self.sensor_data_labeled = self.read_from_file("labeled_data.txt")
|
|
|
|
|
|
"""
|
|
Since we got duplicates with the names, we will have to try to implement it using the IMEIs,
|
|
TODO currently no sensor data is coming through when using IMEIS, figure out why this is the case.
|
|
|
|
"""
|
|
|
|
|
|
|
|
|
|
if __name__ == "__main__":
|
|
USE_IMEIS = False #NOTE if you get by IMEIS it will not do the conversion and you'll only get the data in SV1,SV2,SV3,SV4,SV5,SV6 as it depends on the sensor what kind of data will be on SV_n. (the type of the sensor can be derived from the name, and is done so to do the conversion)
|
|
|
|
|
|
scraper = DataScraper(username=USERNAME,password=PASSWORD,base_url=BASE_URL)
|
|
|
|
storage = Info_storage(scraper)
|
|
|
|
storage.retrieve(use_imei=USE_IMEIS)
|
|
|
|
storage.save()
|
|
storage.load()
|
|
|