corrlib/corrlib/cli.py

from typing import Optional
import typer
from corrlib import __app_name__

from .initialization import create
from .toml import import_tomls, update_project, reimport_project
from .find import find_record, list_projects
from .tools import str2list
from .main import update_aliases
from .meas_io import drop_cache as mio_drop_cache
from .meas_io import load_record as mio_load_record
from .integrity import full_integrity_check

import os
from pyerrors import Corr
from importlib.metadata import version
from pathlib import Path


app = typer.Typer()


def _version_callback(value: bool) -> None:
    if value:
        print(__app_name__, version(__app_name__))
        raise typer.Exit()


@app.command()
def update(
    path: Path = typer.Option(
        Path('./corrlib'),
        "--dataset",
        "-d",
        ),
    uuid: str = typer.Argument(),
    ) -> None:
    """
    Update a project by it's UUID.
    """
    update_project(path, uuid)
    return


@app.command()
def lister(
    path: Path = typer.Option(
        Path('./corrlib'),
        "--dataset",
        "-d",
    ),
    entities: str = typer.Argument('ensembles'),
    ) -> None:
    """
    List entities. (ensembles, projects)
    """
    if entities in ['ensembles', 'Ensembles','ENSEMBLES']:
        print("Ensembles:")
        for item in os.listdir(path / "archive"):
            if os.path.isdir(path / "archive" / item):
                print(item)
    elif entities == 'projects':
        results = list_projects(path)
        print("Projects:")
        header = "UUID".ljust(37) + "| Aliases"
        print(header)
        for project in results:
            if project[1] is not None:
                aliases = " | ".join(str2list(project[1]))
            else:
                aliases = "---"
            print(project[0], "|", aliases)
    return


@app.command()
def alias_add(
    path: Path = typer.Option(
        Path('./corrlib'),
        "--dataset",
        "-d",
        ),
    uuid: str = typer.Argument(),
    alias: str = typer.Argument(),
    ) -> None:
    """
    Add an alias to a project UUID.
    """
    alias_list = alias.split(",")
    update_aliases(path, uuid, alias_list)
    return


@app.command()
def find(
    path: Path = typer.Option(
        Path('./corrlib'),
        "--dataset",
        "-d",
        ),
    ensemble: str = typer.Argument(),
    corr: str = typer.Argument(),
    code: str = typer.Argument(),
    arg: str = typer.Option(
        str('all'),
        "--argument",
        "-a",
    ),
    ) -> None:
    """
    Find a record in the backlog at hand. Through specifying it's ensemble and the measured correlator.
    """
    results = find_record(path, ensemble, corr, code)
    if results.empty:
        return
    if arg == 'all':
        print(results)
    else:
        for r in results[arg].values:
            print(r)


@app.command()
def stat(
    path: Path = typer.Option(
        Path('./corrlib'),
        "--dataset",
        "-d",
        ),
    record_id: str = typer.Argument(),
    ) -> None:
    """
    Show the statistics of a given record.
    """
    record = mio_load_record(path, record_id)
    if isinstance(record, (list, Corr)):
        record = record[0]
    statistics = record.idl
    print(statistics)
    return


@app.command()
def check(path: Path = typer.Option(
        Path('./corrlib'),
        "--dataset",
        "-d",
        ),
    ) -> None:
    full_integrity_check(path)


@app.command()
def importer(
    path: Path = typer.Option(
        Path('./corrlib'),
        "--dataset",
        "-d",
        ),
    files: str = typer.Argument(
        ),
    copy_file: bool = typer.Option(
        bool(True),
        "--save",
        "-s",
        ),
    ) -> None:
    """
    Import a project from a .toml-file via CLI.
    """
    file_list = files.split(",")
    import_tomls(path, file_list, copy_file)
    return


@app.command()
def reimporter(
    path: Path = typer.Option(
        Path('./corrlib'),
        "--dataset",
        "-d",
        ),
    ident: str = typer.Argument()
    ) -> None:
    """
    Reimport the toml file identfied by the ident string.
    """
    uuid = ident.split("::")[0]
    if len(ident.split("::")) > 1:
        toml_file = os.path.join(path, "toml_imports", ident.split("::")[1])
        if os.path.exists(toml_file):
            import_tomls(path, [toml_file], copy_files=False)
        else:
            raise Exception("This file is not known for this project.")
    else:
        reimport_project(path, uuid)
    return


@app.command()
def init(
    path: Path = typer.Option(
        Path('./corrlib'),
        "--dataset",
        "-d",
        ),
    tracker: str = typer.Option(
        str('datalad'),
        "--tracker",
        "-t",
        ),
    ) -> None:
    """
    Initialize a new backlog-database.
    """
    create(path, tracker)
    return


@app.command()
def drop_cache(
    path: Path = typer.Option(
        Path('./corrlib'),
        "--dataset",
        "-d",
        ),
    ) -> None:
    """
    Drop the currect cache directory of the dataset.
    """
    mio_drop_cache(path)
    return


@app.callback()
def main(
    version: Optional[bool] = typer.Option(
        None,
        "--version",
        "-v",
        help="Show the application's version and exit.",
        callback=_version_callback,
        is_eager=True,
        )
    ) -> None:
    return