from http import client import json from django.utils import timezone from .models import Ttnstate, Report, Product, Reportcurrentloop, Ttn import uuid import paho.mqtt.client as mqtt import ssl from django.db import connection import datetime import sys import os import io from django.db.models import Q import time import logging sys.stdout = io.TextIOWrapper(sys.stdout.buffer, encoding='utf-8') # Настройки MQTT MQTT_BROKER = "localhost" MQTT_PORT = 8883 MQTT_USERNAME = 'fomin_a' MQTT_PASSWORD = 'Hs6#2vG#%8bxsKZf4' MQTT_CLIENT_ID = str(uuid.uuid4()) MQTT_TOPICS_SUB = [ ("grand_beton/report/1", 0), ("grand_beton/report/2", 0), ("grand_beton/report/4", 0), ("grand_beton/report/8", 0), ("grand_beton/report/16", 0), ("grand_beton/report/32", 0), ("grand_beton/report/64", 0), ("grand_beton/report/128", 0), ("grand_beton/report/256", 0), ("grand_beton/report/512", 0), ("grand_beton/production/1", 0), ("grand_beton/production/2", 0), ("grand_beton/production/4", 0), ("grand_beton/production/8", 0), ("grand_beton/production/16", 0), ("grand_beton/production/32", 0), ("grand_beton/production/64", 0), ("grand_beton/production/128", 0), ("grand_beton/production/256", 0), ("grand_beton/production/512", 0), ("grand_beton/ttn/1", 0), ("grand_beton/ttn/2", 0), ("grand_beton/ttn/4", 0), ("grand_beton/ttn/8", 0), ("grand_beton/ttn/16", 0), ("grand_beton/ttn/32", 0), ("grand_beton/ttn/64", 0), ("grand_beton/ttn/128", 0), ("grand_beton/ttn/256", 0), ("grand_beton/ttn/512", 0) ] MQTT_TOPIC_PUB = "grand_beton/ttn/back" MQTT_QOS = 0 # Глобальная переменная для клиента client = mqtt.Client() # Словарь для хранения объектов по id objects_by_id = {} # Функция для проверки подключения к базе данных def ensure_connection(): if connection.connection and not connection.is_usable(): connection.close() # Функция для обновления статуса других топиков def update_status_for_other_topics(obj_id, topic_with_st_4, client): if obj_id in objects_by_id: for topic, status, payload in objects_by_id[obj_id]: if topic != topic_with_st_4 and status != 4: payload['st'] = 6 client.publish(topic, json.dumps(payload), qos=0) logging.info(f"Updated id={obj_id} in topic {topic}, set st=6") def save_message(client, topic, payload): ensure_connection() data = json.loads(payload) if "ttn" in topic or "production" in topic: code = data.get("st", None) # Получаем значение st date = data.get("datetime", None) print(f"___________ {code}") # Убедитесь, что дата - строка, прежде чем пытаться ее преобразовать if isinstance(date, str): date = timezone.make_aware(timezone.datetime.strptime(date, "%Y-%m-%d %H:%M:%S"), timezone.get_current_timezone()) # Удаляем временную зону перед сохранением date = date.replace(tzinfo=None) ttn_list = data.get("ttn", []) for ttn in ttn_list: idTtn = ttn.get("ind_ttn", None) if idTtn is not None: try: idTtn = int(str(ttn.get("ind_ttn", "")).strip()) ensure_connection() # Проверяем, существует ли запись с текущими значениями existing_record = Ttnstate.objects.filter( Q(date=date) & Q(idTtn=idTtn) ).exists() if not existing_record: # Создаем запись с полученным значением состояния Ttnstate.objects.create(idTtn=idTtn, state=code, date=date, json=payload) print(f"Saved new Ttnstate: idTtn={idTtn}, state={code}, date={date}") else: print(f"Record with idTtn={idTtn} already exists") ensure_connection() if Ttn.objects.filter(id=idTtn).exists(): Ttn.objects.filter(id=idTtn).update(state=code) print(f"Updated ttn: idTtn={idTtn}, state={code}") else: print(f"No record found in ttn for idTtn={idTtn}") except ValueError as ve: print(f"ValueError: {ve}") except Exception as e: print(f"Error: {e}") if code == 14: bsu = data.get("bsu", None) new_payload = json.loads(payload) new_payload['st'] = 15 new_topic = f"grand_beton/ttn/{bsu}" #print(f"Message with st=15 pddddd to ------") result = client.publish(new_topic, json.dumps(new_payload).encode('utf-8'), qos=MQTT_QOS, retain=False) status = result.rc if status == 0: print(f"Message with st=15 published to {new_topic} ------") else: print(f"Failed to send message with st=15 to {new_topic}, BSU result code: {status}") if code == 13: bsu = data.get("bsu", None) if bsu: new_payload = json.loads(payload) new_payload['st'] = 13 new_topic = f"grand_beton/ttn/{bsu}" result = client.publish(new_topic, json.dumps(new_payload).encode('utf-8'), qos=MQTT_QOS, retain=False) status = result.rc if status == 0: print(f"Message with st=13 published to {new_topic}") else: print(f"Failed to send message with st=13 to {new_topic}, result code: {status}") elif "report" in topic: product_list = data.get("product", []) current_loop_list = data.get("current_loop", []) idPlant = data.get("bsu") for product in product_list: dateStart = data.get("datetime", None) timeEnd = product.get("time_done", None) vProduct = product.get("v_product", None) loopNumber = product.get("num_loop", None) vLoop = product.get("v_loop", None) driver = product.get("driver", None) car = product.get("car", None) classRecipe = product.get("class_recipe") or "" nameRecipe = product.get("name_recipe", None) recipe = product.get("recipe", None) idTtn = product.get("ttn", None) timeStart = product.get("time_start", None) num_loop = product.get("num_loop", None) indProduct = product.get("ind_product", None) ensure_connection() if dateStart: naive_date = timezone.datetime.strptime(dateStart, "%Y-%m-%d %H:%M:%S") dateStart = timezone.make_aware(naive_date, timezone.get_current_timezone()) if not Product.objects.filter(idTtn=idTtn, idPlant=idPlant, indProduct=indProduct).exists(): product_obj = Product( dateStart=dateStart, timeEnd=timeEnd, vProduct=vProduct, loopNumber=loopNumber, vLoop=vLoop, driver=driver, car=car, classRecipe=classRecipe, nameRecipe=nameRecipe, recipe=recipe, idTtn=idTtn, timeStart=timeStart, num_loop=num_loop, idPlant=idPlant, indProduct=indProduct ) product_obj.save() for loop in current_loop_list: if loop.get("ind_product", None) == product.get("ind_product", None): ensure_connection() Reportcurrentloop.objects.create( vLoop=loop.get("v_loop", None), loopNumber=loop.get("num_loop", None), code=loop.get("code", None), dispencer=loop.get("dispenser", None), doisingError=loop.get("err_dosing", None), doisingErrorPersent=loop.get("err_dosing_persent", None), doisingKorr=loop.get("korr_dosing", None), humidityKorr=loop.get("korr_humidity", None), weightFactLoop=loop.get("weight_fact_loop", None), weightFactM3=loop.get("weight_fact_m3", None), weightRecipeLoop=loop.get("weight_recipe_loop", None), weightRecipeM3=loop.get("weight_recipe_m3", None), idProduct=product_obj.id, # Передаем идентификатор Product indProduct=indProduct ) ensure_connection() #Report.objects.create(json=payload) # Callback функция при подключении к брокеру def on_connect(client, userdata, flags, rc): print(f"Connected to MQTT broker with result code {rc}") client.subscribe(MQTT_TOPICS_SUB) # Callback функция при получении сообщения def on_message(client, userdata, msg): save_message(client, msg.topic, msg.payload.decode()) try: payload = json.loads(msg.payload.decode()) obj_id = payload.get("id") obj_status = payload.get("st") if obj_id is not None: if obj_id not in objects_by_id: objects_by_id[obj_id] = [] objects_by_id[obj_id].append((msg.topic, obj_status, payload)) if obj_status == 4: update_status_for_other_topics(obj_id, msg.topic, client) except Exception as e: print(f"Received message:-----otladka") def start_mqtt_client(): global client # Проверка блокировки lock_file = '/tmp/mqtt_client.lock' if os.path.exists(lock_file): print("MQTT client is already running.") return None # Создание файла блокировки with open(lock_file, 'w') as f: f.write(str(os.getpid())) try: # Создание MQTT клиента client = mqtt.Client(client_id=MQTT_CLIENT_ID) client.on_connect = on_connect client.on_message = on_message client.username_pw_set(MQTT_USERNAME, MQTT_PASSWORD) # Настройка TLS client.tls_set(ca_certs=None, cert_reqs=ssl.CERT_NONE, tls_version=ssl.PROTOCOL_TLS_CLIENT) client.tls_insecure_set(True) # Игнорируем проверку сертификата # Подключение к брокеру print(f"Connecting to MQTT broker {MQTT_BROKER}:{MQTT_PORT}...") client.connect(MQTT_BROKER, MQTT_PORT, 60) client.loop_start() print("MQTT client started successfully") return client except Exception as e: print(f"Failed to start MQTT client: {e}") # Удаление файла блокировки при ошибке if os.path.exists(lock_file): os.remove(lock_file) return None def stop_mqtt_client(): global client if client: client.loop_stop() client.disconnect() print("MQTT client stopped") # Удаление файла блокировки lock_file = '/tmp/mqtt_client.lock' if os.path.exists(lock_file): logging.info("MQTT client is already running OTLADKA.") sys.exit() if __name__ == "__main__": client = start_mqtt_client() try: while True: time.sleep(5) except KeyboardInterrupt: print("Exiting...") stop_mqtt_client()