Service

The FastAPI service exposes the preprocessing, generation, training, evaluation, and virtualisation workflows as HTTP endpoints.

Long-running operations run as background jobs. Clients submit a request and poll the job endpoints for its status and result.

The running service provides complete request and response documentation at /docs. Request models that add semantics beyond their corresponding CLI are documented here. Request bodies that mirror CLI arguments are documented by the CLI entry points.

_api

FastAPI service for the CrystaLLM-pi command-line utilities.

Endpoint groups are registered from the _utils._api modules. Long-running generation and training work runs as background jobs and is polled through the jobs endpoints.

Usage

uvicorn _api:app --host 0.0.0.0 --port 8000

Endpoints

root async

root()

Return the API root response.

Source code in _api.py

@app.get("/")
async def root() -> dict:
    """Return the API root response."""
    return {
        "name": "CrystaLLM-pi API",
        "version": "1.0.0",
        "documentation": "/docs",
        "endpoints": {
            "health": ["/healthz"],
            "preprocessing": [
                "/preprocessing/deduplicate",
                "/preprocessing/calc-theor-xrd",
                "/preprocessing/process-exp-xrd",
                "/preprocessing/clean",
                "/preprocessing/save-dataset",
                "/preprocessing/cifs-zip-to-parquet",
            ],
            "training": ["/train"],
            "generation": [
                "/generate/direct",
                "/generate/make-prompts",
                "/generate/cifs",
                "/generate/evaluate-cifs",
                "/generate/postprocess",
            ],
            "metrics": [
                "/metrics/vun",
                "/metrics/ehull",
                "/metrics/xrd",
                "/metrics/property",
            ],
            "virtualiser": [
                "/virtualise",
            ],
            "jobs": [
                "/jobs",
                "/jobs/{job_id}",
                "/jobs/{job_id}/cancel",
            ],
        },
    }

healthz async

healthz()

Return the API health status.

Source code in _api.py

@app.get("/healthz")
async def healthz() -> dict:
    """Return the API health status."""
    return {
        "status": "ok",
        "service": "CrystaLLM-pi API",
        "time": datetime.now().isoformat(),
    }

Job status

JobStatus

Bases: BaseModel

Status of a background job.

Generation and training endpoints return a job status while the work runs in the background. Callers poll the jobs endpoints for completion.

Source code in _api.py

class JobStatus(BaseModel):
    """Status of a background job.

    Generation and training endpoints return a job status while the work runs in the background.
    Callers poll the jobs endpoints for completion.
    """
    job_id: str
    status: Literal['pending', 'running', 'completed', 'failed']
    command: str
    started_at: str | None = None
    completed_at: str | None = None
    error: str | None = None
    output_file: str | None = None
    log_file: str | None = None

Generation requests

DirectGenerationRequest

Bases: BaseModel

Request model for direct Hugging Face model generation.

Source code in _utils/_api/generation.py

class DirectGenerationRequest(BaseModel):
    """Request model for direct Hugging Face model generation."""

    hf_model_path: str = Field(..., description="HuggingFace model path (e.g., c-bone/CrystaLLM-pi_bandgap)")
    model_registry: str | None = Field(None, description="Custom model registry JSON file")
    output_parquet: str | None = Field(None, description="Output parquet file path")
    output_cif_dir: str | None = Field(None, description="Output directory for generated CIF files")
    input_parquet: str | None = Field(None, description="Input prompts parquet (alternative to manual)")
    reduced_formula_list: str | None = Field(None, description="Comma-separated reduced formulas")
    search_zs: bool = Field(False, description="Search through Z=1,2,3,4,6 to find valid structures")
    z_list: str | None = Field(None, description="Comma-separated explicit Z integers mapping 1:1 to formulas")
    condition_lists: list[str] | None = Field(None, description="Condition vectors")
    xrd_files: list[str] | None = Field(None, description="Files with XRD peaks (.csv, .xy, .txt, .dat) for XRD models")
    xrd_wavelength: float | None = Field(None, description="Wavelength in Angstrom (default: CuKa1 1.54056, assumed with a warning)")
    level: Literal['level_1', 'level_2', 'level_3', 'level_4'] = Field("level_2", description="Prompt detail level")
    spacegroups: str | None = Field(None, description="Comma-separated spacegroups (level_4 only)")
    temperature: float = Field(1.0, description="Sampling temperature")
    num_return_sequences: int = Field(1, description="Sequences per sample")
    max_return_attempts: int = Field(1, description="Generation attempts per sample")
    max_samples: int | None = Field(None, description="Max samples to process")
    scoring_mode: str | None = Field(None, description="Scoring mode for filtering (case-insensitive: LOGP or None).")
    target_valid_cifs: int = Field(1, description="Target valid CIFs per prompt (LOGP requires sensibility + formula-structure consistency)")
    multi_gpu: Literal['auto', 'true', 'false'] = Field("auto", description="Deprecated: 'false' translates to --num_workers_gpu 1, other values are ignored. Use nproc_per_node instead")
    nproc_per_node: int | None = Field(None, description="Max GPU workers (passed as --num_workers_gpu, unset uses all visible GPUs)")
    num_workers: int = Field(4, description="Post-processing workers")
    skip_postprocess: bool = Field(False, description="Skip CIF validation/postprocessing")

MakePromptsRequest

Bases: BaseModel

Request body for POST /generate/make-prompts.

Manual mode constructs prompts from compositions and condition_lists according to mode. Automatic mode extracts prompts from a dataset. Exactly one mode is used per request.

Source code in _utils/_api/generation.py

class MakePromptsRequest(BaseModel):
    """Request body for `POST /generate/make-prompts`.

    Manual mode constructs prompts from `compositions` and `condition_lists` according to `mode`.
    Automatic mode extracts prompts from a dataset. Exactly one mode is used per request.
    """
    output_parquet: str = Field(..., description="Output parquet file")
    manual: bool | None = Field(None, description="Manual mode")
    compositions: str | None = Field(None, description="Comma-separated compositions")
    condition_lists: list[str] | None = Field(None, description="Condition value lists")
    level: Literal['level_1', 'level_2', 'level_3', 'level_4'] = Field("level_2", description="Prompt level")
    spacegroups: str | None = Field(None, description="Space groups (for level_4)")
    automatic: bool = Field(False, description="Automatic mode (extract from dataset)")
    HF_dataset: str | None = Field(None, description="HuggingFace dataset name")
    split: str | None = Field(None, description="Dataset split")
    cif_column: str = Field("CIF", description="Column containing CIF strings for automatic mode")
    condition_columns: str | list[str] | None = Field(
        None,
        description="Condition column name(s), either a single string or a list",
    )
    input_parquet: str | None = Field(None, description="Input parquet (alternative to HF dataset)")
    remove_ref_columns: bool = Field(False, description="Keep only prompt and condition columns in automatic output")
    mode: Literal['cartesian', 'paired', 'broadcast'] = Field("cartesian", description="Composition-condition pairing mode for manual prompts")

GenerateCIFsRequest

Bases: BaseModel

Request body for POST /generate/cifs.

The endpoint uses the supplied configuration to run a CIF generation job.

Source code in _utils/_api/generation.py

class GenerateCIFsRequest(BaseModel):
    """Request body for `POST /generate/cifs`.

    The endpoint uses the supplied configuration to run a CIF generation job.
    """
    config_file: str = Field(..., description="Path to generation config JSONC file")

Training requests

TrainRequest

Bases: BaseModel

Request body for POST /train.

When multi_gpu is unset, the API uses torchrun automatically when at least two GPUs are visible. nproc_per_node defaults to the number of visible GPUs.

Source code in _utils/_api/training.py

class TrainRequest(BaseModel):
    """Request body for `POST /train`.

    When `multi_gpu` is unset, the API uses `torchrun` automatically when at least two GPUs are
    visible. `nproc_per_node` defaults to the number of visible GPUs.
    """
    config_file: str = Field(..., description="Path to training config JSONC file")
    multi_gpu: bool | None = Field(None, description="Force multi-GPU on/off. If omitted, API auto-selects torchrun when 2+ GPUs are visible")
    nproc_per_node: int | None = Field(None, description="Number of GPUs for multi-GPU training. If omitted, API uses visible GPU count")

Metrics requests

VUNMetricsRequest

Bases: BaseModel

Request body for POST /metrics/vun.

Source code in _utils/_api/metrics.py

class VUNMetricsRequest(BaseModel):
    """Request body for `POST /metrics/vun`."""
    gen_data: str = Field(..., description="Path to generated structures parquet")
    huggingface_dataset: str | None = Field(None, description="HuggingFace dataset for novelty comparison")
    output_csv: str | None = Field(None, description="Output CSV file")
    output_parquet: str | None = Field(None, description="Optional processed parquet with VUN columns")
    sort_metrics_by: Literal['all', 'Condition Vector', 'both'] = Field("all", description="Grouping mode for VUN metrics output")
    num_workers: int = Field(8, description="Number of parallel workers")
    load_processed_data: str | None = Field(None, description="Pre-processed novelty reference parquet")
    check_comp_novelty: bool = Field(False, description="Whether to compute compositional novelty")
    skip_validity: bool = Field(False, description="Skip validity checks")
    skip_uniqueness: bool = Field(False, description="Skip uniqueness checks")
    bond_length_acceptability_cutoff: float = Field(1.0, description="Cutoff used in bond-length validity checks")
    allow_stated_p1_mismatch: bool = Field(False, description="Allow P1 space-group mismatch during validity checks")

EHullMetricsRequest

Bases: BaseModel

Request body for POST /metrics/ehull.

The endpoint computes the hull metrics using MACE-based structure relaxation rather than DFT.

Source code in _utils/_api/metrics.py

class EHullMetricsRequest(BaseModel):
    """Request body for `POST /metrics/ehull`.

    The endpoint computes the hull metrics using MACE-based structure relaxation rather than DFT.
    """
    post_parquet: str = Field(..., description="Path to postprocessed structures")
    output_parquet: str = Field(..., description="Output parquet with stability metrics")
    mp_data: str = Field("data/mp_computed_structure_entries.json.gz", description="Path to MP computed structure entries data")
    cif_column: str = Field("Generated CIF", description="Name of the CIF column in the parquet input")
    num_workers: int = Field(4, description="Number of parallel workers")
    batch_size: int = Field(100, description="Batch size for MACE E-hull processing")

XRDMetricsRequest

Bases: BaseModel

Request body for POST /metrics/xrd.

Source code in _utils/_api/metrics.py

class XRDMetricsRequest(BaseModel):
    """Request body for `POST /metrics/xrd`."""
    input_parquet: str = Field(..., description="Path to parquet file with generated structures")
    output_parquet: str = Field(..., description="Path to output XRD metrics parquet")
    num_gens: int = Field(0, description="Maximum generations per structure to score")
    num_workers: int = Field(2, description="Number of parallel workers")
    ref_parquet: str | None = Field(None, description="Optional override parquet with reference structures")
    validity_check: Literal['diffcsp', 'crystallm', 'none'] = Field("diffcsp", description="Validity check mode")
    sort_gens: Literal['rank', 'random', 'first'] = Field("rank", description="How to choose generations when num_gens is limited")

PropertyMetricsRequest

Bases: BaseModel

Request body for POST /metrics/property.

Source code in _utils/_api/metrics.py

class PropertyMetricsRequest(BaseModel):
    """Request body for `POST /metrics/property`."""
    post_parquet: str = Field(..., description="Path to processed parquet file with VUN metrics")
    parquet_out: str | None = Field(None, description="Optional parquet output with property metrics columns")
    metrics_out: str | None = Field(None, description="Optional CSV output with property metrics summary")
    sort_metrics_by: Literal['all', 'Condition Vector', 'both'] = Field("all", description="How to group property metrics")
    num_workers: int = Field(8, description="Number of parallel workers")
    property_targets: str = Field("[]", description="JSON list of property columns in condition-vector order")
    property1_normaliser: Literal['power_log', 'linear', 'None'] = Field("None", description="Normalization method for the first property")
    property2_normaliser: Literal['power_log', 'linear', 'None'] = Field("None", description="Normalization method for the second property")
    property3_normaliser: Literal['power_log', 'linear', 'None'] = Field("None", description="Normalization method for the third property")
    max_property1: float = Field(17.89, description="Maximum value for the first property")
    max_property2: float = Field(17.89, description="Maximum value for the second property")
    max_property3: float = Field(17.89, description="Maximum value for the third property")
    min_property1: float = Field(0.0, description="Minimum value for the first property")
    min_property2: float = Field(0.0, description="Minimum value for the second property")
    min_property3: float = Field(0.0, description="Minimum value for the third property")

Virtualiser requests

VirtualiseRequest

Bases: BaseModel

Request body for POST /virtualise.

Source code in _utils/_api/virtualiser.py

class VirtualiseRequest(BaseModel):
    """Request body for `POST /virtualise`."""
    input_cif: str = Field(..., description="Path to input ordered CIF file.")
    output_cif: str = Field(..., description="Path for the output virtual crystal CIF.")
    config_file: str | None = Field(
        None,
        description=(
            "Path to a YAML config file specifying virtual_pairs, symprec, and angle_tolerance. "
            "If omitted, provide virtual_pairs directly."
        ),
    )
    virtual_pairs: list[list[str]] | None = Field(
        None,
        description=(
            'Element pairs to virtualise, e.g. [["Mg", "Zn"]]. '
            "Required when config_file is not provided."
        ),
    )
    symprec: float = Field(0.003, description="Symmetry precision in Å (used when config_file is not provided).")
    angle_tolerance: float = Field(0.5, description="Angle tolerance in degrees (used when config_file is not provided).")