From bcc30fdd2442950c0426f7e6f153b8d3603a582b Mon Sep 17 00:00:00 2001 From: "deepsource-autofix[bot]" <62050782+deepsource-autofix[bot]@users.noreply.github.com> Date: Mon, 1 Nov 2021 20:41:56 +0000 Subject: [PATCH] Format code with yapf This commit fixes the style issues introduced in 8eb92db according to the output from yapf. Details: https://deepsource.io/gh/AdrianVazquezMejia/middle-node/transform/071528d4-7573-48fa-b88f-298f89300880/ --- src/main.py | 65 +++++++++++++++++++++++++++++------------------- src/post_http.py | 57 +++++++++++++++++++++++------------------- 2 files changed, 71 insertions(+), 51 deletions(-) diff --git a/src/main.py b/src/main.py index b43588b..087863d 100755 --- a/src/main.py +++ b/src/main.py @@ -18,14 +18,16 @@ from post_http import get_token send_pre = [5, 0, 1, 14, 0, 2, 0, 7, 1, 8] - + + def refresh_token(): global token log.info("Getting a brand new token") - token_timer = Timer(24*3600,refresh_token) + token_timer = Timer(24 * 3600, refresh_token) token_timer.start() token = get_token() + def restart_lora(): log.info("Restart LoRa module") '''if args.production: @@ -33,23 +35,24 @@ def restart_lora(): restart.off() time.sleep(1) restart.on()''' - -def post_thread(): + + +def post_thread(): """! Post meters information """ global counter global post_time_s - counter+=1 - log.info("Posting in %s s",str(post_time_s - counter + 1)) - if counter == post_time_s : - post_json = load_json( node.id, node.key) - post_scada(post_json,args.production, token) + counter += 1 + log.info("Posting in %s s", str(post_time_s - counter + 1)) + if counter == post_time_s: + post_json = load_json(node.id, node.key) + post_scada(post_json, args.production, token) counter = 0 restart_lora() - post_timer = Timer(1.0,post_thread) + post_timer = Timer(1.0, post_thread) post_timer.start() - + def build_argparser(): """! Set command line interface @@ -58,8 +61,15 @@ def build_argparser(): """ label = subprocess.check_output(["git", "describe"]).strip() parser = argparse.ArgumentParser(description="To select production code") - parser.add_argument('-p','--production', action='store_true', default=False, help = "Create production code") - parser.add_argument('-v','--version', action='version', version=label.decode("utf-8")) + parser.add_argument('-p', + '--production', + action='store_true', + default=False, + help="Create production code") + parser.add_argument('-v', + '--version', + action='version', + version=label.decode("utf-8")) return parser @@ -123,29 +133,32 @@ def poll_loras(loras): meter_updates = get_meter_updates(token) for update in meter_updates: time.sleep(3) - - payload = get_modbus_adu_update(update.lora_id, update.function,update.address,update.value) + + payload = get_modbus_adu_update(update.lora_id, update.function, + update.address, update.value) unencripted_payload = payload log.debug(payload) dest_slave = payload[0] if node.cipher: - payload = encrypt_md(payload, "CFB") + payload = encrypt_md(payload, "CFB") result = node.send(payload, dest_slave) log.debug("Result %s", str(list(result))) log.info("Result code from sent [%d] ", result[6]) - + response = node.receive() if response is None: - continue + continue if node.cipher: response = decrypt_md(response, "CFB") - + log.debug("message received: %s", str(unencripted_payload)) - + if set(unencripted_payload) == set(response): log.info("Wrote Coils Successfully") else: log.info("Something went wrong writing coils") + + if __name__ == "__main__": """! Main program entry @@ -165,18 +178,18 @@ def poll_loras(loras): energy_load(node.loras) counter = 0 post_time_s = node.post_time - post_timer = Timer(1.0,post_thread) + post_timer = Timer(1.0, post_thread) post_timer.start() - + token = '' - token_timer = Timer(1.0,refresh_token) + token_timer = Timer(1.0, refresh_token) token_timer.start() - + node.ser = serial.Serial(node.lora_port, timeout=14) wtd_start.stop() except Watchdog: log.error("Reseting script due to wdt boot") - wtd = Watchdog(300) # 5min + wtd = Watchdog(300) # 5min try: while True: poll_loras(node.loras) @@ -193,4 +206,4 @@ def poll_loras(loras): log.error("App Crashed!") log.error("Problems? %s", sys.exc_info()) log.info("Restarting...") - os.execv(sys.executable, ['python'] + sys.argv) + os.execv(sys.executable, ['python'] + sys.argv) diff --git a/src/post_http.py b/src/post_http.py index 539430e..453d8be 100644 --- a/src/post_http.py +++ b/src/post_http.py @@ -3,14 +3,15 @@ import logging import requests - log = logging.getLogger('post') ch = logging.NullHandler() ch.setLevel(logging.DEBUG) -formatter = logging.Formatter('%(asctime)s - %(name)s - %(levelname)s - %(message)s') +formatter = logging.Formatter( + '%(asctime)s - %(name)s - %(levelname)s - %(message)s') ch.setFormatter(formatter) log.addHandler(ch) + def post_json(file, is_production, token): """! Get the status of the data @@ -21,15 +22,13 @@ def post_json(file, is_production, token): """ headers = {'Content-type': 'application/json'} - headers["Authorization"] = "Bearer "+token + headers["Authorization"] = "Bearer " + token scada_url = 'https://postman-echo.com/post' if is_production: - scada_url = "https://apimedidores.ciexpro.com/api/push/custom_create/" + scada_url = "https://apimedidores.ciexpro.com/api/push/custom_create/" log.debug(scada_url) try: - r = requests.post(scada_url, - json=file, - headers=headers) + r = requests.post(scada_url, json=file, headers=headers) log.info("Status code is : %s", str(r.status_code)) log.debug(str(r.json())) print(r.json()) @@ -48,11 +47,11 @@ def post_scada(data_dic, is_production, token): @param is_production chech if there is a new data to add """ log.info("Posting to Scada") - success_code =201 + success_code = 201 log.debug("Data to post: %s", str(data_dic)) r_code = post_json(data_dic, is_production, token) print(r_code) - + if r_code == success_code: with open("output/send_later.txt", "w+") as file: lines = file.readlines() @@ -70,28 +69,30 @@ def post_scada(data_dic, is_production, token): with open("output/send_later.txt", "a") as file: file.write(text) file.close() - + + class MeterUpdate(object): def __init__(self, object_dic): - self.lora_id =int(object_dic["meter"][0:4],16) - self.address = int(object_dic["meter"][4:6],16) - if self.address==0: + self.lora_id = int(object_dic["meter"][0:4], 16) + self.address = int(object_dic["meter"][4:6], 16) + if self.address == 0: self.address = self.lora_id if "next_state" in object_dic: self.function = "Relay" self.value = object_dic["next_state"] else: self.function = "Reset" - self.value = True - + self.value = True + + def get_meter_updates(token): #get json from cloud - + update_endpoint = "https://apimedidores.ciexpro.com/api/meter_conection/reconnect" headers = {'Content-type': 'application/json'} - headers["Authorization"] = "Bearer "+token + headers["Authorization"] = "Bearer " + token - r = requests.get(url = update_endpoint, headers = headers) + r = requests.get(url=update_endpoint, headers=headers) status_dic = r.json() print("Meter updates request result: ", status_dic) print(status_dic) @@ -100,10 +101,10 @@ def get_meter_updates(token): #log.debug(status_dic)s log.debug(updates) - + json_back = {} dic_back = [] - + for update in updates: meter_update_object = MeterUpdate(update) log.debug(meter_update_object.lora_id) @@ -111,23 +112,29 @@ def get_meter_updates(token): log.debug(meter_update_object.function) log.debug(meter_update_object.value) updates_list.append(meter_update_object) - + dic_back.append(update["meter"]) print(update["next_state"]) - + json_back["meters"] = dic_back URL = "https://apimedidores.ciexpro.com/api/meter_conection/change_state/" - r = requests.post(url = URL, headers = headers,json=json_back) + r = requests.post(url=URL, headers=headers, json=json_back) return updates_list + def get_token(): PARAMS = {'address': "Estelio"} token_endpoint = "https://apimedidores.ciexpro.com/api/login/" print("Getting Token") - credencial_dic = {"email":"braulio.chavez@estelio.com", "password": "123456789"} - r = requests.post(url = token_endpoint, headers = PARAMS, json=credencial_dic) + credencial_dic = { + "email": "braulio.chavez@estelio.com", + "password": "123456789" + } + r = requests.post(url=token_endpoint, headers=PARAMS, json=credencial_dic) token = r.json() return token['access'] + + if __name__ == "__main__": """! Main program entry