from fastapi import FastAPI, HTTPException from fastapi.middleware.cors import CORSMiddleware from fastapi.responses import JSONResponse from fastapi.encoders import jsonable_encoder import pymysql import asyncio from pydantic import BaseModel import datetime import json import traceback import re import difflib from pathlib import Path # --- Mistral Inference Imports --- from mistral_inference.transformer import Transformer from mistral_inference.generate import generate from mistral_common.tokens.tokenizers.mistral import MistralTokenizer try: from mistral_common.protocol.instruct.messages import ChatCompletionRequest, UserMessage, SystemMessage except ImportError: try: from mistral_common.protocol.instruct.request import ChatCompletionRequest from mistral_common.protocol.instruct.messages import UserMessage, SystemMessage except ImportError: import mistral_common.protocol.instruct.messages as msg_module UserMessage = msg_module.UserMessage SystemMessage = msg_module.SystemMessage pass # --------------------------------------------------------- # [설정 1] 데이터베이스 연결 정보 (MMCL) # --------------------------------------------------------- DB_CONFIG = { 'host': 'qst-s.iptime.org', 'port': 33063, 'user': 'mmcl_user', 'password': 'qsentech!1233', 'database': 'mmcl_db', 'autocommit': True, 'cursorclass': pymysql.cursors.DictCursor } def get_db(): try: return pymysql.connect(**DB_CONFIG) except Exception as e: print(f"DB Connection Error: {e}") return None # --------------------------------------------------------- # [설정 2] 모델 로드 # --------------------------------------------------------- mistral_models_path = Path.home().joinpath('mistral_models', '7B-Instruct-v0.3') tokenizer_path = mistral_models_path / "tokenizer.model.v3" print("=== 모델 로딩 중 ===") tokenizer = MistralTokenizer.from_file(str(tokenizer_path)) # 메모리 절약을 위해 max_batch_size를 1로 제한 (기본값이 커서 OOM 발생 가능) model = Transformer.from_folder(mistral_models_path, max_batch_size=1) print("=== 준비 완료 ===") app = FastAPI(title="MMCL Backend API") app.add_middleware( CORSMiddleware, allow_origins=["*"], allow_credentials=True, allow_methods=["*"], allow_headers=["*"], ) class ChatRequest(BaseModel): message: str # --------------------------------------------------------- # [Helper] 모델 호출 함수 # --------------------------------------------------------- def ask_mistral(messages, max_tokens=1024, temperature=0.1): chat_request = ChatCompletionRequest(messages=messages) tokens = tokenizer.encode_chat_completion(chat_request).tokens out_tokens, _ = generate( [tokens], model, max_tokens=max_tokens, temperature=temperature, eos_id=tokenizer.instruct_tokenizer.tokenizer.eos_id ) return tokenizer.instruct_tokenizer.tokenizer.decode(out_tokens[0]) # --------------------------------------------------------- # [프롬프트 정의 및 의도 분석] # --------------------------------------------------------- def get_control_prompt(): return """ You are an equipment control assistant. Read the user message and extract the target machine name, color, and action. Valid colors: RED, YELLOW, GREEN, ALL. (Use ALL if the user wants to control all colors). Valid actions: ON, OFF. Output ONLY a valid JSON object with 'machine_name', 'color', and 'action' keys. Do not include markdown ticks. Example 1: {"machine_name": "CNC 선반 1호기", "color": "ALL", "action": "OFF"} Example 2: {"machine_name": "CNC 2호기", "color": "RED", "action": "ON"} """ def analyze_intent_with_llm(user_message: str): """ 직접 띄운 Mistral LLM을 호출하여 제어할 장비명, 색상, 동작(ON/OFF) 의도를 JSON 형태로 추출. """ try: messages = [ SystemMessage(content=get_control_prompt()), UserMessage(content=f'User Message: "{user_message}"') ] # JSON 출력을 위해 토큰 길이를 넉넉히 줌 result_text = ask_mistral(messages, max_tokens=80, temperature=0.1).strip() # 정규식으로 JSON 부분만 추출 (백틱 등으로 감싸져 있을 경우 대비) match = re.search(r'\{.*?\}', result_text, re.DOTALL) if match: parsed = json.loads(match.group(0)) machine_name = parsed.get("machine_name", "") color = parsed.get("color", "").upper() action = parsed.get("action", "").upper() return color, action, machine_name return None, None, None except Exception as e: print(f"LLM Inference Error: {e}") return None, None, None # --------------------------------------------------------- # [API 엔드포인트] # --------------------------------------------------------- @app.get("/api/machines") def get_machines(): conn = get_db() if not conn: return [] try: with conn.cursor() as cursor: sql = """ SELECT d.dev_id as id, d.dev_name as machine_name, p.pos_name as location, s_nilm.value_ch1_pwr as latest_power, s_led.value_ch1_statusID as led_green, s_led.value_ch2_statusID as led_yellow, s_led.value_ch3_statusID as led_red, s_led.target_ch1_statusID as t_green, s_led.target_ch2_statusID as t_yellow, s_led.target_ch3_statusID as t_red FROM dev_info d LEFT JOIN pos_info p ON d.dev_posID = p.pos_id LEFT JOIN sensor_info s_nilm ON d.nilm_sensorNo = s_nilm.sensor_no LEFT JOIN sensor_info s_led ON d.led_sensorNo = s_led.sensor_no """ cursor.execute(sql) results = cursor.fetchall() formatted = [] for row in results: light_status = 'OFF' if row['led_red'] == 1: light_status = 'RED' elif row['led_yellow'] == 1: light_status = 'YELLOW' elif row['led_green'] == 1: light_status = 'GREEN' target_status = 'OFF' if row['t_red'] == 1: target_status = 'RED' elif row['t_yellow'] == 1: target_status = 'YELLOW' elif row['t_green'] == 1: target_status = 'GREEN' row['light_status'] = light_status row['target_light_status'] = target_status formatted.append(row) return formatted finally: if conn: conn.close() @app.post("/api/chat_control") async def chat_control(req: ChatRequest): # 1. LLM 의도 분석 color_name, action, machine_name = analyze_intent_with_llm(req.message) if not color_name or not action or not machine_name: return {"reply": "[LLM] 전달하신 메시지에서 장비명이나 제어 의도(ON/OFF)를 정확히 파악하지 못했습니다."} # 2. DB 검색 및 target_status 연산 업데이트 conn = get_db() if not conn: raise HTTPException(status_code=500, detail="DB Error") target = None try: with conn.cursor() as cursor: # 전체 장비명을 가져와서 difflib으로 가장 유사한 이름 찾기 cursor.execute("SELECT dev_no, dev_name, led_sensorNo FROM dev_info WHERE led_sensorNo IS NOT NULL") all_devs = cursor.fetchall() if not all_devs: return {"reply": "데이터베이스에 매핑 가능한 장비(센서)가 없습니다."} dev_names = [d['dev_name'] for d in all_devs] matches = difflib.get_close_matches(machine_name, dev_names, n=1, cutoff=0.5) if not matches: return {"reply": f"[LLM] 요청하신 '{machine_name}'과(와) 일치하거나 유사한 장비를 찾을 수 없습니다. 등록된 장비명을 확인해 주세요."} matched_dev_name = matches[0] dev = next(d for d in all_devs if d['dev_name'] == matched_dev_name) dev_no = dev['dev_no'] led_sensor_no = dev['led_sensorNo'] # 해당 센서의 현재 상태값 가져오기 cursor.execute("SELECT value_ch1_statusID, value_ch2_statusID, value_ch3_statusID FROM sensor_info WHERE sensor_no=%s", (led_sensor_no,)) current_state = cursor.fetchone() if not current_state: return {"reply": "해당 장비의 센서 상태 정보를 읽을 수 없습니다."} curr_green = current_state['value_ch1_statusID'] curr_yellow = current_state['value_ch2_statusID'] curr_red = current_state['value_ch3_statusID'] target_green, target_yellow, target_red = curr_green, curr_yellow, curr_red # 상태 연산 로직 (기존 상태 유지하면서 특정 색상만 제어) val = 1 if action == 'ON' else 0 if "ALL" in color_name: target_green, target_yellow, target_red = val, val, val elif "GREEN" in color_name: target_green = val elif "YELLOW" in color_name: target_yellow = val elif "RED" in color_name: target_red = val target = (target_green, target_yellow, target_red) # 이미 현재 상태가 목표 상태와 동일한지 체크 if target == (curr_green, curr_yellow, curr_red): return {"reply": f"[LLM] '{matched_dev_name}' 장비의 경광등은 이미 요청하신 상태입니다. (제어 생략)"} # 상태 업데이트 (Handshake Start) cursor.execute(""" UPDATE sensor_info SET target_ch1_statusID=%s, target_ch2_statusID=%s, target_ch3_statusID=%s, last_updated_by='LLM' WHERE sensor_no=%s """, (target[0], target[1], target[2], led_sensor_no)) # AI 제어 로그 삽입 cursor.execute(""" INSERT INTO ai_control_log (req_text, dev_no, target_val) VALUES (%s, %s, %s) """, (req.message, dev_no, f"{color_name} {action}")) finally: conn.close() # 3. Handshake Waiting Loop (Agent 처리 대기) max_wait = 15 # 15초 waited = 0 while waited < max_wait: conn = get_db() try: with conn.cursor() as cursor: cursor.execute(""" SELECT value_ch1_statusID, value_ch2_statusID, value_ch3_statusID FROM sensor_info WHERE sensor_no=%s """, (led_sensor_no,)) res = cursor.fetchone() if res and res['value_ch1_statusID'] == target[0] and res['value_ch2_statusID'] == target[1] and res['value_ch3_statusID'] == target[2]: return {"reply": f"[LLM] '{matched_dev_name}' 장비 제어 완료: {color_name} {action} 명령이 정상적으로 적용되었습니다."} finally: if conn: conn.close() await asyncio.sleep(1) waited += 1 raise HTTPException(status_code=504, detail="하드웨어 제어 응답 시간 초과") @app.get("/api/ai_control_logs") def get_ai_control_logs(): conn = get_db() if not conn: return [] try: with conn.cursor() as cursor: cursor.execute(""" SELECT l.log_id, l.req_text, l.target_val, l.created_at, d.dev_name FROM ai_control_log l JOIN dev_info d ON l.dev_no = d.dev_no ORDER BY l.created_at DESC LIMIT 50 """) return cursor.fetchall() finally: if conn: conn.close() @app.get("/api/event_logs") def get_event_logs(): conn = get_db() if not conn: return [] try: with conn.cursor() as cursor: cursor.execute(""" SELECT e.event_id, e.event_type, e.status, e.occurred_at, e.resolved_at, d.dev_name FROM event_log e JOIN dev_info d ON e.dev_no = d.dev_no ORDER BY e.occurred_at DESC LIMIT 50 """) return cursor.fetchall() finally: if conn: conn.close() if __name__ == "__main__": import uvicorn uvicorn.run(app, host="0.0.0.0", port=8000)