完善夹爪界面、跨平台打包与运行日志

gripper/common/runtime_logging.py

@@ -0,0 +1,105 @@
+import atexit
+import os
+import shutil
+import sys
+import traceback
+from datetime import datetime
+from pathlib import Path
+
+
+_LOG_STREAM = None
+
+
+def _cleanup_old_sessions(log_root, keep_sessions):
+    sessions = sorted(
+        (path for path in log_root.iterdir() if path.is_dir()),
+        key=lambda path: path.stat().st_mtime,
+        reverse=True,
+    )
+    for old_session in sessions[max(1, int(keep_sessions)) :]:
+        try:
+            shutil.rmtree(old_session)
+        except OSError:
+            pass
+
+
+def _redirect_standard_streams(stream):
+    for descriptor in (1, 2):
+        try:
+            os.dup2(stream.fileno(), descriptor)
+        except OSError:
+            pass
+    sys.stdout = stream
+    sys.stderr = stream
+
+
+def _install_exception_hooks(stream):
+    def handle_exception(exc_type, exc_value, exc_traceback):
+        print("\n[unhandled exception]", file=stream)
+        traceback.print_exception(exc_type, exc_value, exc_traceback, file=stream)
+        stream.flush()
+
+    sys.excepthook = handle_exception
+
+    try:
+        import threading
+
+        def handle_thread_exception(args):
+            print(f"\n[thread exception] name={args.thread.name}", file=stream)
+            traceback.print_exception(
+                args.exc_type,
+                args.exc_value,
+                args.exc_traceback,
+                file=stream,
+            )
+            stream.flush()
+
+        threading.excepthook = handle_thread_exception
+    except (AttributeError, ImportError):
+        pass
+
+
+def configure_frozen_logging(app_name="GripperDemo02", keep_sessions=10):
+    global _LOG_STREAM
+    if _LOG_STREAM is not None:
+        return Path(_LOG_STREAM.name)
+    if not getattr(sys, "frozen", False):
+        return None
+
+    app_dir = Path(sys.executable).resolve().parent
+    log_root = app_dir / "logs"
+    log_root.mkdir(parents=True, exist_ok=True)
+
+    session_env = f"{app_name.upper()}_LOG_SESSION"
+    session_name = os.environ.get(session_env)
+    is_main_process = not session_name
+    if is_main_process:
+        session_name = datetime.now().strftime("%Y%m%d_%H%M%S") + f"_{os.getpid()}"
+        os.environ[session_env] = session_name
+        _cleanup_old_sessions(log_root, keep_sessions)
+
+    session_dir = log_root / session_name
+    session_dir.mkdir(parents=True, exist_ok=True)
+    log_name = "main.log" if is_main_process else f"worker_{os.getpid()}.log"
+    log_path = session_dir / log_name
+    _LOG_STREAM = open(log_path, "a", encoding="utf-8", buffering=1)
+    _redirect_standard_streams(_LOG_STREAM)
+    _install_exception_hooks(_LOG_STREAM)
+
+    if is_main_process:
+        try:
+            (log_root / "latest.txt").write_text(session_name + "\n", encoding="utf-8")
+        except OSError:
+            pass
+
+    print("=" * 72)
+    print(f"{app_name} runtime log")
+    print(f"time={datetime.now().isoformat(timespec='seconds')}")
+    print(f"pid={os.getpid()} role={'main' if is_main_process else 'worker'}")
+    print(f"executable={sys.executable}")
+    print(f"log={log_path}")
+    print(f"argv={sys.argv!r}")
+    print("=" * 72)
+
+    atexit.register(_LOG_STREAM.flush)
+    return log_path