509 lines
22 KiB
Python
509 lines
22 KiB
Python
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)
|