from flask import Flask, request, jsonify from dotenv import dotenv_values from common.runtime import load_root_env, validate_env, as_int, as_str import os import sys import matplotlib.pyplot as plt from Model import Model import numpy as np import matplotlib import importlib import threading import requests import asyncio import shutil import json import gc import logging TORCHSIG_PATH = "/app/torchsig" if TORCHSIG_PATH not in sys.path: # Ensure import torchsig resolves to /app/torchsig/torchsig package. sys.path.insert(0, TORCHSIG_PATH) logging.basicConfig(level=logging.INFO) app = Flask(__name__) loop = asyncio.new_event_loop() asyncio.set_event_loop(loop) queue = asyncio.Queue() semaphore = asyncio.Semaphore(3) prediction_list = [] result_msg = {} results = [] matplotlib.use('Agg') plt.ioff() alg_list = [] model_list = [] ROOT_ENV = load_root_env(__file__) validate_env("NN_server/server.py", { "GENERAL_SERVER_IP": as_str, "GENERAL_SERVER_PORT": as_int, "SERVER_IP": as_str, "SERVER_PORT": as_int, "PATH_TO_NN": as_str, "SRC_RESULT": as_str, "SRC_EXAMPLE": as_str, }) config = dict(dotenv_values(ROOT_ENV)) if not config: raise RuntimeError("[NN_server/server.py] .env was loaded but no keys were parsed") if not any(key.startswith("NN_") for key in config): raise RuntimeError("[NN_server/server.py] no NN_* model entries configured") logging.info("NN config loaded from %s", ROOT_ENV) gen_server_ip = config['GENERAL_SERVER_IP'] gen_server_port = config['GENERAL_SERVER_PORT'] def init_data_for_inference(): try: if os.path.isdir(config['SRC_RESULT']): shutil.rmtree(config['SRC_RESULT']) os.mkdir(config['SRC_RESULT']) if os.path.isdir(config['SRC_EXAMPLE']): shutil.rmtree(config['SRC_EXAMPLE']) os.mkdir(config['SRC_EXAMPLE']) except Exception as exc: print(str(exc)) print() try: global model_list for key in config.keys(): if key.startswith('NN_'): params = config[key].split(' && ') module = importlib.import_module('Models.' + params[4]) classes = {} for value in params[9][1:-1].split(','): classes[len(classes)] = value model = Model(file_model=params[0], file_config=params[1], src_example=params[2], src_result=params[3], type_model=params[4], build_model_func=getattr(module, params[5]), pre_func=getattr(module, params[6]), inference_func=getattr(module, params[7]), post_func=getattr(module, params[8]), classes=classes, number_synthetic_examples=int(params[10]), number_src_data_for_one_synthetic_example=int(params[11]), path_to_src_dataset=params[12]) model_list.append(model) # if key.startswith('ALG_'): # params = config[key].split(' && ') # module = importlib.import_module('Algorithms.' + params[2]) # classes = {} # for value in params[6][1:-1].split(','): # classes[len(classes)] = value # alg = Algorithm(src_example=params[0], src_result=params[1], type_alg=params[2], pre_func=getattr(module, params[3]), # inference_func=getattr(module, params[4]), post_func=getattr(module, params[5]), classes=classes, # number_synthetic_examples=int(params[7]), number_src_data_for_one_synthetic_example=int(params[8]), path_to_src_dataset=params[9]) # alg_list.append(alg) except Exception as exc: print(str(exc)) print() def run_example(): try: for model in model_list: model.get_test_inference() except Exception as exc: print(str(exc)) @app.route('/receive_data', methods=['POST']) def receive_data(): try: print() data = json.loads(request.json) print('#' * 100) print('Получен пакет ' + str(Model.get_ind_inference())) freq = int(data['freq']) print('Частота: ' + str(freq)) # print('Канал: ' + str(data['channel'])) result_msg = {} data_to_send = {} prediction_list = [] #print(model_list) for model in model_list: #print(str(freq)) #print(model.get_model_name()) if str(freq) in model.get_model_name(): print('-' * 100) print(str(model)) result_msg[str(model.get_model_name())] = {'freq': freq} prediction, probability = model.get_inference([np.asarray(data['data_real'], dtype=np.float32), np.asarray(data['data_imag'], dtype=np.float32)]) result_msg[str(model.get_model_name())]['prediction'] = prediction result_msg[str(model.get_model_name())]['probability'] = str(probability) prediction_list.append(prediction) print('-' * 100) print() try: result = 0 if (int(freq) == 2400 and (prediction_list[0] in ['drone', 'drone_noise'] or (prediction_list[0] == 'wifi' and float(probability) >= 0.95))) or (int(freq) == 1200 and (prediction_list[0] in ['drone'] and float(probability) >= 0.95)): result += 8 if int(freq) in [915]: result = 0 if int(freq) in []: result = 8 data_to_send={ 'freq': str(freq), 'amplitude': result #'triggered': False if result < 7 else True, #'light_len': result } response = requests.post("http://{0}:{1}/process_data".format(gen_server_ip, gen_server_port), json=data_to_send) if response.status_code == 200: print("Данные успешно отправлены!") print("Частота: " + str(freq)) print("Отправлено светодиодов: " + str(result)) else: print("Ошибка при отправке данных: ", response.status_code) except Exception as exc: print(str(exc)) break Model.get_inc_ind_inference() print() print('#' * 100) for alg in alg_list: print('-' * 100) print(str(alg)) alg.get_inference([np.asarray(data['data_real'], dtype=np.float32), np.asarray(data['data_imag'], dtype=np.float32)]) print('-' * 100) print() #Algorithm.get_inc_ind_inference() print() print('#' * 100) del data gc.collect() return jsonify(result_msg) except Exception as exc: print(str(exc)) ''' def run_flask(): app.run(host=config['SERVER_IP'], port=int(config['SERVER_PORT'])) async def process_tasks(): workers = [asyncio.create_task(worker(queue=queue, semaphore=semaphore)) for _ in range(2)] await asyncio.gather(*workers) async def main(): asyncio.create_task(process_tasks()) flask_thread = threading.Thread(target=run_flask) flask_thread.start() while True: if queue.qsize() <= 1: asyncio.create_task(process_tasks()) await asyncio.sleep(1) @app.route('/receive_data', methods=['POST']) def add_task(): queue_size = queue.qsize() if queue_size > 1: return {} print() data = json.loads(request.json) print('#' * 100) print('Получен пакет ' + str(Model.get_ind_inference())) freq = int(data['freq']) print('Частота ' + str(freq)) result_msg = {} for model in model_list: if str(freq) in model.get_model_name(): print('-' * 100) print(str(model)) result_msg[str(model.get_model_name())] = {'freq': freq} asyncio.run_coroutine_threadsafe(queue.put({'freq': freq, 'model': model, 'data': data}), loop) do_inference(model=model, data=data, freq=freq) break del data gc.collect() return jsonify(result_msg) async def worker(queue, semaphore): while True: task = await queue.get() if task is None: break async with semaphore: try: await do_inference(model=task['model'], data=task['data'], freq=task['freq']) except Exception as e: print(str(e)) print(results) queue.task_done() async def do_inference(model=None, data=None, freq=0): prediction_list = [] print("Длина очереди" + str(queue.qsize())) inference(model=model, data=data, freq=freq) try: results = [] for pred in prediction_list: if pred[1] == 'drone': results.append([pred[0],8]) else: results.append([pred[0],0]) for result in results: try: data_to_send={ 'freq': result[0], 'amplitude': result[1], 'triggered': False if result[1] < 7 else True, 'light_len': result[1] } response = requests.post("http://{0}:{1}/process_data".format(gen_server_ip, gen_server_port), json=data_to_send) await response.text if response.status_code == 200: print("Данные успешно отправлены!") print("Отправлено светодиодов: " + str(data_to_send['light_len'])) else: print("Ошибка при отправке данных: ", response.status_code) except Exception as exc: print(str(exc)) except Exception as exc: print(str(exc)) Model.get_inc_ind_inference() print() print('#' * 100) del data gc.collect() def inference(model=None, data=None, freq=0): prediction, probability = model.get_inference([np.asarray(data['data_real'], dtype=np.float32), np.asarray(data['data_imag'], dtype=np.float32)]) result_msg[str(model.get_model_name())]['prediction'] = prediction result_msg[str(model.get_model_name())]['probability'] = str(probability) queue_size = queue.qsize() print(queue_size) prediction_list.append([freq, prediction]) print('-' * 100) print() if __name__ == '__main__': init_data_for_inference() #asyncio.run(main) loop.run_until_complete(main()) ''' def run_flask(): print(config['SERVER_IP']) app.run(host=config['SERVER_IP'], port=int(config['SERVER_PORT'])) if __name__ == '__main__': init_data_for_inference() flask_thread = threading.Thread(target=run_flask) flask_thread.start() #app.run(host=config['SERVER_IP'], port=int(config['SERVER_PORT']))