Skip to content

clip.monitor_marsclip_queue

monitor_marsclip_queue

Small terminal monitor for queued MarsCLIP training runs.

load_queue_status 

load_queue_status(queue_dir: Path | str) -> dict[str, Any]

Load a queue status JSON payload.

Source code in src/clip/monitor_marsclip_queue.py 
13
14
15
16
17
def load_queue_status(queue_dir: pathlib.Path | str) -> dict[str, Any]:
    """Load a queue status JSON payload."""
    queue_path = pathlib.Path(queue_dir)
    status_path = queue_path / "queue_status.json"
    return json.loads(status_path.read_text())

extract_out_dir 

extract_out_dir(argv: list[str]) -> pathlib.Path | None

Extract a trainer output directory from a queued argv list.

Source code in src/clip/monitor_marsclip_queue.py 
20
21
22
23
24
25
def extract_out_dir(argv: list[str]) -> pathlib.Path | None:
    """Extract a trainer output directory from a queued argv list."""
    for idx, token in enumerate(argv):
        if token == "--out-dir" and idx + 1 < len(argv):
            return pathlib.Path(argv[idx + 1])
    return None

build_job_snapshot 

build_job_snapshot(job: dict[str, Any]) -> dict[str, Any]

Enrich one queued job record with filesystem/runtime state.

Source code in src/clip/monitor_marsclip_queue.py 
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
def build_job_snapshot(job: dict[str, Any]) -> dict[str, Any]:
    """Enrich one queued job record with filesystem/runtime state."""
    out_dir = extract_out_dir(list(job.get("argv", [])))
    log_path = pathlib.Path(job["log_path"])
    progress = _load_progress(out_dir) if out_dir is not None and out_dir.exists() else None
    return {
        "name": str(job["name"]),
        "status": str(job["status"]),
        "pid": job.get("pid"),
        "pid_alive": _is_pid_alive(job.get("pid")),
        "started_at": job.get("started_at"),
        "finished_at": job.get("finished_at"),
        "heartbeat_at": job.get("heartbeat_at"),
        "exit_code": job.get("exit_code"),
        "out_dir": str(out_dir) if out_dir is not None else None,
        "progress": progress,
        "latest_checkpoint_step": _latest_checkpoint_step(out_dir) if out_dir is not None and out_dir.exists() else None,
        "has_summary": bool(out_dir is not None and (out_dir / "summary.json").exists()),
        "has_preview_start": bool(out_dir is not None and (out_dir / "preview_step_000000.png").exists()),
        "last_log_line": _tail_last_line(log_path),
    }

render_queue_snapshot 

render_queue_snapshot(status: dict[str, Any]) -> str

Render a human-readable terminal snapshot for the queue.

Source code in src/clip/monitor_marsclip_queue.py 
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
def render_queue_snapshot(status: dict[str, Any]) -> str:
    """Render a human-readable terminal snapshot for the queue."""
    lines = [
        f"Queue: {status['queue_name']}",
        f"Status: {status['status']}",
        f"Queue Updated: {status.get('updated_at')}",
        "",
    ]
    for job in status.get("jobs", []):
        snapshot = build_job_snapshot(job)
        lines.append(f"[{snapshot['status']}] {snapshot['name']}")
        lines.append(
            f"  pid={snapshot['pid']} alive={snapshot['pid_alive']} exit={snapshot['exit_code']} heartbeat={snapshot['heartbeat_at']}"
        )
        lines.append(f"  started={snapshot['started_at']} finished={snapshot['finished_at']}")
        lines.append(f"  out_dir={snapshot['out_dir']}")
        progress = snapshot["progress"]
        if progress is not None:
            current_step = progress.get("current_step")
            target_step = progress.get("target_step")
            latest_loss = progress.get("latest_loss")
            best_loss = progress.get("best_loss")
            latest_lr = progress.get("latest_lr")
            lines.append(
                "  progress="
                f"{progress.get('phase')} step={current_step}/{target_step} "
                f"loss={latest_loss} best={best_loss} lr={latest_lr}"
            )
        else:
            lines.append(
                f"  artifacts=summary:{snapshot['has_summary']} preview0:{snapshot['has_preview_start']} checkpoint_step:{snapshot['latest_checkpoint_step']}"
            )
        if snapshot["last_log_line"] is not None:
            lines.append(f"  last_log={snapshot['last_log_line']}")
        lines.append("")
    return "\n".join(lines).rstrip() + "\n"

monitor_queue 

monitor_queue(queue_dir: Path | str, *, refresh_seconds: float = 5.0, once: bool = False) -> None

Print a periodically refreshed queue snapshot.

Source code in src/clip/monitor_marsclip_queue.py 
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
def monitor_queue(
    queue_dir: pathlib.Path | str,
    *,
    refresh_seconds: float = 5.0,
    once: bool = False,
) -> None:
    """Print a periodically refreshed queue snapshot."""
    queue_path = pathlib.Path(queue_dir)
    while True:
        status = load_queue_status(queue_path)
        rendered = render_queue_snapshot(status)
        if once:
            print(rendered, end="")
            return
        print("\033[2J\033[H", end="")  # pragma: no cover
        print(rendered, end="", flush=True)  # pragma: no cover
        time.sleep(refresh_seconds)  # pragma: no cover