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 import requests from pathlib import Path # --------------------------------------------------------- # [설정 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] Ollama 로드 # --------------------------------------------------------- OLLAMA_URL = "http://localhost:11434/api/generate" #MODEL_NAME = "gemma4:e2b" MODEL_NAME = "deepseek-coder-v2:16b" app = FastAPI(title="MMCL Backend API") app.add_middleware( CORSMiddleware, allow_origins=["*"], allow_credentials=True, allow_methods=["*"], allow_headers=["*"], ) class ChatRequest(BaseModel): message: str class AnalyzePowerRequest(BaseModel): sensor_no: int voltage: float current: float power: float # --------------------------------------------------------- # [Helper] 모델 호출 함수 (Ollama) # --------------------------------------------------------- def ask_ollama(prompt_text, max_tokens=150, temperature=0.1): payload = { "model": MODEL_NAME, "prompt": prompt_text, "stream": False, "format": "json", "options": { "temperature": temperature, "num_predict": max_tokens } } try: response = requests.post(OLLAMA_URL, json=payload, timeout=10) response.raise_for_status() return response.json().get("response", "") except Exception as e: print(f"Ollama Request Error: {e}") return "" # --------------------------------------------------------- # [프롬프트 정의 및 의도 분석] # --------------------------------------------------------- def get_control_prompt(): return """ You are an equipment control assistant. Read the user message and extract the target machine name, colors, and action. Valid colors: RED, YELLOW, GREEN, ALL, UNKNOWN. (If the user asks for unsupported colors, use UNKNOWN). Valid actions: ON, OFF, STATUS, UNKNOWN. Output ONLY a valid JSON object with 'machine_name', 'colors' (as an array of strings), and 'action' keys. Do not include markdown ticks. Example 1: {"machine_name": "CNC 선반 1호기", "colors": ["ALL"], "action": "OFF"} Example 2: {"machine_name": "CNC 2호기", "colors": ["RED", "YELLOW"], "action": "ON"} Example 3: {"machine_name": "CNC 1호기", "colors": ["UNKNOWN"], "action": "STATUS"} Example 4: {"machine_name": "알수없음", "colors": ["UNKNOWN"], "action": "UNKNOWN"} """ def analyze_intent_with_llm(user_message: str): """ Ollama API를 호출하여 제어할 장비명, 색상, 동작(ON/OFF) 의도를 JSON 형태로 추출. """ try: prompt = get_control_prompt() + f'\n\nUser Message: "{user_message}"\nOutput JSON:' # JSON 출력을 위해 토큰 길이를 넉넉히 줌 result_text = ask_ollama(prompt, max_tokens=80, temperature=0.1).strip() # 정규식으로 JSON 부분만 추출 (백틱 등으로 감싸져 있을 경우 대비) match = re.search(r'\{.*?\}', result_text, re.DOTALL) if match: intent_data = json.loads(match.group()) colors = intent_data.get("colors", []) # 만약 LLM이 문자열 하나로 줬다면 리스트로 변환 if isinstance(colors, str): colors = [c.strip() for c in colors.split(",")] action = intent_data.get("action", "").upper() machine_name = intent_data.get("machine_name", "") return colors, action, machine_name return [], "", "" except Exception as e: print(f"LLM Inference Error: {e}") return [], "", "" # --------------------------------------------------------- # [전력 분석 기반 상태 추론] # --------------------------------------------------------- def get_power_analysis_prompt(init_volt, init_curr, init_pwr, curr_volt, curr_curr, curr_pwr): return f""" You are an AI that monitors machine status based on power consumption. The machine's baseline (idle/standby) is Voltage: {init_volt:.1f}V, Current: {init_curr:.1f}A, Power: {init_pwr:.1f}W. The current reading is Voltage: {curr_volt:.1f}V, Current: {curr_curr:.1f}A, Power: {curr_pwr:.1f}W. Analyze the machine state: - If current power is significantly higher than baseline (e.g. > 150% of baseline or high absolute value), the machine is Running. - If current power is near the baseline (e.g. within 10-30% margin), the machine is Idle/Standby. - If current power is near 0 or unexpectedly low/high indicating a fault, the machine is in Error. Output ONLY a valid JSON object with a single key 'status' with one of the following values: "GREEN" (Running), "YELLOW" (Idle), "RED" (Error). Do not include markdown ticks. Example 1: {{"status": "GREEN"}} Example 2: {{"status": "YELLOW"}} Example 3: {{"status": "RED"}} """ def analyze_power_state_with_llm(init_volt, init_curr, init_pwr, curr_volt, curr_curr, curr_pwr): try: prompt = get_power_analysis_prompt(init_volt, init_curr, init_pwr, curr_volt, curr_curr, curr_pwr) + '\nOutput JSON:' result_text = ask_ollama(prompt, max_tokens=40, temperature=0.1).strip() match = re.search(r'\{.*?\}', result_text, re.DOTALL) if match: intent_data = json.loads(match.group()) return intent_data.get("status", "YELLOW").upper() return "YELLOW" except Exception as e: print(f"LLM Power Inference Error: {e}") return "YELLOW" # --------------------------------------------------------- # [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/analyze_power") async def analyze_power(req: AnalyzePowerRequest): conn = get_db() if not conn: raise HTTPException(status_code=500, detail="DB Error") try: with conn.cursor() as cursor: # dev_info와 nilm_init_value, sensor_info 가져오기 sql = """ SELECT d.dev_no, d.led_sensorNo, n.ch1_current, n.ch1_volt, d.dev_name, s.value_ch1_statusID as v_green, s.value_ch2_statusID as v_yellow, s.value_ch3_statusID as v_red FROM dev_info d LEFT JOIN nilm_init_value n ON d.dev_no = n.dev_no LEFT JOIN sensor_info s ON d.led_sensorNo = s.sensor_no WHERE d.nilm_sensorNo = %s """ cursor.execute(sql, (req.sensor_no,)) dev_data = cursor.fetchone() if not dev_data: return {"reply": "설비 매핑 정보를 찾을 수 없습니다."} dev_no = dev_data['dev_no'] led_sensor_no = dev_data['led_sensorNo'] dev_name = dev_data['dev_name'] # 기준 전력 계산 (초기값 없으면 임의값 100W 사용) init_volt = dev_data['ch1_volt'] or 220.0 init_curr = dev_data['ch1_current'] or 0.0 init_pwr = init_volt * init_curr if init_pwr <= 0: init_pwr = 100.0 # LLM 호출 status = analyze_power_state_with_llm(init_volt, init_curr, init_pwr, req.voltage, req.current, req.power) if status not in ['GREEN', 'YELLOW', 'RED']: status = 'YELLOW' # target_status 업데이트 t_green = 1 if status == 'GREEN' else 0 t_yellow = 1 if status == 'YELLOW' else 0 t_red = 1 if status == 'RED' else 0 curr_v_green = dev_data.get('v_green') or 0 curr_v_yellow = dev_data.get('v_yellow') or 0 curr_v_red = dev_data.get('v_red') or 0 if t_green == curr_v_green and t_yellow == curr_v_yellow and t_red == curr_v_red: return {"reply": "NO_CHANGE"} if led_sensor_no: upd_sql = """ UPDATE sensor_info SET target_ch1_statusID = %s, target_ch2_statusID = %s, target_ch3_statusID = %s, last_updated_by = 'LLM_POWER_ANALYSIS' WHERE sensor_no = %s """ cursor.execute(upd_sql, (t_green, t_yellow, t_red, led_sensor_no)) # ai_control_log 기록 log_text = f"전력 변동 감지 (현재: {req.power:.1f}W, 기준: {init_pwr:.1f}W) -> LLM 상태 판단: {status}" cursor.execute( "INSERT INTO ai_control_log (req_text, dev_no, target_val) VALUES (%s, %s, %s)", (log_text, dev_no, status) ) return {"reply": f"[AI 전력 분석] {dev_name} 설비가 {status} 상태로 전환되었습니다."} finally: if conn: conn.close() @app.post("/api/chat_control") async def chat_control(req: ChatRequest): # 1. LLM 의도 분석 colors, action, machine_name = analyze_intent_with_llm(req.message) if not colors or not action or not machine_name: return {"reply": "[LLM] 전달하신 메시지에서 장비명이나 제어 의도를 정확히 파악하지 못했습니다."} if action == 'UNKNOWN': return {"reply": "[LLM] 어떤 동작(켜기/끄기/상태확인)을 원하시는지 명확하지 않습니다."} if action in ['ON', 'OFF']: if "UNKNOWN" in colors or not all(c in ['ALL', 'RED', 'YELLOW', 'GREEN'] for c in colors): return {"reply": "[LLM] 지원하지 않는 색상입니다. (빨간색, 노란색, 초록색만 제어 가능합니다)"} # 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") all_devs = cursor.fetchall() if not all_devs: return {"reply": "데이터베이스에 매핑 가능한 장비(센서)가 없습니다."} dev_names = [d['dev_name'] for d in all_devs] machine_name_clean = machine_name.replace(" ", "").lower() target_devs = [] # 1. 특수 키워드 (전체 제어) if machine_name_clean in ["all", "전체", "전부", "모든장비", "모든"]: target_devs = all_devs else: # 2. 부분 일치 검색 (다중 매칭) for d in all_devs: if machine_name_clean in d['dev_name'].replace(" ", "").lower(): target_devs.append(d) # 3. 오타 교정 (difflib) if not target_devs: matches = difflib.get_close_matches(machine_name, dev_names, n=10, cutoff=0.4) for m in matches: target_devs.append(next(d for d in all_devs if d['dev_name'] == m)) if not target_devs: return {"reply": f"[LLM] 요청하신 '{machine_name}'과(와) 일치하거나 유사한 장비를 찾을 수 없습니다."} updated_sensors = [] status_messages = [] invalid_targets = [] already_set_count = 0 color_name_str = ", ".join(colors) for dev in target_devs: dev_no = dev['dev_no'] led_sensor_no = dev['led_sensorNo'] dev_name = dev['dev_name'] if led_sensor_no is None: invalid_targets.append(dev_name) continue 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: continue curr_green = current_state['value_ch1_statusID'] curr_yellow = current_state['value_ch2_statusID'] curr_red = current_state['value_ch3_statusID'] if action == 'STATUS': state_msg = f"- [{dev_name}] RED: {'켜짐' if curr_red else '꺼짐'}, YELLOW: {'켜짐' if curr_yellow else '꺼짐'}, GREEN: {'켜짐' if curr_green else '꺼짐'}" status_messages.append(state_msg) continue target_green, target_yellow, target_red = curr_green, curr_yellow, curr_red if "ALL" in colors: target_green = target_yellow = target_red = (1 if action == 'ON' else 0) else: if "GREEN" in colors: target_green = (1 if action == 'ON' else 0) if "YELLOW" in colors: target_yellow = (1 if action == 'ON' else 0) if "RED" in colors: target_red = (1 if action == 'ON' else 0) target = (target_green, target_yellow, target_red) if target == (curr_green, curr_yellow, curr_red): already_set_count += 1 continue 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)) cursor.execute(""" INSERT INTO ai_control_log (req_text, dev_no, target_val) VALUES (%s, %s, %s) """, (req.message, dev_no, f"{color_name_str} {action}")) updated_sensors.append((led_sensor_no, target)) if action == 'STATUS': if not status_messages and not invalid_targets: return {"reply": "[상태 확인] 장비 상태를 읽을 수 없습니다."} reply_msg = "[상태 확인] 요청하신 장비의 현재 상태입니다.\n" if status_messages: reply_msg += "\n".join(status_messages) + "\n" if invalid_targets: reply_msg += f"- LED 미연결 장비: {', '.join(invalid_targets)}" return {"reply": reply_msg.strip()} if not updated_sensors: reply_msg = "" if already_set_count > 0: names = ", ".join([d['dev_name'] for d in target_devs if d['dev_name'] not in invalid_targets]) reply_msg = f"[LLM] '{names}' 장비의 경광등이 이미 해당 상태로 설정되어 있습니다. (제어 생략)\n" else: reply_msg = "[LLM] 제어할 수 있는 유효한 장비가 없습니다.\n" if invalid_targets: reply_msg += f"- LED 미연결 장비(제어 제외): {', '.join(invalid_targets)}" return {"reply": reply_msg.strip()} finally: conn.close() # 3. Handshake Waiting Loop (Agent 처리 대기) max_wait = 15 # 15초 waited = 0 target_names = ", ".join([d['dev_name'] for d in target_devs]) # 각 센서별(dev_name 포함) 완료 여부를 추적하기 위한 딕셔너리 # key: sensor_no, value: {"name": dev_name, "target": tgt, "done": False} tracking = {} for dev in target_devs: # target_devs 내에서 updated_sensors에 들어간 것만 필터링 for s_no, tgt in updated_sensors: if dev['led_sensorNo'] == s_no: tracking[s_no] = {"name": dev['dev_name'], "target": tgt, "done": False} break while waited < max_wait: conn = get_db() try: with conn.cursor() as cursor: all_done = True for s_no, info in tracking.items(): if info["done"]: continue cursor.execute("SELECT value_ch1_statusID, value_ch2_statusID, value_ch3_statusID FROM sensor_info WHERE sensor_no=%s", (s_no,)) res = cursor.fetchone() tgt = info["target"] if res and res['value_ch1_statusID'] == tgt[0] and res['value_ch2_statusID'] == tgt[1] and res['value_ch3_statusID'] == tgt[2]: info["done"] = True else: all_done = False if all_done: reply_msg = f"[LLM] 장비 제어 완료: {color_name_str} {action} 명령이 정상적으로 적용되었습니다.\n" if invalid_targets: reply_msg += f"- LED 미연결 장비(제어 제외): {', '.join(invalid_targets)}" return {"reply": reply_msg.strip()} finally: if conn: conn.close() await asyncio.sleep(1) waited += 1 # 시간 초과 시 성공/실패 분류 success_names = [info["name"] for info in tracking.values() if info["done"]] failed_names = [info["name"] for info in tracking.values() if not info["done"]] if success_names and failed_names: reply_msg = f"[LLM] 제어 명령이 부분적으로 적용되었습니다. (일부 장비 응답 지연)\n" elif failed_names and not success_names: reply_msg = f"[LLM] 제어 명령 전달 후 장비 응답이 지연되었습니다. 하드웨어를 확인해 주세요.\n" else: reply_msg = f"[LLM] 다중 장비 제어 처리 결과입니다.\n" if success_names: reply_msg += f"- 제어 완료: {', '.join(success_names)}\n" if failed_names: reply_msg += f"- 확인 필요(타임아웃): {', '.join(failed_names)}\n" if invalid_targets: reply_msg += f"- LED 미연결 장비(제어 제외): {', '.join(invalid_targets)}" return {"reply": reply_msg.strip()} @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)