cli

champalimaud.cli

The commands: champalimaud status, fetch, and check.

This is the only module that defines commands.

check_command()

Check that this machine can query FlyWire CAVE.

Reads the materialization versions and table names, samples the tables every fetch reads, and looks the census LC16 ids up on the server. Exits 1, with how to save a token, on a login error.

Source code in champalimaud/cli.py

@app.command("check")
def check_command() -> None:
    """Check that this machine can query FlyWire CAVE.

    Reads the materialization versions and table names, samples the
    tables every fetch reads, and looks the census LC16 ids up on the
    server.
    Exits 1, with how to save a token, on a login error.
    """
    try:
        m = cave_materialize()
        inventory = check_inventory(m)
        say(f"version {inventory.version}; tables {inventory.tables}")
        nuclei, view_row, synapse = sample_tables(m)
        say(
            f"nuclei_v1 ok ({len(nuclei)} rows); "
            f"{Sources.SYNAPSE_VIEW} columns {list(view_row.columns)}; "
            f"{Sources.TRANSMITTER_TABLE} columns {list(synapse.columns)}"
        )
        ids = root_ids_of_type(load_census(), "LC16")
        say(
            f"LC16 census ids in nuclei_v1: {ids_in_nuclei(m, ids)} of "
            f"{len(ids)}"
        )
    # A smoke test reports any failure, whatever its type.
    except Exception as e:
        say(f"FAILED: {type(e).__name__}: {e}")
        if "401" in str(e) or "403" in str(e) or "token" in str(e).lower():
            say(TOKEN_HELP)
        if "503" in str(e):
            say("The materialize host is down.")
        raise typer.Exit(code=1) from e

fetch_command(names=None, yes=False, workers=4)

Fetch the datasets that are missing or unreadable.

On a terminal it asks before each fetch. Elsewhere it needs --yes, so a script never waits for an answer. A dataset named on the command line is fetched even if it is ok.

Source code in champalimaud/cli.py

@app.command("fetch")
def fetch_command(
    names: Annotated[
        list[str] | None,
        typer.Argument(
            help="Datasets to fetch; by default every one that is not ok."
        ),
    ] = None,
    yes: Annotated[
        bool,
        typer.Option("--yes", "-y", help="Fetch without asking."),
    ] = False,
    workers: Annotated[
        int, typer.Option(help="Batches of connections fetched at once.")
    ] = 4,
) -> None:
    """Fetch the datasets that are missing or unreadable.

    On a terminal it asks before each fetch.
    Elsewhere it needs --yes, so a script never waits for an answer.
    A dataset named on the command line is fetched even if it is ok.
    """
    unknown = [n for n in names or [] if n not in FETCHERS]
    if unknown:
        msg = f"unknown dataset {unknown}; choose from {list(FETCHERS)}"
        raise typer.BadParameter(msg)
    states = dict(status(DATASETS).select("dataset", "state").iter_rows())
    todo = names or [n for n in FETCHERS if states[n] != "ok"]
    show_status()
    if not todo:
        console.print("Every dataset is present and readable.")
        return
    if not yes and not interactive():
        console.print(
            f"Not a terminal: rerun with --yes to fetch {', '.join(todo)}."
        )
        raise typer.Exit(code=1)
    for name in todo:
        if not yes and not typer.confirm(f"Fetch {name}?", default=True):
            continue
        console.print(f"Fetching {name}...")
        # fetch_connections is the one fetch with a setting.
        fetch = (
            partial(fetch_connections, workers=workers)
            if name == "connections"
            else FETCHERS[name]
        )
        fetch()
    show_status()

human_size(size)

Format a byte count as a short string.

Parameters:

Name	Type	Description	Default
size	int or None	Bytes; None for a file that is not there.	required

Returns:

Type	Description
str	Such as "310 MB"; empty for None.

Examples:

>>> human_size(309971866)
'310 MB'
>>> human_size(901707)
'902 kB'

Source code in champalimaud/cli.py

def human_size(size: int | None) -> str:
    """Format a byte count as a short string.

    Parameters
    ----------
    size : int or None
        Bytes; ``None`` for a file that is not there.

    Returns
    -------
    str
        Such as ``"310 MB"``; empty for ``None``.

    Examples
    --------
    >>> human_size(309971866)
    '310 MB'
    >>> human_size(901707)
    '902 kB'
    """
    if size is None:
        return ""
    value = float(size)
    for unit in ("B", "kB", "MB", "GB"):
        if value < 1000 or unit == "GB":
            return f"{value:.0f} {unit}"
        value /= 1000
    return ""  # unreachable: the loop returns at GB

interactive()

Return whether stdin is a terminal, so a prompt can be shown.

Source code in champalimaud/cli.py

def interactive() -> bool:
    """Return whether stdin is a terminal, so a prompt can be shown."""
    return sys.stdin.isatty()

say(text)

Print text as it is, with no markup or highlighting.

Source code in champalimaud/cli.py

def say(text: str) -> None:
    """Print text as it is, with no markup or highlighting."""
    console.print(text, markup=False, highlight=False)

show_status()

Print the status of every dataset as a table.

Source code in champalimaud/cli.py

def show_status() -> None:
    """Print the status of every dataset as a table."""
    table = Table()
    for column in ("dataset", "source", "state", "size", "modified", "detail"):
        table.add_column(column)
    for row in status(DATASETS).iter_rows(named=True):
        color = COLORS[row["state"]]
        modified = (
            row["modified"].strftime("%Y-%m-%d") if row["modified"] else ""
        )
        table.add_row(
            row["dataset"],
            row["source"],
            f"[{color}]{row['state']}[/{color}]",
            human_size(row["size"]),
            modified,
            row["detail"],
        )
    console.print(table)

status_command()

Show which datasets are present and readable.

Source code in champalimaud/cli.py

@app.command("status")
def status_command() -> None:
    """Show which datasets are present and readable."""
    show_status()