235 lines
7.5 KiB
Python
235 lines
7.5 KiB
Python
#!/usr/bin/env python3
|
|
"""
|
|
Bannerlord Session Trace Logger — First-Replayable Training Material
|
|
|
|
Captures one Bannerlord session as a replayable trace:
|
|
- Timestamps on every cycle
|
|
- Actions executed with success/failure
|
|
- World-state evidence (screenshots, Steam stats)
|
|
- Hermes session/log ID mapping
|
|
|
|
Storage: ~/.timmy/traces/bannerlord/trace_<session_id>.jsonl
|
|
Manifest: ~/.timmy/traces/bannerlord/manifest_<session_id>.json
|
|
|
|
Each JSONL line is one ODA cycle with full context.
|
|
The manifest bundles metadata for replay/eval.
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import json
|
|
import time
|
|
import uuid
|
|
from dataclasses import dataclass, field, asdict
|
|
from datetime import datetime, timezone
|
|
from pathlib import Path
|
|
from typing import Optional
|
|
|
|
# Storage root — local-first under ~/.timmy/
|
|
DEFAULT_TRACE_DIR = Path.home() / ".timmy" / "traces" / "bannerlord"
|
|
|
|
|
|
@dataclass
|
|
class CycleTrace:
|
|
"""One ODA cycle captured in full."""
|
|
cycle_index: int
|
|
timestamp_start: str
|
|
timestamp_end: str = ""
|
|
duration_ms: int = 0
|
|
|
|
# Observe
|
|
screenshot_path: str = ""
|
|
window_found: bool = False
|
|
screen_size: list[int] = field(default_factory=lambda: [1920, 1080])
|
|
mouse_position: list[int] = field(default_factory=lambda: [0, 0])
|
|
playtime_hours: float = 0.0
|
|
players_online: int = 0
|
|
is_running: bool = False
|
|
|
|
# Decide
|
|
actions_planned: list[dict] = field(default_factory=list)
|
|
decision_note: str = ""
|
|
|
|
# Act
|
|
actions_executed: list[dict] = field(default_factory=list)
|
|
actions_succeeded: int = 0
|
|
actions_failed: int = 0
|
|
|
|
# Metadata
|
|
hermes_session_id: str = ""
|
|
hermes_log_id: str = ""
|
|
harness_session_id: str = ""
|
|
|
|
def to_dict(self) -> dict:
|
|
return asdict(self)
|
|
|
|
|
|
@dataclass
|
|
class SessionManifest:
|
|
"""Top-level metadata for a captured session trace."""
|
|
trace_id: str
|
|
harness_session_id: str
|
|
hermes_session_id: str
|
|
hermes_log_id: str
|
|
game: str = "Mount & Blade II: Bannerlord"
|
|
app_id: int = 261550
|
|
started_at: str = ""
|
|
finished_at: str = ""
|
|
total_cycles: int = 0
|
|
total_actions: int = 0
|
|
total_succeeded: int = 0
|
|
total_failed: int = 0
|
|
trace_file: str = ""
|
|
trace_dir: str = ""
|
|
replay_command: str = ""
|
|
eval_note: str = ""
|
|
|
|
def to_dict(self) -> dict:
|
|
return asdict(self)
|
|
|
|
|
|
class BannerlordTraceLogger:
|
|
"""
|
|
Captures a single Bannerlord session as a replayable trace.
|
|
|
|
Usage:
|
|
logger = BannerlordTraceLogger(hermes_session_id="abc123")
|
|
logger.start_session()
|
|
cycle = logger.begin_cycle(0)
|
|
# ... populate cycle fields ...
|
|
logger.finish_cycle(cycle)
|
|
manifest = logger.finish_session()
|
|
"""
|
|
|
|
def __init__(
|
|
self,
|
|
trace_dir: Optional[Path] = None,
|
|
harness_session_id: str = "",
|
|
hermes_session_id: str = "",
|
|
hermes_log_id: str = "",
|
|
):
|
|
self.trace_dir = trace_dir or DEFAULT_TRACE_DIR
|
|
self.trace_dir.mkdir(parents=True, exist_ok=True)
|
|
|
|
self.trace_id = f"bl_{datetime.now(timezone.utc).strftime('%Y%m%d_%H%M%S')}_{uuid.uuid4().hex[:6]}"
|
|
self.harness_session_id = harness_session_id or str(uuid.uuid4())[:8]
|
|
self.hermes_session_id = hermes_session_id
|
|
self.hermes_log_id = hermes_log_id
|
|
|
|
self.trace_file = self.trace_dir / f"trace_{self.trace_id}.jsonl"
|
|
self.manifest_file = self.trace_dir / f"manifest_{self.trace_id}.json"
|
|
|
|
self.cycles: list[CycleTrace] = []
|
|
self.started_at: str = ""
|
|
self.finished_at: str = ""
|
|
|
|
def start_session(self) -> str:
|
|
"""Begin a trace session. Returns trace_id."""
|
|
self.started_at = datetime.now(timezone.utc).isoformat()
|
|
return self.trace_id
|
|
|
|
def begin_cycle(self, cycle_index: int) -> CycleTrace:
|
|
"""Start recording one ODA cycle."""
|
|
cycle = CycleTrace(
|
|
cycle_index=cycle_index,
|
|
timestamp_start=datetime.now(timezone.utc).isoformat(),
|
|
harness_session_id=self.harness_session_id,
|
|
hermes_session_id=self.hermes_session_id,
|
|
hermes_log_id=self.hermes_log_id,
|
|
)
|
|
return cycle
|
|
|
|
def finish_cycle(self, cycle: CycleTrace) -> None:
|
|
"""Finalize and persist one cycle to the trace file."""
|
|
cycle.timestamp_end = datetime.now(timezone.utc).isoformat()
|
|
# Compute duration
|
|
try:
|
|
t0 = datetime.fromisoformat(cycle.timestamp_start)
|
|
t1 = datetime.fromisoformat(cycle.timestamp_end)
|
|
cycle.duration_ms = int((t1 - t0).total_seconds() * 1000)
|
|
except (ValueError, TypeError):
|
|
cycle.duration_ms = 0
|
|
|
|
# Count successes/failures
|
|
cycle.actions_succeeded = sum(
|
|
1 for a in cycle.actions_executed if a.get("success", False)
|
|
)
|
|
cycle.actions_failed = sum(
|
|
1 for a in cycle.actions_executed if not a.get("success", True)
|
|
)
|
|
|
|
self.cycles.append(cycle)
|
|
|
|
# Append to JSONL
|
|
with open(self.trace_file, "a") as f:
|
|
f.write(json.dumps(cycle.to_dict()) + "\n")
|
|
|
|
def finish_session(self) -> SessionManifest:
|
|
"""Finalize the session and write the manifest."""
|
|
self.finished_at = datetime.now(timezone.utc).isoformat()
|
|
|
|
total_actions = sum(len(c.actions_executed) for c in self.cycles)
|
|
total_succeeded = sum(c.actions_succeeded for c in self.cycles)
|
|
total_failed = sum(c.actions_failed for c in self.cycles)
|
|
|
|
manifest = SessionManifest(
|
|
trace_id=self.trace_id,
|
|
harness_session_id=self.harness_session_id,
|
|
hermes_session_id=self.hermes_session_id,
|
|
hermes_log_id=self.hermes_log_id,
|
|
started_at=self.started_at,
|
|
finished_at=self.finished_at,
|
|
total_cycles=len(self.cycles),
|
|
total_actions=total_actions,
|
|
total_succeeded=total_succeeded,
|
|
total_failed=total_failed,
|
|
trace_file=str(self.trace_file),
|
|
trace_dir=str(self.trace_dir),
|
|
replay_command=(
|
|
f"python -m nexus.bannerlord_harness --mock --replay {self.trace_file}"
|
|
),
|
|
eval_note=(
|
|
"To replay: load this trace, re-execute each cycle's actions_planned "
|
|
"against a fresh harness in mock mode, compare actions_executed outcomes. "
|
|
"Success metric: >=90% action parity between original and replay runs."
|
|
),
|
|
)
|
|
|
|
with open(self.manifest_file, "w") as f:
|
|
json.dump(manifest.to_dict(), f, indent=2)
|
|
|
|
return manifest
|
|
|
|
@classmethod
|
|
def load_trace(cls, trace_file: Path) -> list[dict]:
|
|
"""Load a trace JSONL file for replay or analysis."""
|
|
cycles = []
|
|
with open(trace_file) as f:
|
|
for line in f:
|
|
line = line.strip()
|
|
if line:
|
|
cycles.append(json.loads(line))
|
|
return cycles
|
|
|
|
@classmethod
|
|
def load_manifest(cls, manifest_file: Path) -> dict:
|
|
"""Load a session manifest."""
|
|
with open(manifest_file) as f:
|
|
return json.load(f)
|
|
|
|
@classmethod
|
|
def list_traces(cls, trace_dir: Optional[Path] = None) -> list[dict]:
|
|
"""List all available trace sessions."""
|
|
d = trace_dir or DEFAULT_TRACE_DIR
|
|
if not d.exists():
|
|
return []
|
|
|
|
traces = []
|
|
for mf in sorted(d.glob("manifest_*.json")):
|
|
try:
|
|
manifest = cls.load_manifest(mf)
|
|
traces.append(manifest)
|
|
except (json.JSONDecodeError, IOError):
|
|
continue
|
|
return traces
|