92 lines
3.7 KiB
Python
92 lines
3.7 KiB
Python
import time
|
|
import pymysql
|
|
import datetime
|
|
|
|
DB_CONFIG = {
|
|
'host': 'localhost',
|
|
'user': 'root',
|
|
'password': 'password',
|
|
'database': 'mmcl_db',
|
|
'autocommit': True
|
|
}
|
|
|
|
# ==========================================
|
|
# [하드웨어 제어부]
|
|
# 라즈베리파이 등 환경에 맞게 GPIO 또는 Serial 연동
|
|
# ==========================================
|
|
def set_physical_led(green: int, yellow: int, red: int):
|
|
# import RPi.GPIO as GPIO
|
|
# GPIO.output(PIN_GREEN, green)
|
|
# GPIO.output(PIN_YELLOW, yellow)
|
|
# GPIO.output(PIN_RED, red)
|
|
color = "OFF"
|
|
if red: color = "RED"
|
|
elif yellow: color = "YELLOW"
|
|
elif green: color = "GREEN"
|
|
print(f"[HW Relay Control] 물리적 제어 완료: {color} 상태 켜짐")
|
|
|
|
def get_db_connection():
|
|
try:
|
|
return pymysql.connect(**DB_CONFIG)
|
|
except Exception as e:
|
|
print(f"DB Connection Error: {e}")
|
|
return None
|
|
|
|
def run_led_agent():
|
|
print("Starting LED (Warning Light) Control Agent...")
|
|
while True:
|
|
conn = get_db_connection()
|
|
if not conn:
|
|
time.sleep(2)
|
|
continue
|
|
|
|
try:
|
|
with conn.cursor() as cursor:
|
|
# Target 값이 존재하면서 Current와 다를 때만 조회 (핸드쉐이크 요청 감지)
|
|
sql = """SELECT sensor_no,
|
|
target_ch1_statusID, value_ch1_statusID,
|
|
target_ch2_statusID, value_ch2_statusID,
|
|
target_ch3_statusID, value_ch3_statusID
|
|
FROM sensor_info
|
|
WHERE sensor_typeid = 2
|
|
AND (
|
|
(target_ch1_statusID IS NOT NULL AND target_ch1_statusID != IFNULL(value_ch1_statusID, -1)) OR
|
|
(target_ch2_statusID IS NOT NULL AND target_ch2_statusID != IFNULL(value_ch2_statusID, -1)) OR
|
|
(target_ch3_statusID IS NOT NULL AND target_ch3_statusID != IFNULL(value_ch3_statusID, -1))
|
|
)"""
|
|
cursor.execute(sql)
|
|
tasks = cursor.fetchall()
|
|
|
|
for task in tasks:
|
|
sensor_no = task[0]
|
|
t_ch1, v_ch1 = task[1], task[2] # Green
|
|
t_ch2, v_ch2 = task[3], task[4] # Yellow
|
|
t_ch3, v_ch3 = task[5], task[6] # Red
|
|
|
|
print(f"\n[Agent] 제어 명령 감지! (Sensor No: {sensor_no})")
|
|
|
|
# 1. 물리적 하드웨어 제어
|
|
set_physical_led(t_ch1, t_ch2, t_ch3)
|
|
time.sleep(0.5) # 물리 스위칭 여유 시간
|
|
|
|
# 2. 제어 성공 후 DB에 상태(Current) 동기화 (핸드쉐이크 완료 응답)
|
|
new_v1 = t_ch1 if t_ch1 is not None else v_ch1
|
|
new_v2 = t_ch2 if t_ch2 is not None else v_ch2
|
|
new_v3 = t_ch3 if t_ch3 is not None else v_ch3
|
|
|
|
update_sql = """UPDATE sensor_info
|
|
SET value_ch1_statusID=%s, value_ch2_statusID=%s, value_ch3_statusID=%s, update_time=%s
|
|
WHERE sensor_no=%s"""
|
|
cursor.execute(update_sql, (new_v1, new_v2, new_v3, datetime.datetime.now(), sensor_no))
|
|
print(f"[Agent] DB 상태 동기화 업데이트 완료 -> 파이썬 백엔드로 신호 전달됨")
|
|
|
|
except Exception as e:
|
|
print(f"Error: {e}")
|
|
finally:
|
|
conn.close()
|
|
|
|
time.sleep(1) # 1초 주기로 Polling
|
|
|
|
if __name__ == "__main__":
|
|
run_led_agent()
|