OpenAdapt Tray: Logging & Action Storage¶

# Environment variable override
OPENADAPT_TRAY_LOG_LEVEL=DEBUG

# Or via config.json
{
    "logging": {
        "level": "INFO",
        "console": false,
        "file": true
    }
}

import logging
from logging.handlers import RotatingFileHandler
from pathlib import Path
import platform
import sys

def get_log_directory() -> Path:
    """Get platform-appropriate log directory."""
    if platform.system() == "Darwin":
        base = Path.home() / "Library" / "Application Support"
    elif platform.system() == "Windows":
        base = Path(os.environ.get("APPDATA", Path.home() / "AppData" / "Roaming"))
    else:  # Linux and others
        base = Path(os.environ.get("XDG_DATA_HOME", Path.home() / ".local" / "share"))

    log_dir = base / "OpenAdapt" / "logs"
    log_dir.mkdir(parents=True, exist_ok=True)
    return log_dir

def setup_logging(level: str = "INFO") -> logging.Logger:
    """Configure logging for the tray application."""
    logger = logging.getLogger("openadapt.tray")
    logger.setLevel(getattr(logging, level.upper()))

    # File handler with rotation
    log_file = get_log_directory() / "openadapt-tray.log"
    file_handler = RotatingFileHandler(
        log_file,
        maxBytes=10 * 1024 * 1024,  # 10 MB
        backupCount=5,
        encoding="utf-8",
    )
    file_handler.setFormatter(logging.Formatter(
        "{asctime} | {levelname:8} | {name}:{funcName}:{lineno} - {message}",
        style="{",
        datefmt="%Y-%m-%d %H:%M:%S",
    ))
    logger.addHandler(file_handler)

    return logger

from pathlib import Path
import sqlite3
from typing import Optional
import json

class ActionHistorySync:
    """Sync tray action history with capture database."""

    def __init__(self, tray_db_path: Path, capture_db_path: Optional[Path] = None):
        self.tray_db = tray_db_path
        self.capture_db = capture_db_path

    def sync_session(self, session_id: str) -> bool:
        """Sync a completed session to capture database."""
        if not self.capture_db or not self.capture_db.exists():
            return False

        with sqlite3.connect(self.tray_db) as tray_conn:
            # Get unsynced actions for this session
            actions = tray_conn.execute(
                """
                SELECT id, timestamp, action_type, data
                FROM action_history
                WHERE session_id = ? AND synced = 0
                ORDER BY timestamp
                """,
                (session_id,)
            ).fetchall()

        if not actions:
            return True

        with sqlite3.connect(self.capture_db) as capture_conn:
            # Insert into capture database's session_events table
            for action_id, timestamp, action_type, data in actions:
                capture_conn.execute(
                    """
                    INSERT INTO session_events (timestamp, event_type, data, source)
                    VALUES (?, ?, ?, 'tray')
                    """,
                    (timestamp, action_type, data)
                )
            capture_conn.commit()

        # Mark as synced
        with sqlite3.connect(self.tray_db) as tray_conn:
            tray_conn.executemany(
                "UPDATE action_history SET synced = 1 WHERE id = ?",
                [(a[0],) for a in actions]
            )
            tray_conn.commit()

        return True

# Initialize telemetry in tray app
from openadapt_telemetry import get_telemetry

def init_app():
    """Initialize the tray application."""
    telemetry = get_telemetry()
    telemetry.initialize(
        package_name="openadapt-tray",
        package_version=__version__,
    )

from openadapt_telemetry import get_telemetry, track_errors

class TrayApp:
    @track_errors(reraise=True)
    def start_recording(self, task_name: str) -> None:
        """Start a recording session."""
        try:
            # Recording logic...
            pass
        except PermissionError as e:
            get_telemetry().capture_exception(e, tags={
                "action": "start_recording",
                "platform": platform.system(),
            })
            raise

from openadapt_telemetry import get_telemetry

def track_recording_session(duration: float, success: bool, frame_count: int):
    """Track recording session metrics (opt-in only)."""
    telemetry = get_telemetry()

    if not telemetry.is_analytics_enabled():
        return

    telemetry.capture_event(
        "tray.recording_session",
        {
            "duration_seconds": round(duration, 1),
            "success": success,
            "frame_count_bucket": bucket_count(frame_count),  # 0-10, 10-50, 50-100, 100+
        }
    )

def bucket_count(count: int) -> str:
    """Bucket counts to avoid exact numbers (privacy)."""
    if count <= 10:
        return "0-10"
    elif count <= 50:
        return "10-50"
    elif count <= 100:
        return "50-100"
    else:
        return "100+"

def open_data_directory():
    """Open the OpenAdapt data directory in file explorer."""
    import subprocess
    import platform

    data_dir = get_data_directory()

    if platform.system() == "Darwin":
        subprocess.run(["open", str(data_dir)])
    elif platform.system() == "Windows":
        subprocess.run(["explorer", str(data_dir)])
    else:
        subprocess.run(["xdg-open", str(data_dir)])

def clear_all_data(keep_config: bool = True):
    """Delete all OpenAdapt local data."""
    data_dir = get_data_directory()

    for item in data_dir.iterdir():
        if keep_config and item.name == "config.json":
            continue
        if item.is_dir():
            shutil.rmtree(item)
        else:
            item.unlink()

    logger.info("All local data cleared")

import os
import platform
from pathlib import Path
from typing import Dict

def get_storage_paths() -> Dict[str, Path]:
    """Get all storage paths for the current platform."""

    if platform.system() == "Darwin":
        base = Path.home() / "Library" / "Application Support" / "OpenAdapt"
    elif platform.system() == "Windows":
        appdata = os.environ.get("APPDATA", Path.home() / "AppData" / "Roaming")
        base = Path(appdata) / "OpenAdapt"
    else:  # Linux and others
        xdg_data = os.environ.get("XDG_DATA_HOME", Path.home() / ".local" / "share")
        base = Path(xdg_data) / "openadapt"

    paths = {
        "base": base,
        "logs": base / "logs",
        "crash_logs": base / "logs" / "crash",
        "config": base / "config.json",
        "history_db": base / "history.db",
        "cache": base / "cache",
        "state": base / "state",
    }

    # Ensure directories exist
    for key, path in paths.items():
        if key not in ("config", "history_db"):  # Don't create files
            path.mkdir(parents=True, exist_ok=True)

    return paths

# pyproject.toml
[project]
dependencies = [
    "openadapt-telemetry>=0.1.0",
]

from openadapt_capture import RecordingSession, CaptureConfig
from openadapt_tray.history import ActionHistory

class TrayRecordingController:
    """Bridge between tray UI and capture backend."""

    def __init__(self):
        self.history = ActionHistory()
        self.current_session: Optional[RecordingSession] = None

    def start_recording(self, task_name: str, config: CaptureConfig) -> str:
        """Start a new recording session."""
        import uuid

        session_id = str(uuid.uuid4())

        # Log to action history
        self.history.log_action(
            action_type="recording.start",
            session_id=session_id,
            data={"task_name": task_name, "config": config.to_dict()}
        )

        # Start capture backend
        self.current_session = RecordingSession(
            session_id=session_id,
            task_name=task_name,
            config=config,
            on_error=self._on_capture_error,
        )
        self.current_session.start()

        return session_id

    def stop_recording(self) -> dict:
        """Stop the current recording session."""
        if not self.current_session:
            return {"error": "No active session"}

        result = self.current_session.stop()

        # Log completion
        self.history.log_action(
            action_type="recording.stop",
            session_id=self.current_session.session_id,
            data={
                "duration": result.duration,
                "frame_count": result.frame_count,
                "success": result.success,
            }
        )

        # Sync with capture database
        self.history.sync_session(self.current_session.session_id)

        self.current_session = None
        return result.to_dict()

    def _on_capture_error(self, error: Exception):
        """Handle capture errors."""
        get_telemetry().capture_exception(error)
        self.history.log_action(
            action_type="error.occurred",
            session_id=self.current_session.session_id if self.current_session else None,
            data={"error_type": type(error).__name__}
        )

from openadapt_ml import TrainingJob, TrainingStatus
from openadapt_tray.notifications import show_notification

class TrayTrainingController:
    """Bridge between tray UI and ML training backend."""

    def __init__(self):
        self.history = ActionHistory()
        self.current_job: Optional[TrainingJob] = None

    def start_training(self, model_type: str, demo_ids: list[str]) -> str:
        """Start a training job."""
        job_id = str(uuid.uuid4())

        self.history.log_action(
            action_type="training.start",
            data={
                "job_id": job_id,
                "model_type": model_type,
                "demo_count": len(demo_ids),
            }
        )

        self.current_job = TrainingJob(
            job_id=job_id,
            model_type=model_type,
            demo_ids=demo_ids,
            on_progress=self._on_training_progress,
            on_complete=self._on_training_complete,
            on_error=self._on_training_error,
        )
        self.current_job.start()

        return job_id

    def _on_training_progress(self, progress: float, message: str):
        """Handle training progress updates."""
        # Update tray icon or menu with progress
        pass

    def _on_training_complete(self, result: TrainingStatus):
        """Handle training completion."""
        self.history.log_action(
            action_type="training.complete",
            data={
                "job_id": self.current_job.job_id,
                "duration": result.duration,
                "success": result.success,
            }
        )

        show_notification(
            title="Training Complete",
            message=f"Model trained successfully in {result.duration:.1f}s"
        )

        # Track telemetry (anonymous)
        get_telemetry().capture_event(
            "tray.training_complete",
            {"model_type": self.current_job.model_type, "success": True}
        )

    def _on_training_error(self, error: Exception):
        """Handle training errors."""
        get_telemetry().capture_exception(error)

        self.history.log_action(
            action_type="training.error",
            data={
                "job_id": self.current_job.job_id if self.current_job else None,
                "error_type": type(error).__name__,
            }
        )

        show_notification(
            title="Training Failed",
            message="An error occurred during training. Check logs for details."
        )

from pathlib import Path
from typing import Iterator, NamedTuple
from datetime import datetime

class LogEntry(NamedTuple):
    timestamp: datetime
    level: str
    source: str  # tray, capture, ml, etc.
    message: str

def aggregate_logs(max_entries: int = 1000) -> Iterator[LogEntry]:
    """Aggregate logs from all OpenAdapt components."""

    log_sources = {
        "tray": get_storage_paths()["logs"] / "openadapt-tray.log",
        "capture": get_capture_log_path(),  # From openadapt-capture
        "ml": get_ml_log_path(),            # From openadapt-ml
    }

    entries = []

    for source, log_path in log_sources.items():
        if not log_path.exists():
            continue

        with open(log_path, "r") as f:
            for line in f:
                try:
                    entry = parse_log_line(line, source)
                    if entry:
                        entries.append(entry)
                except Exception:
                    continue

    # Sort by timestamp and return most recent
    entries.sort(key=lambda e: e.timestamp, reverse=True)
    return iter(entries[:max_entries])