跳到导航
Python API

dynamo.mocker

Mock engine used to run Dynamo workflows without a real GPU backend.
以 Markdown 格式查看

dynamo.mocker publishes 2 classes and 18 functions. Source: components/src/dynamo/mocker/__init__.py

Result of processing —planner-profile-data argument. Cleans up tmpdir on deletion.

from dynamo.mocker.args import ProfileDataResult
ProfileDataResult(npz_path: Path | None, tmpdir: tempfile.TemporaryDirectory | None)

components/src/dynamo/mocker/args.py#L27

Public methods

init

__init__(npz_path: Path | None, tmpdir: tempfile.TemporaryDirectory | None)

No summary available.

source

No summary available.

from dynamo.mocker import ReasoningConfig
ReasoningConfig(start_thinking_token_id: int, end_thinking_token_id: int, thinking_ratio: float) -> None

lib/bindings/python/src/dynamo/_core.pyi#L1957

Public methods

init

__init__(start_thinking_token_id: int, end_thinking_token_id: int, thinking_ratio: float) -> None

No summary available.

source

No summary available.

from dynamo.mocker.config import apply_worker_engine_args_overrides
apply_worker_engine_args_overrides(engine_args: Mapping[str, Any], *, kv_bytes_per_token: int | None = None, bootstrap_port: int | None = None, zmq_kv_events_port: int | None = None, zmq_replay_port: int | None = None, ais_mtp_seed: int | None = None) -> dict

components/src/dynamo/mocker/config.py#L191

No summary available.

from dynamo.mocker.config import build_mocker_engine_args
build_mocker_engine_args(args: argparse.Namespace) -> dict

components/src/dynamo/mocker/config.py#L59

No summary available.

from dynamo.mocker.config import build_runtime_config
build_runtime_config(engine_args: Mapping[str, Any]) -> tuple[int, ModelRuntimeConfig]

components/src/dynamo/mocker/config.py#L217

Compute the stagger delay based on worker count to give the frontend time to process registrations. Returns the delay in seconds between worker launches.

from dynamo.mocker.main import compute_stagger_delay
compute_stagger_delay(num_workers: int, stagger_delay: float) -> float

components/src/dynamo/mocker/main.py#L124

No summary available.

from dynamo.mocker.config import engine_limit
engine_limit(config: Mapping[str, Any], name: str) -> int | None

components/src/dynamo/mocker/config.py#L32

Shutdown dynamo distributed runtime instances. The endpoints will be immediately invalidated so no new requests will be accepted.

from dynamo.mocker.main import graceful_shutdown
graceful_shutdown(runtimes: list)

components/src/dynamo/mocker/main.py#L35

Launch mocker worker(s) with isolated DistributedRuntime instances.

from dynamo.mocker.main import launch_workers
launch_workers(args: argparse.Namespace, base_engine_args)

Each worker gets its own DistributedRuntime, which means:

  • Separate etcd/NATS connections
  • Separate Component instances (no shared overhead)
  • Independent service registration and stats scraping
  • But still sharing the same tokio runtime (efficient)

components/src/dynamo/mocker/main.py#L146

No summary available.

from dynamo.mocker.config import load_mocker_engine_args
load_mocker_engine_args(args: argparse.Namespace) -> dict

components/src/dynamo/mocker/config.py#L185

No summary available.

from dynamo.mocker.main import main
main()

components/src/dynamo/mocker/main.py#L312

No summary available.

from dynamo.mocker.config import normalize_mocker_config
normalize_mocker_config(config: Mapping[str, Any] | str | None = None) -> dict

components/src/dynamo/mocker/config.py#L19

Parse command-line arguments for the Dynamo mocker engine.

from dynamo.mocker.args import parse_args
parse_args(argv: list[str] | None = None) -> argparse.Namespace

Returns

  • argparse.Namespace — argparse.Namespace: Parsed command-line arguments.

components/src/dynamo/mocker/args.py#L151

Parse comma-separated bootstrap ports string into list of integers.

from dynamo.mocker.args import parse_bootstrap_ports
parse_bootstrap_ports(ports_str: str | None) -> list[int]

components/src/dynamo/mocker/args.py#L144

No summary available.

from dynamo.mocker.config import performance_config
performance_config(config: Mapping[str, Any]) -> dict | None

components/src/dynamo/mocker/config.py#L25

Resolve model_path to a local directory, fetching config/tokenizer if needed.

from dynamo.mocker.main import prefetch_model
prefetch_model(model_path: str) -> str

fetch_model returns the cached snapshot directory without contacting the hub when config.json and the tokenizer files are already present, and downloads only those files otherwise. Resolving once here means neither the workers nor the KV-bytes estimate below resolve a hub ID over the network. Returns the original path on failure so callers degrade to their previous behavior.

components/src/dynamo/mocker/main.py#L46

Resolve —planner-profile-data to an NPZ file path.

from dynamo.mocker.args import resolve_planner_profile_data
resolve_planner_profile_data(planner_profile_data: Path | None) -> ProfileDataResult

Handles backward compatibility by accepting either:

  1. A mocker-format NPZ file (returned as-is)
  2. A profiler-style results directory (converted to mocker-format NPZ)

Parameters

planner_profile_data
Path | None

Path from —planner-profile-data argument.

Returns

  • ProfileDataResult — ProfileDataResult with npz_path and optional tmpdir for cleanup.

Raises

  • FileNotFoundError — If path doesn’t contain valid profile data in any supported format.

components/src/dynamo/mocker/args.py#L45

No summary available.

from dynamo.mocker import run_mocker_trace_replay
run_mocker_trace_replay(trace_files, extra_engine_args = None, router_config = None, num_workers = 1, replay_concurrency = None, router_mode = 'round_robin', arrival_speedup_ratio = 1.0, trace_block_size = None, trace_format = 'mooncake', trace_shared_prefix_ratio = 0.0, trace_num_prefix_groups = 0, model_name = None, sla_ttft_ms = None, sla_itl_ms = None, sla_e2e_ms = None, capture_per_request = False, agentic_lanes = None, execution_model = None)

components/src/dynamo/mocker/__init__.py#L36

Resolve disaggregation mode from —disaggregation-mode or legacy boolean flags. Raises ValueError if validation fails.

from dynamo.mocker.args import validate_worker_type_args
validate_worker_type_args(args: argparse.Namespace) -> None

components/src/dynamo/mocker/args.py#L98

Main worker function that launches mocker instances.

from dynamo.mocker.main import worker
worker()

Each mocker gets its own DistributedRuntime instance for true isolation, while still sharing the same event loop and tokio runtime.

components/src/dynamo/mocker/main.py#L78