#import math import doctest import json import os # --- update the path of Python py_path = os.sys.path py_pwd = os.getcwd() if (py_pwd not in py_path): (os.sys.path).append(py_pwd) py_pwd = os.getcwd() + "/.." if (py_pwd not in py_path): (os.sys.path).append(py_pwd) import utils.celme as celme # ======================================================== # ======================================================== # === Status_json # ======================================================== # ======================================================== class Status_json: """ Class to manage Json file reports into python variables and vice versa Example a a very wimple Json file report contents: { "frame_model": "1.1", "name_creator": "PLC", "version_creator": "20180909", "entities": [ { "entity_name": "PLC_STATUS", "from": "Raspberry", "version_firmware": "20180924A", "site": "MyObservatory", "date": "2018-09-25T22:24:16.387", "devices": [ { "device_name": "DHT22", "device_type": "meteo", "serial_number": "1", "error_code": "0", "device_values": [ { "name": "OutsideTemperature", "type": "float", "value": "22", "unit": "degC", "monitoring_name": "temperature", "comment": "Temperature" } ] } ] } ] } A value is stored in a UFKI (Unique Four Key Identifier): UFKI = entity_name device_name serial_number name No support for hjson. See the following link to do that: https://github.com/hjson/hjson-py Usage: ------ First, instanciate an object from the class: report = report.Status_json() Informations: ------------- help(Status_json) Status_json().infos("doctest") Status_json().infos("doc_methods") """ # ======================================================== # === attributes # ======================================================== # --- multiplicators for units _status = [] _entity_name = "" _device_name = "" _serial_number = "" # ======================================================== # === internal methods # ======================================================== def _init(self,status=""): """ Initialize internal attributes. Inputs: ------- status is a list of dictionnaries according a predefined format. Usage: ------ report = report.Status_json() """ self._status = status self._entity_name = "" self._device_name = "" self._serial_number = "" def _is_number(self,s): """ Return True if the string is a number else return False. Usage: ------ >>> report = report.Status_json() >>> report._is_number("3e5") True >>> report._is_number("3a5") False """ try: float(s) return True except ValueError: pass try: import unicodedata unicodedata.numeric(s) return True except (TypeError, ValueError): pass return False #################################################################### # To read a Status #################################################################### def _get_indice_from_entities(self, entities, entity_name): indice = 0 found = -1 for entity in entities: #print("{} et {}".format(entity,entity_name)) if entity == entity_name: found = indice break indice += 1 return found def _get_indice_from_devices(self, devices, device_name, serial_number): indice = 0 found = -1 for device in devices: cur_device_name = device['device_name'] cur_serial_number = device['serial_number'] #print("{} et {}".format(device,device_name)) if (cur_device_name == device_name) and ( cur_serial_number == serial_number): found = indice break indice += 1 return found def _get_entity_names(self): #entities = self._status['statuses'][0]['entities'] entities = self._status['entities'] entity_names = [] for entity in entities: entity_names.append(entity['entity_name']) return entity_names def _get_device_nameserials(self, entity_name): entities = self._get_entity_names() indice = self._get_indice_from_entities(entities,entity_name) if indice == -1: return "" #devices = self._status['statuses'][0]['entities'][indice]['devices'] devices = self._status['entities'][indice]['devices'] device_names = [] for device in devices: devs = {'device_name':device['device_name'], 'serial_number':device['serial_number']} #print("devs={} ".format(devs)) device_names.append(devs) return device_names def _get_value_names(self, entity_name, device_name, serial_number): entities = self._get_entity_names() indicee = self._get_indice_from_entities(entities,entity_name) if indicee == -1: return "" #devices = self._status['statuses'][0]['entities'][indicee]['devices'] devices = self._status['entities'][indicee]['devices'] indiced = self._get_indice_from_devices(devices, device_name, serial_number) device = devices[indiced] #print("device={} ".format(device)) values = device['device_values'] value_names = [] for value in values: #print("value={} ".format(value)) value_names.append(value['name']) return value_names def _get_value_name_values(self, entity_name, device_name, serial_number, value_name): entities = self._get_entity_names() indicee = self._get_indice_from_entities(entities,entity_name) if indicee == -1: return "" #devices = self._status['statuses'][0]['entities'][indicee]['devices'] devices = self._status['entities'][indicee]['devices'] indiced = self._get_indice_from_devices(devices, device_name, serial_number) device = devices[indiced] #print("device={} ".format(device)) values = device['device_values'] for value in values: if value_name == value['name']: return value return "" def _set_value_name_values(self, entity_name, device_name, serial_number, value_name, dico): entities = self._get_entity_names() indicee = self._get_indice_from_entities(entities,entity_name) if indicee == -1: return "" #devices = self._status['statuses'][0]['entities'][indicee]['devices'] devices = self._status['entities'][indicee]['devices'] indiced = self._get_indice_from_devices(devices, device_name, serial_number) device = devices[indiced] #print("device={} ".format(device)) values = device['device_values'] indicev = 0 for value in values: if value_name == value['name']: #self._status['statuses'][0]['entities'][indicee]['devices'][indiced]['device_values'][indicev] = dico self._status['entities'][indicee]['devices'][indiced]['device_values'][indicev] = dico break indicev += 1 return "" #################################################################### # To create a Status #################################################################### def _status_new(self, name_creator, version_creator): status_header = {'frame_model':'1.1', 'name_creator':name_creator, 'version_creator':version_creator, 'entities':[]} #self._status = { 'statuses':[] } #self._status['statuses'].append(status_header) self._status = status_header def _status_new_entity(self, entity_name, thefrom, version_firmware, site, date=""): if date=="": date = (celme.Date("now")).iso() entity= {'entity_name':entity_name, 'from':thefrom, 'version_firmware':version_firmware, 'site':site, 'date':date, 'devices':[]} #self._status['statuses'][0]['entities'].append(entity); self._status['entities'].append(entity); def _status_entity_new_device(self, entity_name, device_name, device_type, serial_number, error_code): entities = self._get_entity_names() indicee = self._get_indice_from_entities(entities,entity_name) if indicee == -1: return "" device = {'device_name':device_name, 'device_type':device_type, 'serial_number':serial_number, 'error_code':error_code, 'device_values':[]} #self._status['statuses'][0]['entities'][indicee]['devices'].append(device) self._status['entities'][indicee]['devices'].append(device) def _status_entity_device_new_value(self, entity_name, device_name, serial_number, name, thetype, value, unit="", monitoring_name="", comment=""): entities = self._get_entity_names() indicee = self._get_indice_from_entities(entities,entity_name) if indicee == -1: return "" #devices = self._status['statuses'][0]['entities'][indicee]['devices'] devices = self._status['entities'][indicee]['devices'] indiced = self._get_indice_from_devices(devices, device_name, serial_number) if indiced == -1: return "" dico = {'name':name, 'type':thetype, 'value':value, 'unit':unit, 'monitoring_name':monitoring_name, 'comment':comment} #self._status['statuses'][0]['entities'][indicee]['devices'][indiced]['device_values'].append(dico) self._status['entities'][indicee]['devices'][indiced]['device_values'].append(dico) # ======================================================== # === status_json methods # ======================================================== def new_status(self, name_creator, version_creator): """ Initialize a new status Inputs: ------- Dictionary keys of the creator. Usage: ------ >>> report = report.Status_json() >>> report.new_status("PLC","20180909") Related topics: --------------- Method append_entity """ self._init() self._status_new(name_creator, version_creator) def append_entity(self, entity_name, thefrom, version_firmware, site, date=""): """ Append a new entity to a creator Inputs: ------- Dictionary keys of the entity. Usage: ------ >>> report = report.Status_json() >>> report.new_status("PLC","20180909") >>> rep.append_entity("PLC_STATUS", "Raspberry", "20180924A", "MyObservatory") Related topics: --------------- Methods append_device, get_entities """ self._status_new_entity(entity_name, thefrom, version_firmware, site, date) self._entity_name = entity_name def append_device(self, device_name, device_type, serial_number, error_code): """ Append a new device to an entity Inputs: ------- Dictionary keys of the device. Usage: ------ >>> report = report.Status_json() >>> report.new_status("PLC","20180909") >>> rep.append_entity("PLC_STATUS", "Raspberry", "20180924A", "MyObservatory") >>> rep.append_device("DHT22", "meteo", "1", "0") Related topics: --------------- Methods append_value, get_devices """ entity_name = self._entity_name self._status_entity_new_device(entity_name, device_name, device_type, serial_number, error_code) self._device_name = device_name self._serial_number = serial_number def append_value(self, name, thetype, value, unit="", monitoring_name="", comment=""): """ Append a new value to a device Inputs: ------- Dictionary keys of the device_value. Usage: ------ >>> report = report.Status_json() >>> report.new_status("PLC","20180909") >>> rep.append_entity("PLC_STATUS", "Raspberry", "20180924A", "MyObservatory") >>> rep.append_device("DHT22", "meteo", "1", "0") >>> rep.append_value( "OutsideTemperature", "float","22","degC","temperature","Temperature") Related topics: --------------- Methods append_value, get_devices """ entity_name = self._entity_name device_name = self._device_name serial_number = self._serial_number self._status_entity_device_new_value(entity_name, device_name, serial_number, name, thetype, value, unit, monitoring_name, comment) def load_json(self, full_filename): """ Load a Json file as an object Inputs: ------- full_filename is the path + file name of the Json file Usage: ------ >>> report = report.Status_json() >>> report.load_json("") Related topics: --------------- The Json file does not have any comment """ if (full_filename != ""): with open(full_filename, "r") as read_file: self._status = json.load(read_file) def save_json(self, full_filename): """ Save a Json file from an object Inputs: ------- full_filename is the path + file name of the Json file Usage: ------ >>> report = report.Status_json() >>> report.create_a_simple_report_for_tests() >>> report.save_json("") Related topics: --------------- The Json file will not have any comment """ if (full_filename != "") and (self._status != ""): json_string = json.dumps(self._status, indent=2) with open(full_filename, "w") as write_file: write_file.write(json_string) def create_a_simple_report_for_tests(self): """ Create a simple report for tests Inputs: ------- None. Usage: ------ >>> report = report.Status_json() >>> report.create_a_simple_report_for_tests() Related topics: --------------- The Json file will not have any comment """ self.new_status("PLC","20180909") self.append_entity("PLC_STATUS", "Raspberry", "20180924A", "MyObservatory") self.append_device("DHT22", "meteo", "1", "0") self.append_value( "OutsideTemperature", "float","22","degC","temperature","Temperature") #full_filename = "test.json" #self.save_json(full_filename) # ======================================================== # === get/set methods # ======================================================== def get_entities(self): """ Get the current entities Inputs: ------- None. Usage: ------ >>> report = report.Status_json() >>> print("{}",format(report.get_entities())) Related topics: --------------- """ entities = [] if self._status != "": entities = self._get_entity_names() return entities def get_devices(self, only_entity_name=""): """ Get the current devices Inputs: ------- None or a device name to check if exists. Usage: ------ >>> report = report.Status_json() >>> print("{}",format(report.get_devices())) Related topics: --------------- """ res = [] if self._status != "": entities = self._get_entity_names() for entity_name in entities: devices = [] if (only_entity_name==""): devices = self._get_device_nameserials(entity_name) else: if (only_entity_name==entity_name): devices = self._get_device_nameserials(entity_name) for device in devices: re = [entity_name, device['device_name'], device['serial_number']] res.append(re) return res def get_values(self, only_device_name="", only_serial_number="", getvalue=False): """ Get the current values Inputs: ------- None or a device name + serial number + value to check if exists. Usage: ------ >>> report = report.Status_json() >>> print("{}",format(report.get_devices())) Related topics: --------------- """ res = [] if self._status != "": entities = self._get_entity_names() for entity_name in entities: devices = self._get_device_nameserials(entity_name) for device in devices: device_name = device['device_name'] serial_number = device['serial_number'] value_names = [] if (only_device_name=="") or (only_serial_number==""): value_names = self._get_value_names(entity_name, device_name, serial_number) else: if (only_device_name==device_name) or (only_serial_number==serial_number): value_names = self._get_value_names(entity_name, device_name, serial_number) for value_name in value_names: if getvalue==False: re = [entity_name, device_name, serial_number, value_name] else: val = self._get_value_name_values(entity_name, device_name, serial_number, value_name) re = [entity_name, device_name, serial_number, value_name, val] res.append(re) return res # ======================================================== # === debug methods # ======================================================== def infos(self, action) -> None: """ To get informations about this class :param action: A command to run a debug action (see examples). :type action: string :Example: Status_json().infos("doctest") Status_json().infos("doc_methods") Status_json().infos("internal_attributes") Status_json().infos("public_methods") """ if (action == "doc_methods"): publics = [x for x in dir(self) if x[0]!="_"] for public in publics: varname = "{}".format(public) if (callable(getattr(self,varname))==True): print("\n{:=^40}".format(" method "+varname+" ")) t = "Angle()."+varname+".__doc__" tt =eval(t) print(tt) if (action == "doctest"): if __name__ == "__main__": print("\n{:~^40}".format("doctest")) #doctest.testmod(verbose=True, extraglobs={'objangle': Angle()}) doctest.testmod(verbose=True) if (action == "internal_attributes"): internals = [x for x in dir(self) if x[0]=="_" and x[1]!="_"] for internal in internals: varname = "{}".format(internal) #if (hasattr(self,varname)==True): if (callable(getattr(self,varname))==False): print(varname + "=" + str(getattr(self,varname))) if (action == "public_methods"): publics = [x for x in dir(self) if x[0]!="_"] for public in publics: varname = "{}".format(public) if (callable(getattr(self,varname))==True): print(varname) # ======================================================== # === special methods # ======================================================== def __init__(self, status=""): """ Object initialization where angle is the input in any format. :param angle : An angle in any supported format (cf. help(Angle)) :type angle : string """ self._init(status) # ======================================================== # ======================================================== # ======================================================== # Examples of execution if __name__ == "__main__": # --- update the path of Python if necessary import report # =============================== # --- Write a very sipmle Json # =============================== print("Make a new report") rep = report.Status_json() rep.new_status("PLC","20180909") rep.append_entity("PLC_STATUS", "Raspberry", "20180924A", "MyObservatory") # --- one device with one value rep.append_device("DHT22", "meteo", "1", "0") rep.append_value( "OutsideTemperature", "float","22","degC","Temperature_outside","Temperature") # --- List the UFKIs ufiks = rep.get_values("","",False) print(" List of all UFKIs:") for ufik in ufiks: print(" UFKI = {}".format(ufik)) # --- save the Json file rep.save_json("plc_verysimple.json") # =============================== # --- Write a more complex Json # =============================== print("Make a new report") rep = report.Status_json() rep.new_status("PLC","20180909") rep.append_entity("PLC_STATUS", "Raspberry", "20180924A", "Guitalens Observatory") # --- device Came rep.append_device("Came", "roof_controler", "124", "0") rep.append_value( "Error_code", "int","0","","","0=OK 1=PowerPB 2=SwitchPB") rep.append_value( "Power_input", "int","0","","","0=0Volt 1=24Volts") rep.append_value( "Mode_came", "int","0","","","0=Manual 1=Auto") rep.append_value( "Roof_state", "int","0","","","0=Closed 1=Opened 2=Intermediate") # --- device CV7 rep.append_device("CV7", "weather_station", "RET6789", "0") rep.append_value( "Error_code", "int","0","","","0=OK 1=SerialPB 2=FramePB") rep.append_value( "Power_input", "int","0","","","0=0Volt 1=12Volts (Power from MiFe board)") rep.append_value( "Wind_dir", "float","0","degrees","Wind_direction","0=North 90=East") rep.append_value( "Wind_speed", "float","0","m/s","Wind_spped","Wind speed from ultrasonic sensors") rep.append_value( "OutsideTemperature", "float","0","degC","Temperature_outside","Temperature of the anemometer") # --- device DHT22 rep.append_device("DHT22", "weather_station", "123", "0") rep.append_value( "Error_code", "int","0","","","0=OK 1=CheksumPB 2=LengthPB 3=NodataPB") rep.append_value( "Temperature", "float","21","degC","temperature","Temperature inside PLX box") rep.append_value( "Humidity", "float","77","percent","humidity","Humidity inside PLX box") # --- device RG11 rep.append_device("RG11", "weather_station", "TY67", "0") rep.append_value( "Error_code", "int","0","","","0=OK +1=PowerPB +2=NopowerPB +4=AllonePB") rep.append_value( "RainSate", "int","0","","rain","0=Dry 1=Rain 2=Unknown") # --- device MLX90614 rep.append_device("MLX90614", "weather_station", "1", "0") rep.append_value( "Error_code", "int","0","","","0=OK 1=DataPB") rep.append_value( "SkyTemperature", "float","-15.67","degC","Temperature_sky","Temperature of the sky") rep.append_value( "CanTemperature", "float","22.47","degC","","Temperature of the TO can") # --- List the UFKIs ufiks = rep.get_values("","",False) print(" List of all UFKIs:") for ufik in ufiks: print(" UFKI = {}".format(ufik)) # --- read the value of one UFKI # --- save the Json file rep.save_json("plc_guitalens.json") # =============================== # --- Read an existing Json # =============================== rep = report.Status_json() path_filename = "plc_verysimple.json" print("Analysis of the file {}:".format(path_filename)) rep.load_json(path_filename) res = rep.get_entities() print(" entities = {}".format(res)) res = rep.get_devices() print(" devices = {}".format(res)) print(" List of all UFKIs:") ufiks = rep.get_values("","",False) for ufik in ufiks: print(" UFKI = {}".format(ufik))