Initial commit
This commit is contained in:
180
web_UI/app.py
Normal file
180
web_UI/app.py
Normal file
@@ -0,0 +1,180 @@
|
||||
"""
|
||||
LoadCell Weight Monitor - 실시간 무게 측정 웹 GUI
|
||||
Arduino에서 시리얼로 JSON 데이터를 받아 웹 브라우저에 실시간 그래프로 표시
|
||||
"""
|
||||
|
||||
from flask import Flask, render_template
|
||||
from flask_socketio import SocketIO
|
||||
import serial
|
||||
import serial.tools.list_ports
|
||||
import json
|
||||
import threading
|
||||
import time
|
||||
import logging
|
||||
|
||||
# Werkzeug HTTP 요청 로그 비활성화 (Socket.IO polling 로그 숨김)
|
||||
logging.getLogger('werkzeug').setLevel(logging.ERROR)
|
||||
|
||||
app = Flask(__name__)
|
||||
app.config['SECRET_KEY'] = 'loadcell_secret_2025'
|
||||
socketio = SocketIO(app, cors_allowed_origins="*")
|
||||
|
||||
# 전역 변수
|
||||
serial_port = None
|
||||
serial_thread = None
|
||||
is_running = False
|
||||
|
||||
# ===== Arduino 포트 자동 검색 =====
|
||||
def find_arduino_port():
|
||||
"""사용 가능한 Arduino 포트를 자동으로 찾습니다"""
|
||||
ports = serial.tools.list_ports.comports()
|
||||
for port in ports:
|
||||
# Arduino가 연결된 포트 찾기 (USB Serial, CH340, Arduino 등)
|
||||
if 'USB' in port.description or 'Arduino' in port.description or 'CH340' in port.description:
|
||||
return port.device
|
||||
|
||||
# 못 찾으면 사용 가능한 포트 목록 출력
|
||||
if ports:
|
||||
print("\n사용 가능한 포트:")
|
||||
for port in ports:
|
||||
print(f" - {port.device}: {port.description}")
|
||||
return ports[0].device # 첫 번째 포트 반환
|
||||
return None
|
||||
|
||||
# ===== 시리얼 데이터 읽기 스레드 =====
|
||||
def read_serial_data():
|
||||
"""Arduino로부터 시리얼 데이터를 읽어 WebSocket으로 전송"""
|
||||
global is_running, serial_port
|
||||
|
||||
error_count = 0
|
||||
max_errors = 5 # 연속 에러 허용 횟수
|
||||
|
||||
while is_running:
|
||||
try:
|
||||
if serial_port and serial_port.is_open:
|
||||
# in_waiting 호출을 안전하게 처리
|
||||
try:
|
||||
waiting = serial_port.in_waiting
|
||||
except Exception:
|
||||
waiting = 0
|
||||
|
||||
if waiting > 0:
|
||||
line = serial_port.readline().decode('utf-8', errors='ignore').strip()
|
||||
|
||||
# JSON 데이터 파싱
|
||||
if line.startswith('{') and line.endswith('}'):
|
||||
try:
|
||||
data = json.loads(line)
|
||||
socketio.emit('weight_data', data)
|
||||
print(f"Data: {data}")
|
||||
error_count = 0 # 성공 시 에러 카운트 리셋
|
||||
except json.JSONDecodeError:
|
||||
print(f"Invalid JSON: {line}")
|
||||
elif line: # 빈 줄이 아니면
|
||||
socketio.emit('log_message', {'message': line})
|
||||
print(f"Log: {line}")
|
||||
|
||||
time.sleep(0.02) # CPU 사용률 낮추기
|
||||
|
||||
except Exception as e:
|
||||
error_count += 1
|
||||
print(f"Serial error ({error_count}/{max_errors}): {e}")
|
||||
|
||||
if error_count >= max_errors:
|
||||
print("연속 에러 발생. 포트 재연결 시도...")
|
||||
socketio.emit('log_message', {'message': '시리얼 재연결 시도 중...'})
|
||||
|
||||
# 포트 재연결 시도
|
||||
try:
|
||||
if serial_port:
|
||||
serial_port.close()
|
||||
time.sleep(2)
|
||||
|
||||
port = find_arduino_port()
|
||||
if port:
|
||||
serial_port = serial.Serial(port, 9600, timeout=1)
|
||||
time.sleep(10)
|
||||
print(f"포트 재연결 성공: {port}")
|
||||
socketio.emit('log_message', {'message': f'재연결 성공: {port}'})
|
||||
else:
|
||||
print("Arduino 포트를 찾을 수 없습니다.")
|
||||
socketio.emit('log_message', {'message': 'Arduino 연결 실패'})
|
||||
except Exception as reconnect_error:
|
||||
print(f"재연결 실패: {reconnect_error}")
|
||||
|
||||
error_count = 0 # 재연결 시도 후 카운트 리셋
|
||||
|
||||
time.sleep(1)
|
||||
|
||||
# ===== Flask 라우트 =====
|
||||
@app.route('/')
|
||||
def loadcell_measure():
|
||||
"""메인 페이지"""
|
||||
return render_template('loadcell_measure.html')
|
||||
|
||||
# ===== SocketIO 이벤트 =====
|
||||
@socketio.on('connect')
|
||||
def handle_connect():
|
||||
"""클라이언트 연결"""
|
||||
print('Client connected')
|
||||
socketio.emit('log_message', {'message': 'Connected to server'})
|
||||
|
||||
@socketio.on('disconnect')
|
||||
def handle_disconnect():
|
||||
"""클라이언트 연결 해제"""
|
||||
print('Client disconnected')
|
||||
|
||||
@socketio.on('send_command')
|
||||
def handle_command(data):
|
||||
"""클라이언트에서 Arduino로 명령 전송"""
|
||||
global serial_port
|
||||
command = data.get('command', '')
|
||||
|
||||
if serial_port and serial_port.is_open:
|
||||
try:
|
||||
serial_port.write(command.encode('utf-8'))
|
||||
print(f"Sent command: {command}")
|
||||
socketio.emit('log_message', {'message': f'Command sent: {command}'})
|
||||
except Exception as e:
|
||||
print(f"Error sending command: {e}")
|
||||
socketio.emit('log_message', {'message': f'Error: {e}'})
|
||||
else:
|
||||
socketio.emit('log_message', {'message': 'Serial port not connected'})
|
||||
|
||||
# ===== 메인 실행 =====
|
||||
if __name__ == '__main__':
|
||||
# Arduino 포트 찾기
|
||||
port = find_arduino_port()
|
||||
|
||||
if port:
|
||||
print(f"\nArduino 포트 발견: {port}")
|
||||
try:
|
||||
# 시리얼 포트 열기
|
||||
serial_port = serial.Serial(port, 9600, timeout=1)
|
||||
time.sleep(5) # Arduino 리셋 대기
|
||||
print("시리얼 포트 연결 성공!")
|
||||
|
||||
# 시리얼 읽기 스레드 시작
|
||||
is_running = True
|
||||
serial_thread = threading.Thread(target=read_serial_data, daemon=True)
|
||||
serial_thread.start()
|
||||
|
||||
# Flask 서버 시작
|
||||
print("\n=================================")
|
||||
print("웹 서버 시작: http://localhost:5000")
|
||||
print("=================================\n")
|
||||
socketio.run(app, host='0.0.0.0', port=5000, debug=False)
|
||||
|
||||
except serial.SerialException as e:
|
||||
print(f"시리얼 포트 오류: {e}")
|
||||
print("Arduino가 제대로 연결되어 있는지 확인하세요.")
|
||||
except KeyboardInterrupt:
|
||||
print("\n서버 종료 중...")
|
||||
finally:
|
||||
is_running = False
|
||||
if serial_port and serial_port.is_open:
|
||||
serial_port.close()
|
||||
print("서버 종료 완료")
|
||||
else:
|
||||
print("Arduino 포트를 찾을 수 없습니다!")
|
||||
print("Arduino가 연결되어 있는지 확인하세요.")
|
||||
Reference in New Issue
Block a user