import mimetypes import os import requests import json class SmartDataContextAPIClient: def __init__(self, cert_file, key_file, url="https://iot.lisha.ufsc.br/", verifyCertificate=True, logfile=None): if not url.endswith('/'): url += '/' self._url = url self._client = requests.Session() self._client.cert = (cert_file, key_file) self._client.verify = verifyCertificate self._logfile = logfile if not verifyCertificate: requests.urllib3.disable_warnings() def _log_request(self, method, url, headers, request_body, response_body): if self._logfile: with open(self._logfile, 'a') as log_file: log_file.write(f"URL: {method} {url}\n") log_file.write(f"Headers: {json.dumps(headers, indent=2)}\n") if request_body: log_file.write(f"Request Body: {json.dumps(request_body, indent=2)}\n") log_file.write(f"Response Body: {response_body}\n") log_file.write("\n" + "-"*50 + "\n") def _doJsonPost(self, json_data, return_attribute = '', endpoint="api/v1_1/context.php"): headers = { 'Content-Type': 'application/json' } url = self._url + endpoint response = self._client.post(url, headers=headers, json=json_data) try: response_json = response.json() self._log_request("POST", url, headers, json_data, response.json()) if "errors" in response_json: raise Exception(f"Error processing request: {json.dumps(response_json['errors'])}") else: if return_attribute: return response_json['result'][return_attribute] else: return response_json['result'] except Exception as e: raise Exception(f"Invalid response from API: {response.content} - {e}") def createSmartDataContext(self, content, features, t0=-1, t1=-1, smartDataSources=[], smartDataUnits=[]): json_request = { "command": "/create", "request": { "content": content, "features": features, "t0": t0, "t1": t1, "smartDataUnits": smartDataUnits, "smartDataSouces": smartDataSources } } return self._doJsonPost(json_request, 'smartDataContextId') def associateSmartDataContext(self, smartDataContextIds, smartDataUnits=[], smartDataSources=[]): if len(smartDataSources) == 0 and len(smartDataUnits) == 0: raise Exception("At least one smartDataSource or smartDataUnit must be informed") json_request = { "command": "/associate", "request": { "smartDataContextIds": smartDataContextIds, "smartDataUnits": smartDataUnits, "smartDataSources": smartDataSources } } result = self._doJsonPost(json_request) if len(result) == 1: return result[0] else: return result def unassociateSmartDataContext(self, smartDataContextIds, smartDataUnits=[], smartDataSources=[]): if len(smartDataSources) == 0 and len(smartDataUnits) == 0: raise Exception("At least one smartDataSource or smartDataUnit must be informed") json_request = { "command": "/unassociate", "request": { "smartDataContextIds": smartDataContextIds, "smartDataUnits": smartDataUnits, "smartDataSources": smartDataSources } } result = self._doJsonPost(json_request) if len(result) == 1: return result[0] else: return result def getSmartDataContext(self, smartDataContextId): json_request = { "command": "/get", "request": { "smartDataContextId": smartDataContextId } } return self._doJsonPost(json_request) def findSmartDataContext(self, smartDataUnits=[], smartDataSources=[], t0=None, t1=None): if len(smartDataSources) == 0 and len(smartDataUnits) == 0: raise Exception("At least one smartDataSource or smartDataUnit must be informed") json_request = { "command": "/contexts", "request": { "smartDataUnits": smartDataUnits, "smartDataSources": smartDataSources } } if t0: json_request['request']['t0'] = t0 if t1: json_request['request']['t1'] = t1 return self._doJsonPost(json_request) def querySmartDataContext(self, query): json_request = { "command": "/query", "request": query } return self._doJsonPost(json_request) def updateSmartDataContext(self, id, content=None, features=None, t0=None, t1=None, smartDataSources=None, smartDataUnits=None): json_request = { "command": "/update", "request": { "smartDataContextId": id } } if content: json_request['request']['content'] = content if features: json_request['request']['features'] = features if t0: json_request['request']['t0'] = t0 if t1: json_request['request']['t1'] = t1 if smartDataSources: json_request['request']['smartDataSources'] = smartDataSources if smartDataUnits: json_request['request']['smartDataUnits'] = smartDataUnits return self._doJsonPost(json_request) def addUnstructuredDataFromFile(self, smartDataContextId, filePath, fileName=None, mimeType=None): if not fileName: fileName = os.path.basename(filePath) if not mimeType: mimeType, _ = mimetypes.guess_type(filePath) with open(filePath, 'rb') as f: return self.addUnstructuredData(smartDataContextId, fileName, mimeType, f) def addUnstructuredData(self, smartDataContextId, fileName, mimeType, data): headers = { 'Content-Type': mimeType, 'Filename': fileName } url = self._url + f"api/v1_1/context.php?action=add-unstructured&smartDataContextId={smartDataContextId}" response = self._client.post(url, headers=headers, data=data) try: response_json = response.json() self._log_request("POST", url, headers, {"data": "binary data"}, response.json()) if "errors" in response_json: raise Exception(f"Error processing request: {json.dumps(response_json['errors'])}") else: return response_json['result']['objectId'] except Exception as e: raise Exception(f"Invalid response from API: {response.content} - {e}") def getUnstructuredData(self, smartDataContextId, objectId): headers = { } url = self._url + "api/v1_1/context.php" json_request = { "command": "/unstructured/get", "request": { "smartDataContextId": smartDataContextId, "objectId": objectId } } response = self._client.post(url, headers=headers, json=json_request) try: self._log_request("POST", url, dict(response.headers), json_request, "") response.raise_for_status() return response.content except Exception as e: raise Exception(f"Invalid response from API: {e}") def saveUnstructuredDataToFile(self, smartDataContextId, objectId, filePath): data = self.getUnstructuredData(smartDataContextId, objectId) with open (filePath, "wb") as f: f.write(data) def removeUnstructuredData(self, smartDataContextId, objectId): json_request = { "command": "/unstructured/remove", "request": { "smartDataContextId": smartDataContextId, "objectId": objectId } } return self._doJsonPost(json_request)