diff --git a/README.md b/README.md index b37beb8..39064ab 100644 --- a/README.md +++ b/README.md @@ -4,9 +4,11 @@ This is a light-weight, simplified ABM. ## Usage -This project uses uv as Python manager. To install uv, please visit https://docs.astral.sh/uv/getting-started/installation/ +This project uses uv as Python manager. To install uv, please visit +https://docs.astral.sh/uv/getting-started/installation/ -Once uv is installed on the machine, create a new Python environment for lighthouse and install dependencies. +Once uv is installed on the machine, create a new Python environment for lighthouse and install +dependencies. ```bash uv sync --locked @@ -18,9 +20,15 @@ To run the model with test data, use the following command: uv run activitysim run -c model/configs_mp -c model/configs -d model/data -o model/output ``` +## Monitoring + +See [Monitoring a model run](docs/sys-monitor.md) for CPU/memory sampling and progress tracking from +ActivitySim run plans and logs. + ## Contents - `model`: ActivitySim inputs (configs, data) for the lighthouse model. Currently `model/data` contains test-scale data from CTPS. - `notebooks`: Demo notebooks to test if the model still works. -- `src/lighthouse`: Python code used to implement this model. This may grow to include extensions to ActivitySim for things we want the lighthouse model to do. +- `src/lighthouse`: Python code used to implement this model. This may grow to include extensions to + ActivitySim for things we want the lighthouse model to do. diff --git a/docs/sys-monitor.md b/docs/sys-monitor.md new file mode 100644 index 0000000..f4f00f1 --- /dev/null +++ b/docs/sys-monitor.md @@ -0,0 +1,99 @@ +# Monitoring a model run + +`sys_monitor.py` samples CPU and memory usage and tracks ActivitySim progress using the resolved +`run_list.txt` and the main `activitysim.log`. It discovers model names, phase membership, and +expected worker counts from the run plan. Adding a component or splitting/renaming a phase requires +no monitor code changes. + +```bash +uv run python src/lighthouse/sys_monitor.py \ + --log model/output/log/activitysim.log \ + --csv model/output/log/sys_usage.csv +``` + +Use the actual log location for your logging configuration; some runs put logs directly in the +output directory. By default, the monitor looks for `run_list.txt` next to `--log`. Override this +with `--run-list PATH`. Use the main process log, not an individual worker's log. The CSV parent +directory must already exist. + +The monitor can start before the files exist or attach during a run. It replays existing log history +once, then reads appended lines, retaining incomplete lines until the writer finishes them. It +replays again when the plan changes and resets execution evidence if the log is replaced or +truncated. `--tail-bytes` remains accepted, but now controls the read chunk size, not the amount of +history retained. A small value no longer loses early completions or worker starts. Replaying a +large existing log can delay the first resource sample being displayed. + +Existing `--interval`, `--delay`, `--parent-pid`, and `--csv` options and CSV columns are preserved. +Worker counts are now available for every phase, including phases with one worker. `--parent-pid` +stops sampling after the parent exits, after reading available final log records; exit does not +itself mean the model succeeded. + +## Interpreting progress + +The displayed step is the earliest model step not yet satisfied by all workers expected by the plan. +It is not necessarily the step occupied by every worker: one worker can finish several steps before +another starts. Worker counts come from resolved `num_processes`, not just observed start messages. +Repeated completion messages do not count twice. Model order is a step count, not a time-based +percentage or ETA. + +The status values are: + +- `unknown`: the run plan or log is missing, unreadable, or unsupported. Resource sampling continues + with blank step/worker fields and a diagnostic; the monitor retries automatically. +- `waiting`: a plan is available but execution has not yet been observed. +- `resuming`: a resumed phase has workers whose restored progress is not yet known. The displayed + step is the earliest unproven step, not a claim that the worker is executing it. +- `running`: execution has been observed and some planned work remains. +- `finalizing`: all model steps are satisfied, but overall success has not been logged. Coalescing + or other finalization can still fail. +- `finished`: the main log contains ActivitySim's successful overall timing record. +- `failed`: the main log reports a subprocess failure or an unrecoverable run error. Later timing + records cannot turn that failure into success. + +## Resumes + +The adapter reads the saved breadcrumb snapshot in `run_list.txt`, not the live `breadcrumbs.yaml` +that ActivitySim updates while running. Phases already marked simulated are satisfied. Workers saved +as completed are credited only for a phase resuming with `_`; a named checkpoint can deliberately +rerun workers that finished previously. + +The monitor does not assume that a requested checkpoint is the checkpoint actually restored. For a +resumed worker, a completion of a later model proves that its earlier phase models were either run +or restored. A successful worker-exit message also satisfies that worker's phase. Until such +evidence arrives, partial resume progress may remain unresolved. Single-process tracking +additionally reads the runner's resolved `resume_after` message. + +## Format and compatibility limits + +The adapter targets ActivitySim's `print_run_list()` text format, checked against ActivitySim 1.5.1 +and a local development run. This format is informational, not ordinary YAML: it contains repeated +`step:` blocks and Python-style breadcrumb values. The parser deliberately rejects incomplete phase +coverage, invalid worker counts, unknown resume workers, and duplicate model names rather than +silently reporting misleading progress. Repeated invocations with identical model names are not +supported; distinct execution names including arguments are retained. + +Single-process tracking accepts a corresponding `multiprocess: False` run plan. ActivitySim entry +points do not all write a single-process `run_list.txt`; supply one exported from the actual +resolved run configuration with `--run-list` if it is available. The monitor does not reconstruct +settings or substitute a multiprocess plan. Without a plan it continues resource sampling with +unknown progress. + +Use matching run-plan and log files from the same invocation, preferably in a separate output +directory. The text artifacts have no common invocation identifier, so the monitor cannot reliably +detect every stale/mismatched pair or recover evidence removed before it attaches. Custom logging +formats that omit logger names or completion messages may leave progress unresolved. Use a separate +log per invocation rather than appending multiple invocations to one file. + +A supported replacement for this adapter has been requested in +[ActivitySim issue #1120](https://github.com/ActivitySim/activitysim/issues/1120): a versioned JSON +manifest of the resolved run plan. This implementation does not assume an unimplemented JSON schema. + +## Focused tests + +```bash +uv run pytest tests/test_sys_monitor.py +``` + +The small synthetic fixtures cover new components, renamed phases, staggered and one-worker +execution, late attachment, named and last-checkpoint resumes, single-process logs, failures, +partial writes, and log replacement. They do not require a model simulation or input data. diff --git a/src/lighthouse/sys_monitor.py b/src/lighthouse/sys_monitor.py index b1c8577..2431928 100644 --- a/src/lighthouse/sys_monitor.py +++ b/src/lighthouse/sys_monitor.py @@ -1,446 +1,528 @@ #!/usr/bin/env python3 -""" -Simple CPU and memory tracker for Windows (also works cross-platform). - -- Default interval: 0.5 seconds -- Prints timestamp, CPU %, Memory Used (GB), Memory Available (GB), Available Memory (%) -- Optional CSV logging with --csv FILE -- Optional program log tracking with --log FILE -- Sequence-aware step inference: only advances when steps complete in order -- Wait-for-all-workers for mp_households (requires all mp_households_N workers to complete a step) -- Phase-aware display (mp_initialize, mp_households, mp_summarize) -- Step status shown (running; for households shows done/total workers) - -Usage examples: - python sys_monitor.py - python sys_monitor.py --interval 1.0 - python sys_monitor.py --csv sys_usage.csv - python sys_monitor.py --log full/path/to/activitysim.log +"""Sample CPU/memory and track an ActivitySim run using its resolved run_list.txt. + +The run-list adapter is deliberately separate from execution tracking: ActivitySim's +informational text format is not YAML or a versioned public interchange format. """ -import sys -import os -import time import argparse +import ast +import csv +from dataclasses import dataclass from datetime import datetime +import os +from pathlib import Path import re -from typing import Optional, Dict, Set, Tuple +import sys +import time +from typing import Optional try: import psutil except ImportError: print( - "ERROR: 'psutil' is not installed.\n" - "Install it with:\n python -m pip install psutil", + "ERROR: 'psutil' is not installed. Install it with: python -m pip install psutil", file=sys.stderr, ) sys.exit(1) -# --- Ordered step list and phase mapping ------------------------------------- - -STEP_NAMES = [ - ### mp_initialize step - "initialize_landuse", - "initialize_households", - ### mp_accessibility step - "compute_accessibility", - ### mp_households step - "school_location", - "workplace_location", - "auto_ownership_simulate", - "free_parking", - "cdap_simulate", - "mandatory_tour_frequency", - "mandatory_tour_scheduling", - "non_mandatory_tour_frequency", - "non_mandatory_tour_destination", - "non_mandatory_tour_scheduling", - "tour_mode_choice_simulate", - "atwork_subtour_frequency", - "atwork_subtour_destination", - "atwork_subtour_scheduling", - "atwork_subtour_mode_choice", - "stop_frequency", - "trip_purpose", - "trip_destination", - "trip_purpose_and_destination", - "trip_scheduling", - "trip_mode_choice", - ### mp_summarize step - "write_data_dictionary", - "write_trip_matrices", - "write_tables", -] -STEP_TO_PHASE: Dict[str, str] = {} -for name in STEP_NAMES: - if name in { - "initialize_landuse", - "initialize_households", - }: - STEP_TO_PHASE[name] = "mp_initialize" - elif name in { - "compute_accessibility" - }: - STEP_TO_PHASE[name] = "mp_accessibility" - elif name in { - "write_data_dictionary", - "track_skim_usage", - "write_trip_matrices", - "write_tables", - }: - STEP_TO_PHASE[name] = "mp_summarize" - else: - STEP_TO_PHASE[name] = "mp_households" - - -# --- Regex patterns (strict to avoid false matches) -------------------------- - -# Completion lines come from activitysim.core.mp_tasks and include a runtime like ": 12.34 seconds" -# Examples: -# "INFO - activitysim.core.mp_tasks - mp_initialize initialize_landuse : 0.636 seconds" -# "INFO - activitysim.core.mp_tasks - mp_households_3 school_location : 290.146 seconds" -# "INFO - activitysim.core.mp_tasks - mp_summarize write_tables : 12.34 seconds" -PAT_COMPLETED_INIT_SUM = re.compile( - rf"activitysim\.core\.mp_tasks\s*-\s*mp_(?:initialize|accessibility|summarize)\s+" - rf"({'|'.join(map(re.escape, STEP_NAMES))})\s*:\s*([\d.]+)\s+seconds\b", - re.IGNORECASE, -) -PAT_COMPLETED_ACCESS = re.compile( - rf"activitysim\.core\.mp_tasks\s*-\s*mp_accessibility_(\d+)\s+" - rf"({'|'.join(map(re.escape, STEP_NAMES))})\s*:\s*([\d.]+)\s+seconds\b", - re.IGNORECASE, -) -PAT_COMPLETED_HH = re.compile( - rf"activitysim\.core\.mp_tasks\s*-\s*mp_households_(\d+)\s+" - rf"({'|'.join(map(re.escape, STEP_NAMES))})\s*:\s*([\d.]+)\s+seconds\b", - re.IGNORECASE, -) - -# Worker start lines (to infer the mp_households worker count) -# Example: "INFO - activitysim.core.mp_tasks - start process mp_households_7" -PAT_START_WORKER = re.compile( - r"activitysim\.core\.mp_tasks\s*-\s*start process mp_households_(\d+)\b", - re.IGNORECASE, -) -PAT_START_ACCESS_WORKER = re.compile( - r"activitysim\.core\.mp_tasks\s*-\s*start process mp_accessibility_(\d+)\b", - re.IGNORECASE, -) - -# Phase run lines (optional; for additional robustness/UX) -# Example: "INFO - activitysim.core.mp_tasks - run_sub_simulations step mp_households models ..." -PAT_RUN_PHASE = re.compile( - r"activitysim\.core\.mp_tasks\s*-\s*run_sub_simulations step (mp_initialize|mp_accessibility|mp_households|mp_summarize)\b", - re.IGNORECASE, -) - - -# --- Utilities --------------------------------------------------------------- - -def format_now() -> str: - return datetime.now().strftime("%Y-%m-%d %H:%M:%S") +@dataclass(frozen=True) +class Phase: + name: str + models: tuple[str, ...] + workers: tuple[str, ...] + resume_after: Optional[str] = None + completed_workers: tuple[str, ...] = () + simulated: bool = False -def bytes_to_gb(n_bytes: int) -> float: - return n_bytes / (1024**3) +@dataclass(frozen=True) +class RunPlan: + models: tuple[str, ...] + phases: tuple[Phase, ...] + multiprocess: bool + resume_after: Optional[str] + @classmethod + def from_text(cls, text: str) -> "RunPlan": + """Read the narrow print_run_list format, rejecting incomplete/ambiguous plans. -def _read_log_tail(log_path: str, max_read_bytes: int) -> str: - """Read the last max_read_bytes of the log (UTF-8 decode, ignore errors).""" - try: - size = os.path.getsize(log_path) - start = max(0, size - max_read_bytes) - with open(log_path, "rb") as f: # read-only, avoids writer contention - if start: - f.seek(start) - chunk = f.read() - return chunk.decode("utf-8", errors="ignore") if chunk else "" - except (FileNotFoundError, PermissionError, OSError): - return "" + In particular, repeated ``step:`` blocks must not be loaded as a YAML + mapping (which could silently discard all but the last phase). + Ignore unrelated fields, but validate every field used for progress. + """ + headers = {} + models = [] + phases = [] + breadcrumbs = {} + section = None + block = None + in_models = False + for line in text.splitlines(): + value = line.strip() + if not value: + continue + if not line.startswith(" "): + in_models = False + if value in ("models:", "multiprocess_steps:", "breadcrumbs:"): + section = value[:-1] + block = None + elif ":" in value: + key, val = value.split(":", 1) + headers[key] = val.strip() + section = None + continue + if section == "models": + if not value.startswith("- "): + raise ValueError("Invalid models list in run_list.txt") + models.append(value[2:].strip()) + elif section in ("multiprocess_steps", "breadcrumbs"): + if line.startswith(" step: "): + name = value.split(":", 1)[1].strip() + block = {"name": name, "models": []} + in_models = False + if section == "multiprocess_steps": + phases.append(block) + else: + if name in breadcrumbs: + raise ValueError("Duplicate breadcrumb phase") + breadcrumbs[name] = block + elif block is not None: + if section == "multiprocess_steps": + if value == "models:": + in_models = True + elif in_models and value.startswith("- "): + block["models"].append(value[2:].strip()) + else: + in_models = False + key, sep, val = value.partition(":") + if sep and key in ("num_processes", "resume_after", "name"): + if key == "name" and val.strip() != block["name"]: + raise ValueError("Inconsistent phase name") + block[key] = val.strip() + else: + # print_run_list prints breadcrumb fields WITHOUT colons. + key, _, val = value.partition(" ") + if key in ("simulate", "completed"): + try: + block[key] = ast.literal_eval(val.strip()) + except (ValueError, SyntaxError) as exc: + raise ValueError("Invalid resume breadcrumbs") from exc + if ( + headers.get("multiprocess") not in ("True", "False") + or "resume_after" not in headers + ): + raise ValueError( + "Missing multiprocess/resume_after headers in run_list.txt" + ) + if not models or any(not m for m in models) or len(set(models)) != len(models): + raise ValueError("Expected a nonempty, unique model sequence") + resume = headers["resume_after"] + resume = None if resume == "None" else resume + if resume not in (None, "_") and resume not in models: + raise ValueError("Resume checkpoint is not in the model sequence") + multiprocess = headers["multiprocess"] == "True" + if not multiprocess: + if phases or breadcrumbs: + raise ValueError("Unexpected phases in a single-process plan") + return cls( + tuple(models), + (Phase("single_process", tuple(models), ("MainProcess",), resume),), + False, + resume, + ) + if not phases or [m for p in phases for m in p["models"]] != models: + raise ValueError("Phase model lists do not match the full model sequence") + names = [p["name"] for p in phases] + if len(set(names)) != len(names) or any(not n for n in names): + raise ValueError("Phase names must be nonempty and unique") + if list(breadcrumbs) != names[: len(breadcrumbs)]: + raise ValueError("Resume breadcrumbs do not match the phase sequence") + result = [] + for p in phases: + try: + count = int(p.get("num_processes", "")) + except ValueError as exc: + raise ValueError("Missing/invalid resolved num_processes") from exc + if count < 1 or not p["models"]: + raise ValueError("Each phase needs models and a positive worker count") + workers = ( + (p["name"],) + if count == 1 + else tuple(f"{p['name']}_{i}" for i in range(count)) + ) + checkpoint = p.get("resume_after") + checkpoint = None if checkpoint == "None" else checkpoint + if checkpoint not in (None, "_") and checkpoint not in p["models"]: + raise ValueError("Phase resume checkpoint is not in that phase") + crumb = breadcrumbs.get(p["name"], {}) if resume else {} + completed = crumb.get("completed", []) + if not isinstance(completed, list) or any( + not isinstance(w, str) or w not in workers for w in completed + ): + raise ValueError("Invalid completed worker list") + simulated = crumb.get("simulate") + if simulated is not None and not isinstance(simulated, bool): + raise ValueError("Invalid simulate breadcrumb") + # A named checkpoint reruns workers even if they previously completed. + result.append( + Phase( + p["name"], + tuple(p["models"]), + workers, + checkpoint, + tuple(completed) if checkpoint == "_" else (), + simulated is True, + ) + ) + all_workers = [w for p in result for w in p.workers] + if len(set(all_workers)) != len(all_workers): + raise ValueError("Ambiguous worker names across phases") + return cls(tuple(models), tuple(result), True, resume) -# --- State and inference from log ------------------------------------------- +# Match logger and message structure, never a fixed vocabulary of model/phase names. +LOG_MESSAGE = re.compile( + r"\b(activitysim\.core\.[\w.]+|activitysim\.cli\.run)\s*-\s*(.*)" +) +DURATION = r"[\d.]+\s+seconds\b" +COMPLETED = re.compile(rf"(\S+)\s+(\S+)\s*:\s*{DURATION}") +PROCESS_DONE = re.compile( + r"process (\S+) completed(?: with exitcode 0)?$", re.IGNORECASE +) +PROCESS_FAILED = re.compile(r"process (\S+) failed with exitcode", re.IGNORECASE) +SINGLE_DONE = re.compile(rf"time to execute run\.(\S+)\s*:\s*{DURATION}", re.IGNORECASE) +RUN_DONE = re.compile( + rf"Time to execute all models(?: completed)?\s*:\s*{DURATION}", re.IGNORECASE +) class StepTracker: - """ - Tracks step completion strictly by sequence. - For mp_households, a step completes only after all known workers complete it. - """ - - def __init__(self) -> None: - # Index of last completed step in STEP_NAMES (-1 means nothing completed yet) - self.last_completed_idx: int = -1 - # Set of worker ids (ints) seen via "start process mp_households_N" - self.active_workers: Set[int] = set() - # Set of worker ids (ints) seen via "start process mp_accessibility_N" - self.active_access_workers: Set[int] = set() - # For accessibility, which workers have completed the step - self.access_completed_workers: Dict[str, Set[int]] = {} - # For the current households step, which workers have completed - self.hh_completed_workers: Dict[str, Set[int]] = {} - # Keep a cache of seen completion tokens to avoid re-processing duplicates - # token: tuple('init/sum' or 'hh', phase_id or worker_id, step_name) - self.seen_completions: Set[Tuple[str, str, str]] = set() - # Optional: track last observed/announced phase - self.last_observed_phase: Optional[str] = None - - def _phase_of_index(self, idx: int) -> Optional[str]: - if 0 <= idx < len(STEP_NAMES): - return STEP_TO_PHASE[STEP_NAMES[idx]] - return None - - def update_from_tail(self, text: str) -> None: - """Parse the tail text and update internal state (workers and completions).""" - if not text: - return + """Report the earliest step not satisfied by every planned worker. - # 1) Record any new workers that started (used to know how many must complete a household step) - for m in PAT_START_WORKER.finditer(text): - wid = int(m.group(1)) - self.active_workers.add(wid) - - for m in PAT_START_ACCESS_WORKER.finditer(text): - wid = int(m.group(1)) - self.active_access_workers.add(wid) + In resumed phases a worker completing a later step proves that its earlier + steps were run or restored. A request to resume alone does not prove that. + """ - # 2) Record phase runs (optional, just for UX) - for m in PAT_RUN_PHASE.finditer(text): - self.last_observed_phase = m.group(1) + def __init__(self, plan: RunPlan): + self.plan = plan + self.phase_by_worker = {w: p for p in plan.phases for w in p.workers} + self.phase_by_model = {m: p for p in plan.phases for m in p.models} + self.reset() + + def reset(self) -> None: + self.completed = {m: set() for m in self.plan.models} + self.resume_pending = { + w + for phase in self.plan.phases + if phase.resume_after and not phase.simulated + for w in phase.workers + if w not in phase.completed_workers + } + self.observed = False + self.failed = False + self.finished = False + for phase in self.plan.phases: + for model in phase.models: + self.completed[model].update( + phase.workers if phase.simulated else phase.completed_workers + ) - # 3) Record completions (init/summarize) - for m in PAT_COMPLETED_INIT_SUM.finditer(text): - step_name = m.group(1) - token = ("init/sum", "single", step_name) - if token in self.seen_completions: - continue - self.seen_completions.add(token) - # We only mark completion during resolution below (respecting order) - - # 3b) Record completions (accessibility per-worker) - for m in PAT_COMPLETED_ACCESS.finditer(text): - wid = m.group(1) - step_name = m.group(2) - token = ("access", wid, step_name) - if token in self.seen_completions: - continue - self.seen_completions.add(token) - self.access_completed_workers.setdefault(step_name, set()).add(int(wid)) - - # 4) Record completions (households per-worker) - for m in PAT_COMPLETED_HH.finditer(text): - wid = m.group(1) - step_name = m.group(2) - token = ("hh", wid, step_name) - if token in self.seen_completions: - continue - self.seen_completions.add(token) - # Cache worker completion for this step - self.hh_completed_workers.setdefault(step_name, set()).add(int(wid)) - - # 5) Resolve sequence strictly, possibly advancing multiple steps if evidence exists - advanced = True - while advanced: - advanced = False - next_idx = self.last_completed_idx + 1 - if next_idx >= len(STEP_NAMES): - break - - next_step = STEP_NAMES[next_idx] - phase = STEP_TO_PHASE[next_step] - - if phase in ("mp_initialize", "mp_summarize"): - # Completion is true if we have a completion token for this step - token = ("init/sum", "single", next_step) - if token in self.seen_completions: - self.last_completed_idx = next_idx - advanced = True - # Clean any stale household worker sets for previous steps - self.hh_completed_workers.pop(next_step, None) - continue - # not completed yet -> stop resolving - break - - if phase == "mp_accessibility": - completed_workers = self.access_completed_workers.get(next_step, set()) - if self.active_access_workers and completed_workers >= self.active_access_workers: - self.last_completed_idx = next_idx - advanced = True - self.access_completed_workers.pop(next_step, None) - continue - break - - # mp_households: require all active workers to complete the step - if phase == "mp_households": - completed_workers = self.hh_completed_workers.get(next_step, set()) - # If we don't yet know active workers, we can't mark complete. - # We'll wait until starts are seen; still show running status with ?/? below. - if self.active_workers and completed_workers >= self.active_workers: - # step completed by all workers - self.last_completed_idx = next_idx - advanced = True - # prepare for next step: keep hh_completed_workers but this step is done - # (we can drop its set to save memory) - self.hh_completed_workers.pop(next_step, None) - continue - # not completed yet -> stop resolving - break + def complete(self, worker: str, model: str) -> None: + phase = self.phase_by_worker.get(worker) + if phase is None or model not in phase.models: + return + self.observed = True + self.resume_pending.discard(worker) + models = ( + phase.models[: phase.models.index(model) + 1] + if phase.resume_after + else (model,) + ) + for name in models: + self.completed[name].add(worker) + + def update(self, line: str) -> None: + match = LOG_MESSAGE.search(line) + if not match: + return + logger, msg = match.groups() + # Overall success is separate from model completion (coalescing may fail). + if logger in ( + "activitysim.core.tracing", + "activitysim.core.workflow.runner", + "activitysim.cli.run", + ): + if RUN_DONE.match(msg): + self.finished = not self.failed + self.observed = True + if ( + "activitysim run encountered an unrecoverable error" in msg + or "all models until this error" in msg + ): + self.failed = True + self.finished = False + if logger == "activitysim.core.mp_tasks" and self.plan.multiprocess: + match = COMPLETED.fullmatch(msg.split(" (", 1)[0]) + if match: + self.complete(*match.groups()) + match = PROCESS_DONE.fullmatch(msg) + if match and match[1] in self.phase_by_worker: + phase = self.phase_by_worker[match[1]] + for model in phase.models: + self.complete(match[1], model) + if PROCESS_FAILED.match(msg): + self.failed = True + self.finished = False + if msg.startswith(("start process ", "run_sub_simulations step ")): + self.observed = True + elif not self.plan.multiprocess and logger in ( + "activitysim.core.pipeline", + "activitysim.core.workflow.runner", + ): + match = SINGLE_DONE.match(msg) + if match: + self.complete("MainProcess", match[1]) + if msg.startswith("resume_after "): + checkpoint = msg.removeprefix("resume_after ") + if checkpoint in self.plan.models: + self.observed = True + self.resume_pending.discard("MainProcess") + for model in self.plan.models[ + : self.plan.models.index(checkpoint) + 1 + ]: + self.completed[model].add("MainProcess") + if "#run_model running step " in msg: + self.observed = True def current_step_info( self, - ) -> Tuple[Optional[int], Optional[str], str, Optional[int], Optional[int]]: - """ - Returns: - - current_step_idx (None if all done), - - current_step_name (None if all done), - - phase string ("mp_initialize"|"mp_households"|"mp_summarize"| "DONE"), - - done_workers (for household phase, else None), - - total_workers (for household phase, else None) - """ - idx = self.last_completed_idx + 1 - if idx >= len(STEP_NAMES): + ) -> tuple[Optional[int], Optional[str], str, Optional[int], Optional[int]]: + """Return zero-based index, model, phase, satisfied workers, expected workers.""" + if self.finished: return None, None, "DONE", None, None + for idx, model in enumerate(self.plan.models): + phase = self.phase_by_model[model] + done = len(self.completed[model]) + if done < len(phase.workers): + return idx, model, phase.name, done, len(phase.workers) + return None, None, "FINALIZING", None, None + + @property + def status(self) -> str: + if self.failed: + return "failed" + if self.finished: + return "finished" + idx, model, _, _, _ = self.current_step_info() + if idx is None: + return "finalizing" + phase = self.phase_by_model[model] + if self.resume_pending.intersection(phase.workers): + return "resuming" + return "running" if self.observed else "waiting" + + +class LogReader: + """Replay once, then read appended complete lines in bounded chunks. + + Retain partial lines across ticks. Detect replacement/truncation, including + truncate-and-regrow when the previously read boundary has changed. + """ - step_name = STEP_NAMES[idx] - phase = STEP_TO_PHASE[step_name] + def __init__(self, path: Path, chunk_bytes: int = 256 * 1024): + if chunk_bytes <= 0: + raise ValueError("Log read chunk size must be positive") + self.path = Path(path) + self.chunk_bytes = chunk_bytes + self.reset() + + def reset(self): + self.identity = None + self.offset = 0 + self.pending = b"" + self.boundary = b"" + + def lines(self, on_reset): + with self.path.open("rb") as stream: + stat = os.fstat(stream.fileno()) + identity = (stat.st_dev, stat.st_ino) + stream.seek(max(0, self.offset - len(self.boundary))) + boundary = stream.read(len(self.boundary)) + if self.identity is not None and ( + identity != self.identity + or stat.st_size < self.offset + or boundary != self.boundary + ): + self.reset() + on_reset() + self.identity = identity + stream.seek(self.offset) + # Only consume the snapshot available at the start of this poll. + remaining = stat.st_size - self.offset + while remaining > 0: + chunk = stream.read(min(self.chunk_bytes, remaining)) + if not chunk: + break + self.offset += len(chunk) + remaining -= len(chunk) + self.boundary = (self.boundary + chunk)[-128:] + parts = (self.pending + chunk).split(b"\n") + self.pending = parts.pop() + for line in parts: + yield line.decode("utf-8", errors="replace").rstrip("\r") + + +class RunMonitor: + """Reload changed plans and replay logs so late plan creation loses no events.""" + + def __init__(self, log_path, run_list_path=None, chunk_bytes=256 * 1024): + self.reader = LogReader(Path(log_path), chunk_bytes) + self.plan_path = ( + Path(run_list_path) + if run_list_path + else Path(log_path).with_name("run_list.txt") + ) + self.plan_stamp = None + self.tracker = None + self.warning = None + + def poll(self) -> None: + try: + stat = self.plan_path.stat() + stamp = (stat.st_dev, stat.st_ino, stat.st_mtime_ns, stat.st_size) + if stamp != self.plan_stamp: + plan = RunPlan.from_text(self.plan_path.read_text(encoding="utf-8")) + self.tracker = StepTracker(plan) + self.plan_stamp = stamp + self.reader.reset() + for line in self.reader.lines(self.tracker.reset): + self.tracker.update(line) + self.warning = None + except (OSError, ValueError) as exc: + self.warning = f"Progress unavailable: {exc}. Waiting for readable, valid run_list.txt and log files." + # Never retain DONE or guessed progress while an artifact is invalid. + self.tracker = None + self.plan_stamp = None - if phase == "mp_households": - done = len(self.hh_completed_workers.get(step_name, set())) - total = len(self.active_workers) if self.active_workers else None - return idx, step_name, phase, done, total - return idx, step_name, phase, None, None +def format_now() -> str: + return datetime.now().strftime("%Y-%m-%d %H:%M:%S") -# --- Monitor loop ------------------------------------------------------------ +def bytes_to_gb(n_bytes: int) -> float: + return n_bytes / (1024**3) def monitor( - interval: float = 0.5, - csv_path: Optional[str] = None, - log_path: Optional[str] = None, - tail_bytes: int = 256 * 1024, # bigger default to improve catch-up robustness - parent_pid: Optional[int] = None, - delay_seconds: float = 0, -) -> None: - # Delay start if requested + interval=0.5, + csv_path=None, + log_path=None, + tail_bytes=256 * 1024, + parent_pid=None, + delay_seconds=0, + run_list_path=None, +): + """Sample resources; --tail-bytes now controls read chunks, not retained history.""" if delay_seconds > 0: print(f"Delaying sys_monitor start for {delay_seconds} seconds...") time.sleep(delay_seconds) - - # Warm up psutil's CPU measurement for better first reading psutil.cpu_percent(interval=None) - - csv_file = None - tracker = StepTracker() - parent_process = None - log_path_warned_missing = False - - # If parent PID provided, get parent process object + progress = RunMonitor(log_path, run_list_path, tail_bytes) if log_path else None + parent = None if parent_pid is not None: try: - parent_process = psutil.Process(parent_pid) - print(f"Monitoring parent process PID {parent_pid} ({parent_process.name()})") - except (psutil.NoSuchProcess, psutil.AccessDenied) as e: - print(f"Warning: Cannot access parent process {parent_pid}: {e}") - print("Continuing without parent process monitoring.") - parent_process = None - + parent = psutil.Process(parent_pid) + print(f"Monitoring parent process PID {parent_pid} ({parent.name()})") + except (psutil.NoSuchProcess, psutil.AccessDenied) as exc: + print(f"Warning: Cannot access parent process {parent_pid}: {exc}") + csv_file = None + previous_warning = None try: if csv_path: - # Create file and header if new, else append - file_exists = False - try: - with open(csv_path, "r", encoding="utf-8"): - file_exists = True - except FileNotFoundError: - file_exists = False - + has_content = Path(csv_path).exists() and Path(csv_path).stat().st_size > 0 csv_file = open(csv_path, "a", encoding="utf-8", newline="") - if not file_exists: - csv_file.write( - "timestamp,cpu_percent,memory_used_gb,memory_available_gb,available_memory_pct,phase,step_index,step_name,status,workers_done,workers_total\n" + writer = csv.writer(csv_file) + if not has_content: + writer.writerow( + [ + "timestamp", + "cpu_percent", + "memory_used_gb", + "memory_available_gb", + "available_memory_pct", + "phase", + "step_index", + "step_name", + "status", + "workers_done", + "workers_total", + ] ) - csv_file.flush() - print("Press Ctrl+C to stop.") - total_steps = len(STEP_NAMES) - while True: - # Check if parent process is still alive - if parent_process is not None: - try: - if not parent_process.is_running(): - print(f"\nParent process {parent_pid} has exited. Stopping monitor.") - break - except (psutil.NoSuchProcess, psutil.AccessDenied): - print(f"\nParent process {parent_pid} no longer accessible. Stopping monitor.") - break ts = format_now() - cpu = psutil.cpu_percent(interval=None) # non-blocking snapshot - + cpu = psutil.cpu_percent(interval=None) vm = psutil.virtual_memory() - total_gb = bytes_to_gb(vm.total) + used_gb = bytes_to_gb(vm.total - vm.available) avail_gb = bytes_to_gb(vm.available) - used_gb = total_gb - avail_gb - avail_pct = (vm.available / vm.total) * 100 if vm.total else 0.0 - - # Update tracker from log (if provided) - if log_path: - if not os.path.exists(log_path) and not log_path_warned_missing: - print( - f"Warning: Log file not found at '{log_path}'. " - "Phase/step tracking will stay at the initial step until a valid log path is provided." - ) - log_path_warned_missing = True - text = _read_log_tail(log_path, max_read_bytes=tail_bytes) - tracker.update_from_tail(text) - - curr_idx, curr_step, phase, done_workers, total_workers = ( + avail_pct = vm.available / vm.total * 100 if vm.total else 0.0 + tracker = None + if progress: + progress.poll() + if progress.warning and progress.warning != previous_warning: + print(progress.warning) + previous_warning = progress.warning + tracker = progress.tracker + idx, step, phase, done, total = ( tracker.current_step_info() + if tracker + else (None, None, "UNKNOWN", None, None) ) - - # Build status string - if phase == "DONE": - status = "finished" - display_phase = "DONE" - step_progress = f"{total_steps}/{total_steps}" - step_display = "DONE" - workers_display = "" - else: - status = "running" - display_phase = phase - step_progress = f"{(curr_idx or 0) + 1}/{total_steps}" - if phase == "mp_households": - if total_workers is None: - workers_display = " (workers: ?/?)" - else: - workers_display = f" (workers: {done_workers}/{total_workers})" - else: - workers_display = "" - step_display = f"{curr_step}{workers_display} ({status})" - - line = ( - f"{ts} | CPU: {cpu:5.1f}% | " - f"Used: {used_gb:6.2f} GB | " + status = tracker.status if tracker else "unknown" + count = len(tracker.plan.models) if tracker else None + step_progress = ( + f"{idx + 1 if idx is not None else count}/{count}" if tracker else "?/?" + ) + workers = f" (workers: {done}/{total})" if total is not None else "" + print( + f"{ts} | CPU: {cpu:5.1f}% | Used: {used_gb:6.2f} GB | " f"Available: {avail_gb:6.2f} GB ({avail_pct:5.1f}%) | " - f"Phase: {display_phase} | Step: {step_progress} | {step_display}" + f"Phase: {phase} | Step: {step_progress} | " + f"{step or phase}{workers} ({status})" ) - print(line) - if csv_file: - csv_file.write( - f"{ts},{cpu:.1f},{used_gb:.2f},{avail_gb:.2f},{avail_pct:.1f}," - f"{display_phase},{'' if curr_idx is None else curr_idx+1},{'' if curr_step is None else curr_step}," - f"{status if phase != 'DONE' else 'finished'}," - f"{'' if done_workers is None else done_workers}," - f"{'' if total_workers is None else total_workers}\n" + writer.writerow( + [ + ts, + f"{cpu:.1f}", + f"{used_gb:.2f}", + f"{avail_gb:.2f}", + f"{avail_pct:.1f}", + phase, + idx + 1 if idx is not None else "", + step or "", + status, + done if done is not None else "", + total if total is not None else "", + ] ) csv_file.flush() - + # Consume final log records before stopping; exit never implies success. + if parent is not None: + try: + if not parent.is_running(): + print( + f"Parent process {parent_pid} has exited. Stopping monitor ({status})." + ) + break + except (psutil.NoSuchProcess, psutil.AccessDenied): + print( + f"Parent process {parent_pid} is unavailable. Stopping monitor ({status})." + ) + break time.sleep(interval) - except KeyboardInterrupt: print("\nStopped by user.") finally: @@ -448,22 +530,29 @@ def monitor( csv_file.close() -# --- CLI --------------------------------------------------------------------- +def positive_float(value): + try: + number = float(value) + except ValueError as exc: + raise argparse.ArgumentTypeError("Value must be a number") from exc + if not 0 < number < float("inf"): + raise argparse.ArgumentTypeError("Value must be finite and > 0") + return number -def positive_float(value: str) -> float: +def positive_int(value): try: - v = float(value) - except ValueError: - raise argparse.ArgumentTypeError("Interval must be a number.") - if v <= 0: - raise argparse.ArgumentTypeError("Interval must be > 0.") - return v + number = int(value) + except ValueError as exc: + raise argparse.ArgumentTypeError("Value must be an integer") from exc + if number <= 0: + raise argparse.ArgumentTypeError("Value must be > 0") + return number -def main() -> None: +def main(): parser = argparse.ArgumentParser( - description="Track CPU/memory usage and current ActivitySim step (phase-aware, waits for all workers)." + description="Track CPU/memory and ActivitySim progress from the resolved run plan and log." ) parser.add_argument( "--interval", @@ -473,37 +562,30 @@ def main() -> None: help="Sampling interval in seconds (default: 0.5)", ) parser.add_argument( - "--csv", - type=str, - default="../output/log/sys_usage.csv", - help="Optional path to CSV file for logging (e.g., sys_usage.csv)", + "--csv", default="../output/log/sys_usage.csv", help="CSV output path" ) parser.add_argument( "--log", - type=str, default="../output/log/activitysim.log", - help="Optional path to an ActivitySim log file to infer the current step.", + help="ActivitySim main log path", ) parser.add_argument( - "--tail-bytes", - type=int, - default=256 * 1024, - help="Number of bytes to tail from the log each tick (default: 262144). Increase if starting mid-run.", + "--run-list", help="Resolved run_list.txt path (default: alongside --log)" ) parser.add_argument( - "--parent-pid", - type=int, - default=None, - help="Parent process PID to monitor. Monitor will exit if parent process dies.", + "--tail-bytes", + type=positive_int, + default=256 * 1024, + help="Read chunk size in bytes (default: 262144); all existing log history is replayed", ) + parser.add_argument("--parent-pid", type=int, help="Stop when this process exits") parser.add_argument( "--delay", type=positive_float, default=0, - help="Delay in seconds before starting monitoring (default: 0)", + help="Delay before monitoring, in seconds", ) args = parser.parse_args() - monitor( interval=args.interval, csv_path=args.csv, @@ -511,6 +593,7 @@ def main() -> None: tail_bytes=args.tail_bytes, parent_pid=args.parent_pid, delay_seconds=args.delay, + run_list_path=args.run_list, ) diff --git a/tests/fixtures/sys_monitor/resume_latest.txt b/tests/fixtures/sys_monitor/resume_latest.txt new file mode 100644 index 0000000..a21ce7e --- /dev/null +++ b/tests/fixtures/sys_monitor/resume_latest.txt @@ -0,0 +1,38 @@ +resume_after: _ +multiprocess: True +models: + - initialize_landuse + - school_location + - constraint_new_access + - write_tables + +multiprocess_steps: + step: prepare + name: prepare + num_processes: 1 + models: + - initialize_landuse + step: choices_renamed + name: choices_renamed + num_processes: 2 + models: + - school_location + - constraint_new_access + resume_after: _ + step: export + name: export + num_processes: 1 + models: + - write_tables + +breadcrumbs: + step: prepare + name prepare + apportion True + simulate True + completed ['prepare'] + coalesce True + step: choices_renamed + name choices_renamed + apportion True + completed ['choices_renamed_0'] diff --git a/tests/fixtures/sys_monitor/resume_named.txt b/tests/fixtures/sys_monitor/resume_named.txt new file mode 100644 index 0000000..9f4c8e7 --- /dev/null +++ b/tests/fixtures/sys_monitor/resume_named.txt @@ -0,0 +1,37 @@ +resume_after: school_location +multiprocess: True +models: + - initialize_landuse + - school_location + - constraint_new_access + - write_tables + +multiprocess_steps: + step: prepare + name: prepare + num_processes: 1 + models: + - initialize_landuse + step: choices_renamed + name: choices_renamed + num_processes: 2 + models: + - school_location + - constraint_new_access + resume_after: school_location + step: export + name: export + num_processes: 1 + models: + - write_tables + +breadcrumbs: + step: prepare + name prepare + simulate True + completed ['prepare'] + step: choices_renamed + name choices_renamed + simulate None + completed ['choices_renamed_0', 'choices_renamed_1'] + coalesce None diff --git a/tests/fixtures/sys_monitor/resumed.log.txt b/tests/fixtures/sys_monitor/resumed.log.txt new file mode 100644 index 0000000..59606e7 --- /dev/null +++ b/tests/fixtures/sys_monitor/resumed.log.txt @@ -0,0 +1,6 @@ +01/01/2026 12:00:00 - INFO - activitysim.core.mp_tasks - run_sub_simulations step choices_renamed models resume_after _ +01/01/2026 12:00:00 - INFO - activitysim.core.mp_tasks - start process choices_renamed_1 +01/01/2026 12:00:01 - INFO - activitysim.core.mp_tasks - choices_renamed_1 constraint_new_access : 0.02 seconds (0.0 minutes) +01/01/2026 12:00:02 - INFO - activitysim.core.mp_tasks - process choices_renamed_1 completed +01/01/2026 12:00:03 - INFO - activitysim.core.mp_tasks - export write_tables : 0.04 seconds (0.0 minutes) +01/01/2026 12:00:04 - INFO - activitysim.core.tracing - Time to execute all models : 4.00 seconds (0.1 minutes) diff --git a/tests/fixtures/sys_monitor/run_list.txt b/tests/fixtures/sys_monitor/run_list.txt new file mode 100644 index 0000000..2841706 --- /dev/null +++ b/tests/fixtures/sys_monitor/run_list.txt @@ -0,0 +1,31 @@ +resume_after: None +multiprocess: True +models: + - initialize_landuse + - school_location + - constraint_new_access + - write_tables + +multiprocess_steps: + step: prepare + name: prepare + begin: initialize_landuse + num_processes: 1 + slice: None + chunk_size: 0 + models: + - initialize_landuse + step: choices_renamed + name: choices_renamed + begin: school_location + num_processes: 2 + slice: {'tables': ['households', 'persons'], 'exclude': None} + models: + - school_location + - constraint_new_access + step: export + name: export + begin: write_tables + num_processes: 1 + models: + - write_tables diff --git a/tests/fixtures/sys_monitor/single_process.log.txt b/tests/fixtures/sys_monitor/single_process.log.txt new file mode 100644 index 0000000..f89a30a --- /dev/null +++ b/tests/fixtures/sys_monitor/single_process.log.txt @@ -0,0 +1,5 @@ +01/01/2026 12:00:00 - INFO - activitysim.core.workflow.runner - resume_after initialize_landuse +01/01/2026 12:00:01 - INFO - activitysim.core.workflow.runner - #run_model running step constraint_new_access +01/01/2026 12:00:02 - Level 25 - activitysim.core.workflow.runner - time to execute run.constraint_new_access : 0.02 seconds (0.0 minutes) +01/01/2026 12:00:03 - Level 25 - activitysim.core.workflow.runner - time to execute run.write_tables : 0.03 seconds (0.0 minutes) +01/01/2026 12:00:04 - INFO - activitysim.core.tracing - Time to execute all models completed : 4.00 seconds (0.1 minutes) diff --git a/tests/fixtures/sys_monitor/single_process.txt b/tests/fixtures/sys_monitor/single_process.txt new file mode 100644 index 0000000..767c121 --- /dev/null +++ b/tests/fixtures/sys_monitor/single_process.txt @@ -0,0 +1,6 @@ +resume_after: _ +multiprocess: False +models: + - initialize_landuse + - constraint_new_access + - write_tables diff --git a/tests/fixtures/sys_monitor/staggered.log.txt b/tests/fixtures/sys_monitor/staggered.log.txt new file mode 100644 index 0000000..49e90e8 --- /dev/null +++ b/tests/fixtures/sys_monitor/staggered.log.txt @@ -0,0 +1,13 @@ +01/01/2026 12:00:00 - INFO - activitysim.core.mp_tasks - run_sub_simulations step prepare models resume_after None +01/01/2026 12:00:00 - INFO - activitysim.core.mp_tasks - start process prepare +01/01/2026 12:00:01 - INFO - activitysim.core.mp_tasks - prepare initialize_landuse : 0.636 seconds +01/01/2026 12:00:02 - INFO - activitysim.core.mp_tasks - run_sub_simulations step choices_renamed models resume_after None +01/01/2026 12:00:02 - INFO - activitysim.core.mp_tasks - start process choices_renamed_0 +01/01/2026 12:00:03 - INFO - activitysim.core.mp_tasks - choices_renamed_0 school_location : 0.001 seconds (0.0 minutes) +01/01/2026 12:00:03 - INFO - activitysim.core.mp_tasks - choices_renamed_0 constraint_new_access : 0.001 seconds (0.0 minutes) +01/01/2026 12:00:04 - INFO - activitysim.core.mp_tasks - start process choices_renamed_1 +01/01/2026 12:00:06 - INFO - activitysim.core.mp_tasks - choices_renamed_1 school_location : 2.00 seconds (0.0 minutes) +01/01/2026 12:00:07 - INFO - activitysim.core.mp_tasks - choices_renamed_1 constraint_new_access : 1.00 seconds +01/01/2026 12:00:08 - INFO - activitysim.core.mp_tasks - run_sub_simulations step export models resume_after None +01/01/2026 12:00:09 - INFO - activitysim.core.mp_tasks - export write_tables : 1.00 seconds +01/01/2026 12:00:10 - INFO - activitysim.core.tracing - Time to execute all models : 10.00 seconds (0.2 minutes) diff --git a/tests/test_sys_monitor.py b/tests/test_sys_monitor.py new file mode 100644 index 0000000..3b4fcd3 --- /dev/null +++ b/tests/test_sys_monitor.py @@ -0,0 +1,408 @@ +"""Small synthetic fixtures in ActivitySim's print_run_list/log formats.""" + +import csv +import importlib.util +from pathlib import Path +import sys +from types import SimpleNamespace + +import pytest + +# Import the standalone script without loading unrelated Lighthouse dependencies. +SCRIPT = Path(__file__).resolve().parents[1] / "src/lighthouse/sys_monitor.py" +spec = importlib.util.spec_from_file_location("sys_monitor", SCRIPT) +monitor = importlib.util.module_from_spec(spec) +sys.modules[spec.name] = monitor +spec.loader.exec_module(monitor) +FIXTURES = Path(__file__).parent / "fixtures/sys_monitor" + + +def fixture(name): + return (FIXTURES / name).read_text() + + +def tracker(name="run_list.txt"): + return monitor.StepTracker(monitor.RunPlan.from_text(fixture(name))) + + +def log(message, logger="mp_tasks"): + return f"INFO - activitysim.core.{logger} - {message}" + + +def test_plan_discovers_custom_step_phases_and_worker_counts(): + plan = tracker().plan + assert plan.models == ( + "initialize_landuse", + "school_location", + "constraint_new_access", + "write_tables", + ) + assert [p.name for p in plan.phases] == ["prepare", "choices_renamed", "export"] + assert plan.phases[0].workers == ("prepare",) + assert plan.phases[1].workers == ("choices_renamed_0", "choices_renamed_1") + + +def test_staggered_workers_do_not_advance_before_unstarted_worker(): + state = tracker() + lines = fixture("staggered.log.txt").splitlines() + for line in lines[:7]: + state.update(line) + assert state.current_step_info() == (1, "school_location", "choices_renamed", 1, 2) + # Repeated messages are idempotent, and the second worker is still required. + state.update(lines[5]) + assert state.current_step_info()[-2:] == (1, 2) + state.update(lines[7]) + state.update(lines[8]) + assert state.current_step_info() == ( + 2, + "constraint_new_access", + "choices_renamed", + 1, + 2, + ) + state.update(lines[9]) + assert state.current_step_info()[1] == "write_tables" + for line in lines[10:12]: + state.update(line) + assert state.current_step_info()[2] == "FINALIZING" + assert state.status == "finalizing" + state.update(lines[12]) + assert state.current_step_info()[2] == "DONE" + assert state.status == "finished" + + +def test_unknown_steps_workers_and_unrelated_timings_are_ignored(): + state = tracker() + for message in ( + "outsider initialize_landuse : 1.0 seconds", + "prepare nonexistent : 1.0 seconds", + "prepare initialize_landuse UNTIL ERROR : 1.0 seconds", + ): + state.update(log(message)) + state.update(log("prepare initialize_landuse : 1.0 seconds", "tracing")) + assert state.current_step_info() == (0, "initialize_landuse", "prepare", 0, 1) + + +def test_renamed_one_worker_phase_and_phase_scoping(): + state = tracker() + state.update(log("choices_renamed_0 initialize_landuse : 1.0 seconds")) + assert state.current_step_info()[0] == 0 + state.update(log("prepare initialize_landuse : 1.0 seconds")) + assert state.current_step_info()[0] == 1 + + +def test_latest_resume_seeds_only_saved_success_and_waits_for_remaining_worker(): + state = tracker("resume_latest.txt") + assert state.current_step_info() == (1, "school_location", "choices_renamed", 1, 2) + for line in fixture("resumed.log.txt").splitlines()[:3]: + state.update(line) + assert state.current_step_info() == (3, "write_tables", "export", 0, 1) + + +def test_named_resume_does_not_reuse_old_completed_worker_list(): + state = tracker("resume_named.txt") + assert state.current_step_info() == (1, "school_location", "choices_renamed", 0, 2) + assert state.status == "resuming" + state.update(log("choices_renamed_0 constraint_new_access : 0.02 seconds")) + assert state.current_step_info()[-2:] == (1, 2) + state.update(log("choices_renamed_1 constraint_new_access : 0.02 seconds")) + assert state.current_step_info()[1] == "write_tables" + + +def test_resumed_worker_can_restore_entire_phase_without_model_completion_lines(): + state = tracker("resume_latest.txt") + state.update(log("process choices_renamed_1 completed")) + assert state.current_step_info()[1] == "write_tables" + + +def test_resume_request_without_breadcrumbs_does_not_skip_work(): + text = fixture("run_list.txt").replace("resume_after: None", "resume_after: _") + state = monitor.StepTracker(monitor.RunPlan.from_text(text)) + assert state.current_step_info()[0] == 0 + + +@pytest.mark.parametrize("logger", ["workflow.runner", "pipeline"]) +def test_single_process_resume_and_completion(logger): + state = tracker("single_process.txt") + lines = ( + fixture("single_process.log.txt") + .replace("workflow.runner", logger) + .splitlines() + ) + state.update(lines[0]) + assert state.current_step_info() == ( + 1, + "constraint_new_access", + "single_process", + 0, + 1, + ) + for line in lines[1:-1]: + state.update(line) + assert state.status == "finalizing" + state.update(lines[-1]) + assert state.status == "finished" + + +def test_failure_is_not_overridden_by_late_completion_or_exit(): + state = tracker() + state.update(log("process choices_renamed_1 failed with exitcode -9")) + state.update(log("Time to execute all models : 10.0 seconds", "tracing")) + assert state.status == "failed" + assert state.current_step_info()[2] != "DONE" + + +def test_coalesce_failure_after_model_steps_complete(): + state = tracker() + for line in fixture("staggered.log.txt").splitlines()[:-1]: + state.update(line) + state.update( + log("Time to execute all models until this error : 10.0 seconds", "tracing") + ) + assert state.status == "failed" + + +@pytest.mark.parametrize( + "old,new", + [ + ("multiprocess: True", "multiprocess: maybe"), + ("num_processes: 2", "num_processes: 0"), + ("num_processes: 2", "num_processes: many"), + (" - constraint_new_access\n", ""), + (" - school_location\n", " - initialize_landuse\n"), + ("name: export", "name: inconsistent"), + ], +) +def test_malformed_or_partial_plan_is_rejected(old, new): + with pytest.raises(ValueError): + monitor.RunPlan.from_text(fixture("run_list.txt").replace(old, new)) + + +def test_unknown_resume_worker_is_rejected(): + with pytest.raises(ValueError, match="worker"): + monitor.RunPlan.from_text( + fixture("resume_latest.txt").replace( + "['choices_renamed_0']", "['unknown_0']" + ) + ) + + +def test_reordered_breadcrumbs_are_rejected(): + with pytest.raises(ValueError, match="breadcrumbs"): + monitor.RunPlan.from_text( + fixture("resume_latest.txt").replace( + "breadcrumbs:\n step: prepare", "breadcrumbs:\n step: export" + ) + ) + + +def test_late_attachment_replays_more_than_tail_window(tmp_path): + (tmp_path / "run_list.txt").write_text(fixture("run_list.txt")) + path = tmp_path / "activitysim.log" + lines = fixture("staggered.log.txt").splitlines(keepends=True) + # Important early completions are far outside the old tail window. + path.write_text( + "".join(lines[:10]) + "irrelevant detail\n" * 20000 + "".join(lines[10:]) + ) + progress = monitor.RunMonitor(path, chunk_bytes=17) + progress.poll() + assert progress.warning is None + assert progress.tracker.status == "finished" + offset = progress.reader.offset + progress.poll() + assert progress.reader.offset == offset + assert progress.tracker.status == "finished" + + +def test_partial_line_is_preserved_until_complete(tmp_path): + path = tmp_path / "activitysim.log" + path.write_bytes(b"first\npar") + reader = monitor.LogReader(path, chunk_bytes=2) + assert list(reader.lines(lambda: pytest.fail("unexpected reset"))) == ["first"] + with path.open("ab") as stream: + stream.write("tial café\n".encode()) + assert list(reader.lines(lambda: pytest.fail("unexpected reset"))) == [ + "partial café" + ] + + +@pytest.mark.parametrize("replace", [False, True]) +def test_log_truncation_or_replacement_discards_completion(tmp_path, replace): + (tmp_path / "run_list.txt").write_text(fixture("run_list.txt")) + path = tmp_path / "activitysim.log" + path.write_text(fixture("staggered.log.txt")) + progress = monitor.RunMonitor(path) + progress.poll() + assert progress.tracker.status == "finished" + if replace: + path.rename(tmp_path / "old.log") + # Regrow beyond the old size, exercising the boundary check too. + path.write_text("new run starting\n" * 1000) + progress.poll() + assert progress.tracker.current_step_info()[0] == 0 + assert progress.tracker.status == "waiting" + + +def test_missing_then_partial_then_valid_plan_replays_existing_log(tmp_path): + path = tmp_path / "activitysim.log" + path.write_text(fixture("staggered.log.txt")) + progress = monitor.RunMonitor(path) + progress.poll() + assert progress.tracker is None + assert "Progress unavailable" in progress.warning + plan = tmp_path / "run_list.txt" + plan.write_text("resume_after: None\nmultiprocess: True\nmodels:\n") + progress.poll() + assert progress.tracker is None + plan.write_text(fixture("run_list.txt")) + progress.poll() + assert progress.tracker.status == "finished" + assert progress.warning is None + + +def test_explicit_plan_path_and_changed_plan(tmp_path): + plan = tmp_path / "custom_plan.txt" + plan.write_text(fixture("run_list.txt")) + path = tmp_path / "activitysim.log" + path.write_text("") + progress = monitor.RunMonitor(path, plan) + progress.poll() + assert progress.tracker.current_step_info()[0] == 0 + plan.write_text(fixture("resume_latest.txt")) + progress.poll() + assert progress.tracker.current_step_info()[0] == 1 + + +def test_missing_log_never_claims_running_or_done(tmp_path): + (tmp_path / "run_list.txt").write_text(fixture("run_list.txt")) + progress = monitor.RunMonitor(tmp_path / "missing.log") + progress.poll() + assert progress.tracker is None + assert progress.warning + + +def test_cli_csv_and_parent_exit_do_not_claim_success(tmp_path, monkeypatch, capsys): + path = tmp_path / "activitysim.log" + path.write_text(fixture("staggered.log.txt").splitlines()[0] + "\n") + (tmp_path / "run_list.txt").write_text(fixture("run_list.txt")) + csv_path = tmp_path / "usage.csv" + monkeypatch.setattr(monitor.psutil, "cpu_percent", lambda **kwargs: 12.5) + monkeypatch.setattr( + monitor.psutil, + "virtual_memory", + lambda: SimpleNamespace(total=2048, available=1024), + ) + monkeypatch.setattr( + monitor.psutil, + "Process", + lambda pid: SimpleNamespace(name=lambda: "model", is_running=lambda: False), + ) + monkeypatch.setattr( + sys, + "argv", + [ + str(SCRIPT), + "--log", + str(path), + "--csv", + str(csv_path), + "--parent-pid", + "123", + "--tail-bytes", + "8", + ], + ) + monitor.main() + with csv_path.open() as stream: + rows = list(csv.DictReader(stream)) + assert rows[0]["step_name"] == "initialize_landuse" + assert rows[0]["status"] == "running" + assert rows[0]["workers_total"] == "1" + assert "Stopping monitor (running)" in capsys.readouterr().out + + +@pytest.mark.parametrize("value", ["0", "-1", "nan", "inf"]) +def test_invalid_interval(value): + with pytest.raises(monitor.argparse.ArgumentTypeError): + monitor.positive_float(value) + + +def test_fresh_single_process_does_not_skip_missing_completions(): + plan = monitor.RunPlan.from_text( + fixture("single_process.txt").replace("resume_after: _", "resume_after: None") + ) + state = monitor.StepTracker(plan) + state.update( + log( + "time to execute run.constraint_new_access : 1.0 seconds", "workflow.runner" + ) + ) + assert state.current_step_info()[1] == "initialize_landuse" + state.update( + log("time to execute run.initialize_landuse : 1.0 seconds", "workflow.runner") + ) + assert state.current_step_info()[1] == "write_tables" + + +def test_named_resume_can_rerun_checkpoint_instead_of_restoring_it(): + state = tracker("resume_named.txt") + state.update(log("choices_renamed_0 school_location : 0.02 seconds")) + assert state.current_step_info()[1] == "school_location" + assert state.current_step_info()[-2:] == (1, 2) + assert state.status == "resuming" + state.update(log("choices_renamed_1 school_location : 0.02 seconds")) + assert state.current_step_info()[1] == "constraint_new_access" + assert state.status == "running" + + +def test_appended_log_advances_without_replaying_existing_history(tmp_path): + (tmp_path / "run_list.txt").write_text(fixture("run_list.txt")) + path = tmp_path / "activitysim.log" + lines = fixture("staggered.log.txt").splitlines(keepends=True) + path.write_text("".join(lines[:7])) + progress = monitor.RunMonitor(path, chunk_bytes=31) + progress.poll() + assert progress.tracker.current_step_info()[-2:] == (1, 2) + state = progress.tracker + with path.open("a") as stream: + stream.writelines(lines[7:]) + progress.poll() + assert progress.tracker is state + assert state.status == "finished" + + +def test_resource_csv_continues_with_missing_plan_and_appends_one_header( + tmp_path, monkeypatch +): + monkeypatch.setattr(monitor.psutil, "cpu_percent", lambda **kwargs: 25.0) + monkeypatch.setattr( + monitor.psutil, + "virtual_memory", + lambda: SimpleNamespace(total=2048, available=1024), + ) + monkeypatch.setattr( + monitor.psutil, + "Process", + lambda pid: SimpleNamespace(name=lambda: "model", is_running=lambda: False), + ) + path = tmp_path / "usage.csv" + for _ in range(2): + monitor.monitor( + csv_path=path, log_path=tmp_path / "missing.log", parent_pid=123 + ) + with path.open() as stream: + rows = list(csv.DictReader(stream)) + assert len(rows) == 2 + assert all( + row["status"] == "unknown" + and row["step_name"] == "" + and row["workers_total"] == "" + and row["cpu_percent"] == "25.0" + for row in rows + ) + + +@pytest.mark.parametrize("value", ["0", "-1", "abc", "1.5"]) +def test_invalid_log_chunk_size(value): + with pytest.raises(monitor.argparse.ArgumentTypeError): + monitor.positive_int(value)