test commit

apiserver.py

@@ -0,0 +1,153 @@
+
+#!/usr/bin/env python3
+
+"""
+FASTAPI + Socket.IO 서버 - LoadCell 실시간 데이터 중계
+센서 데이터를 웹에 실시간 전달 (Socket.IO)
+"""
+
+import os
+import socketio
+from fastapi import FastAPI
+from fastapi.staticfiles import StaticFiles
+from fastapi.responses import FileResponse
+from pydantic import BaseModel
+
+# ===== Socket.IO 서버 생성 =====
+sio = socketio.AsyncServer(async_mode="asgi", cors_allowed_origins="*")
+
+# ===== FastAPI 인스턴스 생성 =====
+fastapi_app = FastAPI(
+    title="LoadCell API Server",
+    description="HX711 로드셀 측정 데이터 API + Socket.IO",
+    version="2.0.0",
+)
+
+# Socket.IO ASGI 앱 (FastAPI와 결합)
+app = socketio.ASGIApp(sio, fastapi_app)
+
+# 정적 파일 서빙
+if os.path.exists("static"):
+    fastapi_app.mount("/static", StaticFiles(directory="static"), name="static")
+
+# 공유 데이터 (main.py에서 set_shared_data()로 설정)
+shared_data = None
+
+
+def set_shared_data(data: dict):
+    """main.py에서 호출하여 측정 데이터 연결"""
+    global shared_data
+    shared_data = data
+
+
+# ===== 요청 모델 정의 =====
+class CommandRequest(BaseModel):
+    """명령어 요청 모델"""
+
+    command: str  # 'tare', 'pause', 'resume'
+
+
+# ===== FastAPI 라우트 =====
+@fastapi_app.get("/")
+def index():
+    """메인 페이지"""
+    return FileResponse("web.html")
+
+
+@fastapi_app.get("/uro/data")
+def get_data():
+    """REST API로 현재 데이터 조회 (폴백용)"""
+    if shared_data:
+        return shared_data
+    return {"volume_ml": 0, "time": 0, "status": "대기 중"}
+
+
+@fastapi_app.post("/uro/command")
+def handle_command(req: CommandRequest):
+    """REST API로 명령어 처리"""
+    import rapimeasure
+
+    return process_command(req.command)
+
+
+def process_command(command: str):
+    """명령어 처리 공통 함수"""
+    import rapimeasure
+    if command == "tare":
+        rapimeasure.tare()
+        return {"success": True, "message": "영점 조정 완료"}
+
+    elif command == "pause":
+        rapimeasure.set_paused(True)
+        if shared_data:
+            shared_data["status"] = "일시정지"
+        return {"success": True, "message": "측정 일시 정지"}
+
+    elif command == "resume":
+        rapimeasure.set_paused(False)
+        if shared_data:
+            shared_data["status"] = "측정 중"
+        return {"success": True, "message": "측정 재시작"}
+
+    return {"success": False, "message": f"알 수 없는 명령: {command}"}
+
+
+# ===== Socket.IO 이벤트 핸들러 =====
+@sio.event
+async def connect(sid, environ):
+    """클라이언트 연결"""
+    print(f"[Socket.IO] Client connected: {sid}")
+    await sio.emit("log_message", {"message": "서버에 연결되었습니다."}, to=sid)
+
+
+@sio.event
+async def disconnect(sid):
+    """클라이언트 연결 해제"""
+    print(f"[Socket.IO] Client disconnected: {sid}")
+
+
+@sio.event
+async def send_command(sid, data):
+    """클라이언트에서 명령어 수신"""
+    command = data.get("command", "")
+    print(f"[Socket.IO] Command from {sid}: {command}")
+
+    result = process_command(command)
+    await sio.emit("command_result", result, to=sid)
+
+
+@sio.event
+async def set_density(sid, data):
+    """밀도 설정"""
+    import rapimeasure
+
+    density = data.get("density", 1.0)
+    rapimeasure.set_density(density)
+    print(f"[Socket.IO] Density set to: {density}")
+    await sio.emit("log_message", {"message": f"밀도 설정: {density} g/mL"}, to=sid)
+
+
+@sio.event
+async def reset_measurement(sid):
+    """측정 초기화"""
+    print(f"[Socket.IO] Reset measurement from {sid}")
+    await sio.emit("log_message", {"message": "측정 데이터 초기화"}, to=sid)
+
+
+# ===== 외부에서 호출할 emit 함수 =====
+async def emit_weight_data(data: dict):
+    """측정 데이터를 모든 클라이언트에게 브로드캐스트"""
+    # 디버깅: 데이터 전송 로그 출력 (너무 많으면 주석 처리)
+    # print(f"[Socket.IO] Emitting data: {data['weight_g']}g")
+    await sio.emit("weight_data", data)
+
+
+# ===== 직접 실행 (테스트용) =====
+if __name__ == "__main__":
+    import uvicorn
+
+    # 더미 데이터
+    shared_data = {"volume_ml": 0, "time": 0, "status": "테스트 모드"}
+
+    print("API Server Test Mode")
+    uvicorn.run(app, host="0.0.0.0", port=8080)