rf server 수정

rpiKivyStreamDeck/mk_rf_server.py

@@ -0,0 +1,150 @@
+import bluetooth
+import subprocess
+import webbrowser
+import GPUtil
+import psutil
+import time
+import signal
+import sys
+import os
+
+# 블루투스 서버 설정
+HOST_MAC = '' # 모든 기기로부터 연결 허용
+PORT = 5
+
+def signal_handler(sig, frame):
+    print("Ctrl+C가 감지되었습니다. 서버를 종료합니다.")
+    sys.exit(0)
+
+signal.signal(signal.SIGINT, signal_handler)
+
+def format_uptime():
+    uptime_seconds = time.time() - psutil.boot_time()
+    days = int(uptime_seconds // 86400)
+    hours = int((uptime_seconds % 86400) // 3600)
+    minutes = int((uptime_seconds % 3600) // 60)
+    return f"{days}d {hours:02d}h {minutes:02d}m" if days > 0 else f"{hours:02d}h {minutes:02d}m"
+
+def get_system_status():
+    """시스템 자원 사용량 정보를 문자열로 반환합니다."""
+    cpu_usage = psutil.cpu_percent(interval=1)
+    ram_usage = psutil.virtual_memory().percent
+    gpu_temp = GPUtil.getGPUs()[0].temperature
+    disk_usage = psutil.disk_usage('/').percent
+    # NPU 사용량은 표준 라이브러리로 얻기 어려워 0으로 고정
+    return f"{format_uptime()} | CPU: {cpu_usage}% | RAM: {ram_usage}% | GPU: {gpu_temp}℃| Disk: {disk_usage}%"
+
+def run_command(command):
+    """지정된 명령어를 실행하고 결과를 반환합니다."""
+    if command == "youtube":
+        try:
+            webbrowser.open("https://youtube.com")
+            return "Youyube앱이 시작되었습니다."
+        except Exception as e:
+            return f"Youyube 실행 실패: {e}"
+    elif command == "copilot":
+        try:
+            webbrowser.open("https://copilot.microsoft.com/chats/")
+            return "copilot앱이 시작되었습니다."
+        except Exception as e:
+            return f"copilot 실행 실패: {e}"
+    elif command == "gemini":
+        try:
+            os.system("start chrome -new-window https://gemini.google.com")
+            return "Gemini 앱이 시작되었습니다."
+        except Exception as e:
+            return f"Gemini 실행 실패: {e}"
+    elif command == "chatgpt":
+        try:
+            webbrowser.open("https://chatgpt.com/")
+            return "chatgpt앱이 시작되었습니다."
+        except Exception as e:
+            return f"chatgpt 실행 실패: {e}"
+    elif command == "gitea":
+        try:
+            os.system("start firefox -new-window https://gitea.mkaidevops.xyz")
+            return "gitea앱이 시작되었습니다."
+        except Exception as e:
+            return f"gitea 실행 실패: {e}"
+    elif command == "redmine":
+        try:
+            os.system("start firefox -new-window https://redmine.mkaidevops.xyz")
+            return "redmine앱이 시작되었습니다."
+        except Exception as e:
+            return f"redmine 실행 실패: {e}"
+    elif command == "github":
+        try:
+            os.system("start chrome -new-window https://redmine.mkaidevops.xyz")
+            return "github앱이 시작되었습니다."
+        except Exception as e:
+            return f"github 실행 실패: {e}"
+    elif command == "taskmgr":
+        os.system("taskmgr")
+        return("run taskmgr")
+    elif command == "get_status":
+        return get_system_status()
+    else:
+        return f"알 수 없는 명령어: {command}"
+
+def main():
+    try:
+        server_sock = bluetooth.BluetoothSocket(bluetooth.RFCOMM)
+        server_sock.bind((HOST_MAC, PORT))
+        server_sock.listen(1)
+        
+        print("Bluetooth 서버 대기 중...")
+        print(f"Server Port: {PORT}")
+
+        client_sock = None
+        
+        while True:
+            try:
+                # 1초 타임아웃을 설정하여 블로킹을 방지하고 Ctrl+C를 감지합니다.
+                server_sock.settimeout(1.0)
+                client_sock, client_info = server_sock.accept()
+                print(f"클라이언트 연결됨: {client_info}")
+                client_sock.settimeout(5.0) # 연결 후 통신 타임아웃을 5초로 설정
+                
+                while True:
+                    try:
+                        print("[DEBUG] 클라이언트로부터 데이터 수신 대기 중...")
+                        data = client_sock.recv(1024)
+                        if not data:
+                            print("[DEBUG] 클라이언트가 연결을 종료했습니다.")
+                            break
+                        
+                        command = data.decode('utf-8').strip()
+                        print(f"[DEBUG] 명령어 수신: {command}")
+                        
+                        response = run_command(command)
+                        client_sock.send(response.encode('utf-8'))
+                        print(f"[DEBUG] 응답 전송: {response}")
+
+                    except bluetooth.btcommon.BluetoothError as e:
+                        print(f"[DEBUG] 통신 오류: {e}")
+                        break # 내부 루프 종료
+                    except Exception as e:
+                        print(f"[DEBUG] 예상치 못한 통신 오류: {e}")
+                        break # 내부 루프 종료
+
+            except bluetooth.btcommon.BluetoothError as e:
+                # 타임아웃 오류는 무시하고 루프를 계속합니다.
+                if "timed out" in str(e):
+                    continue
+                print(f"예상치 못한 오류: {e}")
+                
+            finally:
+                if client_sock:
+                    print("[DEBUG] 클라이언트 소켓 닫는 중...")
+                    client_sock.close()
+                print("[DEBUG] 연결 종료.")
+
+    except Exception as e:
+        print(f"서버 초기화 오류: {e}")
+    finally:
+        server_sock.close()
+        print("서버 소켓 종료.")
+
+if __name__ == "__main__":
+    main()
+