""" Session Manager for KB Capture (Simplified) One session = one continuous recording (with pause/resume). No clips, no keep/delete. Just record → transcribe → done. """ import json from pathlib import Path from dataclasses import dataclass from datetime import datetime from typing import Optional, List from enum import Enum class SessionType(Enum): DESIGN = "design" # CAD/Design KB ANALYSIS = "analysis" # FEA/Analysis KB class SessionStatus(Enum): RECORDING = "recording" PAUSED = "paused" TRANSCRIBING = "transcribing" READY = "ready" # Transcribed, ready for sync PROCESSED = "processed" # Clawdbot has processed it @dataclass class Session: """A recording session.""" id: str name: str project: str session_type: SessionType created_at: datetime duration: float = 0.0 status: SessionStatus = SessionStatus.RECORDING video_file: str = "recording.mp4" transcript_file: Optional[str] = None def to_dict(self) -> dict: return { "id": self.id, "name": self.name, "project": self.project, "session_type": self.session_type.value, "created_at": self.created_at.isoformat(), "duration": self.duration, "status": self.status.value, "video_file": self.video_file, "transcript_file": self.transcript_file, } @classmethod def from_dict(cls, data: dict) -> "Session": return cls( id=data["id"], name=data["name"], project=data["project"], session_type=SessionType(data.get("session_type", "design")), created_at=datetime.fromisoformat(data["created_at"]), duration=data.get("duration", 0.0), status=SessionStatus(data.get("status", "ready")), video_file=data.get("video_file", "recording.mp4"), transcript_file=data.get("transcript_file"), ) class SessionManager: """ Manages recording sessions. Project-centric structure: /Projects// └── _capture/ └── / ├── session.json # Metadata ├── recording.mp4 # Video └── transcript.json # Whisper output """ def __init__(self, projects_root: Path): self.projects_root = Path(projects_root) self.current_session: Optional[Session] = None self._current_project_path: Optional[Path] = None def list_projects(self) -> List[str]: """List available projects.""" projects = [] if self.projects_root.exists(): for p in sorted(self.projects_root.iterdir()): if p.is_dir() and not p.name.startswith((".", "_")): # Check if it looks like a project if (p / "KB").exists() or (p / "_context.md").exists(): projects.append(p.name) return projects def get_project_path(self, project: str) -> Path: """Get full path to a project.""" return self.projects_root / project def start_session( self, name: str, project: str, session_type: SessionType = SessionType.DESIGN, ) -> Session: """Start a new recording session.""" self._current_project_path = self.get_project_path(project) if not self._current_project_path.exists(): raise ValueError(f"Project not found: {project}") session_id = datetime.now().strftime("%Y%m%d-%H%M%S") session = Session( id=session_id, name=name, project=project, session_type=session_type, created_at=datetime.now(), status=SessionStatus.RECORDING, ) # Create session directory session_dir = self._current_project_path / "_capture" / session_id session_dir.mkdir(parents=True, exist_ok=True) self.current_session = session self._save_session() return session def get_session_dir(self) -> Path: """Get current session directory.""" if not self.current_session or not self._current_project_path: raise RuntimeError("No active session") return self._current_project_path / "_capture" / self.current_session.id def get_video_path(self) -> Path: """Get path for video file.""" return self.get_session_dir() / self.current_session.video_file def update_status(self, status: SessionStatus) -> None: """Update session status.""" if self.current_session: self.current_session.status = status self._save_session() def set_duration(self, duration: float) -> None: """Set recording duration.""" if self.current_session: self.current_session.duration = duration self._save_session() def set_transcript(self, transcript_file: str) -> None: """Set transcript file name.""" if self.current_session: self.current_session.transcript_file = transcript_file self._save_session() def end_session(self) -> Session: """End current session.""" if not self.current_session: raise RuntimeError("No active session") self.current_session.status = SessionStatus.READY self._save_session() session = self.current_session self.current_session = None self._current_project_path = None return session def cancel_session(self) -> None: """Cancel session and delete files.""" if self.current_session: import shutil session_dir = self.get_session_dir() if session_dir.exists(): shutil.rmtree(session_dir) self.current_session = None self._current_project_path = None def list_sessions(self, project: Optional[str] = None) -> List[Session]: """List sessions for a project or all projects.""" sessions = [] if project: capture_dir = self.get_project_path(project) / "_capture" if capture_dir.exists(): for session_dir in sorted(capture_dir.iterdir(), reverse=True): if session_dir.is_dir(): session_file = session_dir / "session.json" if session_file.exists(): try: with open(session_file) as f: sessions.append(Session.from_dict(json.load(f))) except: pass else: for proj in self.list_projects(): sessions.extend(self.list_sessions(proj)) sessions.sort(key=lambda s: s.created_at, reverse=True) return sessions def _save_session(self) -> None: """Save current session to disk.""" if not self.current_session: return session_file = self.get_session_dir() / "session.json" with open(session_file, "w") as f: json.dump(self.current_session.to_dict(), f, indent=2)