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): if not url.endswith('/'): url += '/' self._url = url self._client = requests.Session() self._client.cert = (cert_file, key_file) self._client.verify = verifyCertificate if not verifyCertificate: requests.urllib3.disable_warnings() 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() 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() 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: 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)