Skip to content

python-api.md

Latest commit

 

History

History
211 lines (115 loc) · 7.61 KB

python-api.md

File metadata and controls

211 lines (115 loc) · 7.61 KB
Raw

Machine Python API

Generated from source signatures and docstrings without importing the runtime.

capabilities.py

Detect and report hardware capabilities at registration.

def default_machine_name(caps: dict | None=None)

Produce a useful hardware-first name without exposing an internal ID.

def get_capabilities(compute_mode: str='auto')

Return a capabilities dict for the registration message.

cli.py

def main()

Main.

config.py

class Settings

Settings.

  • def host_id(self) — Host id.
  • def manager_url(self) — Manager url.
  • def sharing_enabled(self) — Sharing enabled.
  • def sharing_supported(self) — Whether this installation includes the per-workload OS sandbox.
  • def saved_identity(self) — Saved identity.
  • def saved_credential(self) — Saved credential.
  • def work_dir(self) — Work dir.
  • def compute_mode(self) — Compute mode.
  • def cli_name(self) — Cli name.
  • def update_identity(self, **values: object) — Update identity.
  • def save_identity(self, host_id: str, credential: str, broker_url: str='') — Save identity.
  • def clear_identity(self) — Clear identity.

delta.py

Build sparse-or-dense fp16 model deltas and compress them with Zstandard.

def state_hash(state: dict[str, torch.Tensor])

SHA-256 over the fp32-canonicalized state_dict — verifies the base/result of a delta-based model sync (see executor.update_model). MUST stay byte-identical to the platform protocol. Both repositories lock this contract to the same golden digest in their own tests.

class DeltaAborted

Delta construction was cancelled by the active task.

def compute_delta_file(original: dict[str, torch.Tensor], modified: dict[str, torch.Tensor], destination: str, threshold: float=0.0, abort_event: threading.Event | None=None)

Write a compressed delta to disk for direct multipart upload.

executor.py

Isolated GPU training executor with disk-backed models and shard cache.

class StopExecution

BaseException prevents user except Exception blocks swallowing Stop.

class DownloadAborted

A deliberate, non-retryable transfer cancellation.

class Executor

Executor.

  • def work_dir(self) — Work dir.
  • def abort_event(self) — Task-scoped cancellation signal for transfers owned by the session.
  • def load(self, model_url: str, dataset_shards: list[dict | str], code: str, batch_size: int=32, params: dict | None=None, initial_shard_index: int=0) — Download the model and stage only the first bounded microshard.
  • def run(self, steps: int=100, round_num: int=0, total_rounds: int=1, mode: str='training', step_offset: int=0) — Start user code in a background thread.
  • def update_model(self, model_url: str, params: dict | None=None, dataset_shards: list[dict | str] | None=None, delta_url: str | None=None, base_hash: str | None=None, result_hash: str | None=None) — Prepare the model for the next round: apply middleware's fp16 sync delta to the model file already on disk when possible (see _try_delta_update), otherwise download the full model from model_url — which also serves as the automatic fallback whenever the delta path can't be used (hash mismatch, missing base, any error). A sync must never fail BECAUSE of the delta mechanism.
  • def get_original_state(self) — Returns the pre-training state_dict, reloading it from self._model_path on demand — it's deliberately not kept resident in RAM between load()/update_model() and this call (see their comments). mmap-backed: this reload sits on every round boundary's critical path right before the delta computation, and mmap turns a full multi-GB read+copy into lazy page-in as compute_delta walks the keys. Safe here because nothing rewrites _model_path while the delta is being computed (update_model comes later).
  • def get_shard(self, i: int) — Load one disk-backed shard; retain only its decoded value in memory.
  • def get_shard_path(self, i: int) — Return a task-local shard file for the isolated runner.
  • def wait(self) — Block until user code finishes (used for testing / sequential flow).
  • def stop(self) — Stop.
  • def reset(self) — Reset.
  • def is_running(self) — Is running.
  • def shutdown(self) — stop() + remove the temp directory (downloaded model files). Called when the WS session ends — every reconnect builds a fresh Executor with a fresh tmpdir, so without this each reconnect leaked a directory holding a potentially multi-GB model file until the disk filled. Idempotent.

gpu_device.py

Which physical GPU NVML calls should target on this process.

def nvml_device_index(device_count: int)

Physical GPU index for an NVML call, given how many devices NVML itself currently enumerates (pynvml.nvmlDeviceGetCount()).

main.py

ComputeField Machine WebSocket client.

async def main()

Main.

model_artifact.py

Safe preparation and automatic materialization of Hugging Face bundles.

def prepare_model_artifact(path: str, work_dir: str)

Return a canonical state file and optional local Hugging Face config root.

def materialize_hf_model(root: str, state: dict)

Build a built-in architecture and apply Transformers' key conversions.

monitor.py

Background hardware monitor — polls GPU (pynvml/MPS) and CPU/RAM (psutil) on a fixed interval.

def start_monitor(interval: int, callback: Callable[[dict], None])

Starts a daemon thread that calls callback(stats_dict) every interval seconds. Returns a threading.Event — set it to stop the monitor. stats_dict always contains cpu_util, ram_used_mb, and ram_total_mb. GPU fields added when available: NVIDIA (pynvml): gpu_util, gpu_mem_used_mb, gpu_mem_total_mb, gpu_temp_c Apple MPS: gpu_mem_used_mb (unified memory allocated by MPS)

pairing.py

Two-sided account pairing for ComputeField Machine.

def pair(api_url: str, code: str, *, name: str='', wait: bool=True)

Claim a browser-created code and wait for owner confirmation.

def finish_pairing()

Finish pairing.

sandbox_runtime.py

OS-backed workload sandboxing without a VM or desktop container product.

class SandboxUnavailable

The required host sandbox could not be constructed.

def backend_name()

Backend name.

def sandbox_command(command: list[str], task_dir: str)

Wrap command in the native host sandbox; never return it unwrapped.

def popen(command: list[str], task_dir: str)

Start a credential-free, network-free workload process.

def self_test(work_dir: str)

Prove that a new process can run but cannot read identity or use IP networking.

script_guard.py

Fail-closed syntax and capability gate for Workbench programs.

class UnsafeScript

Unsafescript.

def validate_script(code: str)

Parse and validate Machine code against the explicit capability set.

def execute_script(code: str, namespace: dict)

Execute validated code with a reduced builtin and import namespace.

speedtest.py

Estimate broker link throughput using one bounded echoed binary frame.

class SpeedTestError

Speedtesterror.

async def run_speed_test(ws, payload_mb: int, clock=time.monotonic)

Returns {"upload_mbps", "download_mbps", "payload_bytes"}. Raises SpeedTestError if the echoed payload doesn't match what was sent.

workload_runner.py

Credential-free child process used exclusively for workload execution.

def main()

Main.

workspace.py

Exclusive, crash-cleaned storage for tenant-bound Machine artifacts.

class WorkRoot

Own one Machine work root for the daemon's complete lifetime.

  • def close(self) — Close.