초기화
This commit is contained in:
@@ -0,0 +1,321 @@
|
||||
import paho.mqtt.client as mqtt
|
||||
import json
|
||||
import logging
|
||||
import time
|
||||
import re
|
||||
import asyncio
|
||||
|
||||
logger = logging.getLogger("mqtt_client")
|
||||
|
||||
# ── 설정 ──────────────────────────────────────────────────
|
||||
BROKER_HOST = "localhost"
|
||||
BROKER_PORT = 1883
|
||||
|
||||
# MQTT 토픽 규칙:
|
||||
# 장비→서버(읽기): /{prefix}/m/in, /{prefix}/d/in, /{prefix}/event
|
||||
# 서버→장비(쓰기): /{prefix}/m/out, /{prefix}/d/out
|
||||
|
||||
_client: mqtt.Client = None
|
||||
|
||||
|
||||
from .line_manager import line_manager
|
||||
|
||||
def _get_equipment_id_by_prefix(prefix: str) -> int:
|
||||
"""
|
||||
토픽 프리픽스를 기반으로 DB의 설비 ID를 반환합니다 (LineManager 연동).
|
||||
"""
|
||||
return line_manager.get_equipment_id(prefix)
|
||||
|
||||
|
||||
def _publish_hmi_update(equipment_id: int):
|
||||
"""
|
||||
HMI 실시간 상태 업데이트 데이터를 조회하여 WebSocket으로 브로드캐스트합니다.
|
||||
MQTT 수신 스레드가 블로킹되는 현상을 피하기 위해 Future 대기 없이 이벤트 루프에 비동기 처리만 위임합니다.
|
||||
"""
|
||||
try:
|
||||
from equipment.db_manager import get_machine_status_detail_joined
|
||||
from communication.websocket_manager import ws_manager
|
||||
|
||||
result = get_machine_status_detail_joined(equipment_id)
|
||||
if result:
|
||||
# Decimal 타입 등의 변환 오류 방지 안전 필터 적용
|
||||
# 그룹 0,5(순환펌프 재작동 시간)만 정수, 나머지 d_read와 모든 d_write는 PLC REAL이므로 소수점 보존
|
||||
for i in range(10):
|
||||
key = f"d_read_g{i}"
|
||||
if key in result and result[key] is not None:
|
||||
try:
|
||||
result[key] = int(result[key]) if i in (0, 5) else float(result[key])
|
||||
except:
|
||||
result[key] = 0
|
||||
for i in range(4):
|
||||
key = f"d_write_g{i}"
|
||||
if key in result and result[key] is not None:
|
||||
try:
|
||||
result[key] = float(result[key])
|
||||
except:
|
||||
result[key] = 0
|
||||
|
||||
# 웹소켓 매니저의 메인 이벤트 루프로 비동기 브로드캐스트 위임
|
||||
loop = getattr(ws_manager, 'loop', None)
|
||||
if loop and loop.is_running():
|
||||
asyncio.run_coroutine_threadsafe(
|
||||
ws_manager.broadcast({
|
||||
"type": "equipment_update",
|
||||
"equipment_id": equipment_id,
|
||||
"data": result,
|
||||
"timestamp": int(time.time() * 1000)
|
||||
}),
|
||||
loop
|
||||
)
|
||||
except Exception as e:
|
||||
logger.error(f"[WS BROADCAST ERROR] Failed to broadcast for equipment_id {equipment_id}: {e}", exc_info=True)
|
||||
|
||||
|
||||
def _on_connect(client, userdata, flags, rc):
|
||||
if rc == 0:
|
||||
logger.info("[MQTT] Connected to broker successfully.")
|
||||
|
||||
# line_config.json에 등록된 모든 라인의 토픽 동적 구독
|
||||
prefixes = line_manager.get_all_prefixes()
|
||||
for prefix in prefixes:
|
||||
client.subscribe(f"/{prefix}/m/in")
|
||||
client.subscribe(f"/{prefix}/d/in")
|
||||
client.subscribe(f"/{prefix}/event")
|
||||
client.subscribe(f"/{prefix}/ack")
|
||||
client.subscribe(f"/{prefix}/status")
|
||||
|
||||
logger.info(f"[MQTT] Dynamically subscribed to topics for prefixes: {prefixes}")
|
||||
else:
|
||||
logger.error(f"[MQTT] Connection failed: rc={rc}")
|
||||
|
||||
|
||||
def _on_message(client, userdata, msg):
|
||||
"""
|
||||
배치 데이터 [[id, val], ...] 수집 및 DB 동기화, HMI 실시간 전계 (성능 모니터링 포함)
|
||||
"""
|
||||
start_time = time.time()
|
||||
topic = msg.topic
|
||||
try:
|
||||
payload_str = msg.payload.decode()
|
||||
logger.info(f"[MQTT RECEIVE] Topic: {topic} | Payload: {payload_str}")
|
||||
|
||||
payload = json.loads(payload_str)
|
||||
|
||||
# 토픽 프리픽스 동적 해석
|
||||
match = re.search(r'/([^/]+)/(m|d|event|ack|status)', topic)
|
||||
line_prefix = match.group(1) if match else "line1"
|
||||
equipment_id = _get_equipment_id_by_prefix(line_prefix)
|
||||
|
||||
from database.connection import get_db_connection
|
||||
|
||||
if "/ack" in topic:
|
||||
command_id = payload.get("command_id")
|
||||
status = payload.get("status")
|
||||
applied_value = payload.get("applied_value") # 적용 데이터 파싱
|
||||
logger.info(f"[MQTT ACK Callback] Received Event command {command_id} status={status} applied_value={applied_value} for prefix: {line_prefix}")
|
||||
|
||||
# DB 상태 및 적용값 분리 저장 업데이트
|
||||
from equipment.db_manager import update_control_command_status
|
||||
update_control_command_status(command_id, status, applied_value)
|
||||
|
||||
# WebSocket HMI 브로드캐스트
|
||||
from communication.websocket_manager import ws_manager
|
||||
loop = getattr(ws_manager, 'loop', None)
|
||||
if loop and loop.is_running():
|
||||
asyncio.run_coroutine_threadsafe(
|
||||
ws_manager.broadcast({
|
||||
"type": "command_ack",
|
||||
"command_id": command_id,
|
||||
"status": status,
|
||||
"applied_value": applied_value,
|
||||
"timestamp": int(time.time() * 1000)
|
||||
}),
|
||||
loop
|
||||
)
|
||||
|
||||
elif "/status" in topic:
|
||||
gateway = payload.get("gateway", "OFFLINE")
|
||||
plc_state = payload.get("plc_state", "OFFLINE")
|
||||
|
||||
logger.info(
|
||||
f"[MQTT STATUS Callback] gateway={gateway}, "
|
||||
f"plc_state={plc_state} for prefix: {line_prefix}"
|
||||
)
|
||||
|
||||
# 상태 토픽을 기준으로 장비 상태와 동기화 상태를 함께 갱신
|
||||
sync_state = None
|
||||
|
||||
if gateway == "OFFLINE" or plc_state == "OFFLINE":
|
||||
device_state = "OFFLINE"
|
||||
elif gateway == "SYNCING":
|
||||
device_state = "SYNCING"
|
||||
sync_state = "SYNCING"
|
||||
else:
|
||||
device_state = "ONLINE"
|
||||
sync_state = "IDLE"
|
||||
|
||||
with get_db_connection() as conn:
|
||||
with conn.cursor() as cursor:
|
||||
cursor.execute(
|
||||
"""
|
||||
UPDATE machine_status
|
||||
SET gateway_state = %s,
|
||||
plc_state = %s,
|
||||
device_state = %s,
|
||||
sync_state = COALESCE(%s, sync_state),
|
||||
timestamp = CURRENT_TIMESTAMP
|
||||
WHERE equipment_id = %s
|
||||
""",
|
||||
(
|
||||
gateway,
|
||||
plc_state,
|
||||
device_state,
|
||||
sync_state,
|
||||
equipment_id
|
||||
)
|
||||
)
|
||||
conn.commit()
|
||||
|
||||
_publish_hmi_update(equipment_id)
|
||||
|
||||
elif "/event" in topic:
|
||||
event_id = payload.get("id")
|
||||
event_val = payload.get("val")
|
||||
logger.info(f"[MQTT Event Callback] Received Event {event_id} (value={event_val}) for prefix: {line_prefix}")
|
||||
|
||||
# Full-Sync 혹은 PLC 오프라인 상태값 업데이트
|
||||
device_state = "ONLINE"
|
||||
sync_state = "IDLE"
|
||||
if event_id == 999: # Full-Sync 시작 (STATE_INIT 진입)
|
||||
device_state = "SYNCING"
|
||||
sync_state = "SYNCING"
|
||||
elif event_id == 998: # PLC Modbus 통신 실패 감지
|
||||
device_state = "OFFLINE"
|
||||
elif event_id == 997: # Full-Sync 완료 (정상 루프 진입)
|
||||
device_state = "ONLINE"
|
||||
sync_state = "IDLE"
|
||||
|
||||
with get_db_connection() as conn:
|
||||
with conn.cursor() as cursor:
|
||||
if event_id == 998:
|
||||
cursor.execute(
|
||||
"UPDATE machine_status SET device_state = %s, plc_state = 'OFFLINE', timestamp = CURRENT_TIMESTAMP WHERE equipment_id = %s",
|
||||
(device_state, equipment_id)
|
||||
)
|
||||
else:
|
||||
cursor.execute(
|
||||
"UPDATE machine_status SET device_state = %s, sync_state = %s, timestamp = CURRENT_TIMESTAMP WHERE equipment_id = %s",
|
||||
(device_state, sync_state, equipment_id)
|
||||
)
|
||||
conn.commit()
|
||||
|
||||
logger.info(f"[MQTT Event DB Update Completed] device_state updated to: {device_state}")
|
||||
_publish_hmi_update(equipment_id)
|
||||
|
||||
elif "/m/in" in topic:
|
||||
if not isinstance(payload, list):
|
||||
raise ValueError("M_in payload is not a JSON list")
|
||||
logger.info(f"[MQTT Processing M_in] Size of incoming batch: {len(payload)} for prefix: {line_prefix}")
|
||||
# 1. 기존 m_read_raw 및 m_write_raw JSON 데이터 조회
|
||||
with get_db_connection() as conn:
|
||||
with conn.cursor(dictionary=True) as cursor:
|
||||
cursor.execute("SELECT m_read_raw, m_write_raw FROM machine_status WHERE equipment_id = %s", (equipment_id,))
|
||||
row = cursor.fetchone()
|
||||
|
||||
m_read = {}
|
||||
if row and row.get("m_read_raw"):
|
||||
try:
|
||||
m_read = json.loads(row["m_read_raw"])
|
||||
except:
|
||||
pass
|
||||
|
||||
m_write = {}
|
||||
if row and row.get("m_write_raw"):
|
||||
try:
|
||||
m_write = json.loads(row["m_write_raw"])
|
||||
except:
|
||||
pass
|
||||
|
||||
# 2. 배치 배열 수신값 머지 [[bit_id, val], ...]
|
||||
for item in payload:
|
||||
if isinstance(item, list) and len(item) == 2:
|
||||
bit_id = str(item[0])
|
||||
val = int(item[1])
|
||||
m_read[bit_id] = val
|
||||
|
||||
# 3. 모멘터리 비트 정리: 쓰기 버퍼(m_write_raw) 중 1로 켜져있던 비트들을 0으로 복구
|
||||
# (ESP32 측에서 이미 쓰기 실행 및 자동 리셋을 처리했으므로 서버측 DB만 동기화하고 패킷은 송신 안함)
|
||||
for k, v in m_write.items():
|
||||
if v == 1:
|
||||
m_write[k] = 0
|
||||
|
||||
# 4. DB에 병합된 M 데이터 업데이트
|
||||
cursor.execute(
|
||||
"UPDATE machine_status SET m_read_raw = %s, m_write_raw = %s, device_state = 'ONLINE', timestamp = CURRENT_TIMESTAMP WHERE equipment_id = %s",
|
||||
(json.dumps(m_read), json.dumps(m_write), equipment_id)
|
||||
)
|
||||
conn.commit()
|
||||
|
||||
logger.info(f"[MQTT Processing M_in Completed] Successfully merged and saved {len(payload)} bits to DB.")
|
||||
_publish_hmi_update(equipment_id)
|
||||
|
||||
elif "/d/in" in topic:
|
||||
if not isinstance(payload, list):
|
||||
raise ValueError("D_in payload is not a JSON list")
|
||||
logger.info(f"[MQTT Processing D_in] Size of incoming batch: {len(payload)} for prefix: {line_prefix}")
|
||||
# D영역 업데이트: [[group_id, val], ...]
|
||||
with get_db_connection() as conn:
|
||||
with conn.cursor() as cursor:
|
||||
for item in payload:
|
||||
if isinstance(item, list) and len(item) == 2:
|
||||
group_id = int(item[0])
|
||||
# 그룹 0,5(순환펌프 재작동 시간)는 UINT16 정수, 그 외는 PLC REAL이므로 소수점 보존
|
||||
val = int(item[1]) if group_id in (0, 5) else float(item[1])
|
||||
if 0 <= group_id <= 9:
|
||||
sql = f"UPDATE machine_status SET d_read_g{group_id} = %s, device_state = 'ONLINE', timestamp = CURRENT_TIMESTAMP WHERE equipment_id = %s"
|
||||
cursor.execute(sql, (val, equipment_id))
|
||||
conn.commit()
|
||||
|
||||
logger.info(f"[MQTT Processing D_in Completed] Successfully saved {len(payload)} D-registers to DB.")
|
||||
_publish_hmi_update(equipment_id)
|
||||
|
||||
# 성공 처리 시간 기록
|
||||
duration_ms = (time.time() - start_time) * 1000
|
||||
from monitoring.metrics import record_metric
|
||||
record_metric(topic, duration_ms)
|
||||
|
||||
except Exception as e:
|
||||
duration_ms = (time.time() - start_time) * 1000
|
||||
from monitoring.metrics import record_metric
|
||||
record_metric(topic, duration_ms, error=True, error_msg=str(e))
|
||||
logger.error(f"[MQTT ERROR] Message parse/update error: {e}", exc_info=True)
|
||||
|
||||
|
||||
def start_mqtt_client(db=None):
|
||||
"""
|
||||
FastAPI 어플리케이션 구동 시 호출됩니다.
|
||||
"""
|
||||
global _client
|
||||
_client = mqtt.Client(client_id="ChemiFactory_HMI_Server")
|
||||
|
||||
# MQTT 인증 계정 정보 적용 (rc=5 방지)
|
||||
_client.username_pw_set("ctnt_root", "Umsang6595!!")
|
||||
|
||||
_client.on_connect = _on_connect
|
||||
_client.on_message = _on_message
|
||||
|
||||
try:
|
||||
logger.info(f"[MQTT INIT] Connecting to broker at {BROKER_HOST}:{BROKER_PORT}...")
|
||||
_client.connect(BROKER_HOST, BROKER_PORT, 60)
|
||||
_client.loop_start()
|
||||
logger.info("[MQTT INIT] Background client started and loop active.")
|
||||
except Exception as e:
|
||||
logger.error(f"[MQTT INIT ERROR] Startup failed: {e}", exc_info=True)
|
||||
|
||||
|
||||
def stop_mqtt_client():
|
||||
global _client
|
||||
if _client:
|
||||
_client.loop_stop()
|
||||
_client.disconnect()
|
||||
logger.info("[MQTT] Background client stopped successfully.")
|
||||
Reference in New Issue
Block a user