api

api

Module contains high level API of cssfinder.

TaskOptions dataclass

Container for extra task options.

Source code in cssfinder/api.py

@dataclass
class TaskOptions:
    """Container for extra task options."""

    is_debug: bool
    is_rich: bool = True

AmbiguousTaskKeyError

Bases: KeyError

Raised during report creation when name pattern selects more than one task.

Source code in cssfinder/api.py

class AmbiguousTaskKeyError(KeyError):
    """Raised during report creation when name pattern selects more than one task."""

run_project_from

run_project_from(
    project_file_path: Path | str,
    tasks: list[str] | None = None,
    *,
    is_debug: bool = False,
    is_rich: bool = True,
    force_sequential: bool = False,
    max_parallel: int = -1
) -> None

Load project and run all tasks.

Source code in cssfinder/api.py

def run_project_from(
    project_file_path: Path | str,
    tasks: list[str] | None = None,
    *,
    is_debug: bool = False,
    is_rich: bool = True,
    force_sequential: bool = False,
    max_parallel: int = -1,
) -> None:
    """Load project and run all tasks."""
    project = CSSFProject.load_project(project_file_path)
    logging.info(
        "Loaded project %r by %r <%r>.",
        project.meta.name,
        project.meta.author,
        project.meta.email,
    )
    run_project(
        project,
        tasks,
        is_debug=is_debug,
        is_rich=is_rich,
        force_sequential=force_sequential,
        max_parallel=max_parallel,
    )

run_project

run_project(
    project: CSSFProject,
    tasks: list[str] | None = None,
    *,
    is_debug: bool = False,
    is_rich: bool = True,
    force_sequential: bool = False,
    max_parallel: int = -1
) -> list[Task]

Run all tasks defined in project.

Source code in cssfinder/api.py

def run_project(
    project: CSSFProject,
    tasks: list[str] | None = None,
    *,
    is_debug: bool = False,
    is_rich: bool = True,
    force_sequential: bool = False,
    max_parallel: int = -1,
) -> list[Task]:
    """Run all tasks defined in project."""
    logging.debug("Running project %r", project.meta.name)

    message = "\n    |  ".join(project.json(indent=2).split("\n"))
    logging.info("%s", "\n    |  " + message)

    task_list = project.select_tasks(tasks)

    if force_sequential:
        for out in map(
            run_task,
            task_list,
            repeat(TaskOptions(is_debug=is_debug)),
        ):
            if isinstance(out, BaseException):
                raise out

    else:
        with ProcessPoolExecutor(
            max_parallel if max_parallel > 0 else None,
        ) as executor:
            out = executor.map(
                run_task,
                task_list,
                repeat(TaskOptions(is_debug=is_debug, is_rich=is_rich)),
            )
        for o in out:
            if isinstance(o, BaseException):
                raise o

    return task_list

run_task

run_task(task: Task, options: TaskOptions) -> None

Run task until completed.

Source code in cssfinder/api.py

def run_task(task: Task, options: TaskOptions) -> None:
    """Run task until completed."""
    try:
        return _run_task(task, options)
    except Exception as e:
        logging.critical(
            "Task %r failed due to exception:\n%s",
            task.task_name,
            str.join("", traceback.format_exception(type(e), e, e.__traceback__)),
        )
        raise

run_gilbert

run_gilbert(
    config: GilbertCfg,
    task_output_directory: Path,
    *,
    is_debug: bool = False,
    is_rich: bool = True
) -> None

Run Gilbert algorithm part of task.

Source code in cssfinder/api.py

def run_gilbert(
    config: GilbertCfg,
    task_output_directory: Path,
    *,
    is_debug: bool = False,
    is_rich: bool = True,
) -> None:
    """Run Gilbert algorithm part of task."""
    asset_io = GilbertIO()

    task_output_directory.mkdir(0o777, parents=True, exist_ok=True)
    logging.debug("Created directory: %r", task_output_directory.as_posix())

    algorithm = create_gilbert(config, asset_io, is_debug=is_debug)

    logging.info("Task %r started.", config.task_name)

    if is_rich:
        rich.print(Panel.fit(f"[blue]Task {config.task_name} started."))
    else:
        print("-----------------------")
        print(f"| Task {config.task_name} started |")
        print("-----------------------")

    for epoch_index in algorithm.run(
        max_epochs=config.runtime.max_epochs,
        iterations_per_epoch=config.runtime.iters_per_epoch,
        max_corrections=config.runtime.max_corrections,
    ):
        if corrections_count := algorithm.get_corrections_count():
            corrections = algorithm.get_corrections()
            state = algorithm.get_state()

            logging.info(
                "Executing epoch %r / %r (%.1f%%) - corrections: %r best: %r",
                epoch_index + 1,
                config.runtime.max_epochs,
                ((epoch_index + 1) / config.runtime.max_epochs) * 100,
                corrections_count,
                corrections[-1][2],
            )
            asset_io.dump_state(state, config.output_state_file)
            asset_io.dump_corrections(corrections, config.output_corrections_file)

        else:
            logging.info(
                "Executing epoch %r / %r (%.1f%%) - no corrections.",
                epoch_index + 1,
                config.runtime.max_epochs,
                ((epoch_index + 1) / config.runtime.max_epochs) * 100,
            )

    logging.warning("Task %r finished.", config.task_name)

    if is_rich:
        rich.print(Panel.fit(f"[blue]Task {config.task_name} finished."))
    else:
        print("-----------------------")
        print(f"| Task {config.task_name} finished |")
        print("-----------------------")

create_gilbert

create_gilbert(
    config: GilbertCfg,
    asset_io: GilbertIO,
    *,
    is_debug: bool
) -> Gilbert

Create Gilbert object from configuration with help of specified IO.

Parameters:

Name	Type	Description	Default
config	GilbertCfg	Algorithm configuration.	required
asset_io	GilbertIO	IO manager to use for loading assets.	required
is_debug	bool	Debug mode flag.	required

Returns:

Type	Description
Gilbert	Initialized

Source code in cssfinder/api.py

def create_gilbert(
    config: GilbertCfg,
    asset_io: GilbertIO,
    *,
    is_debug: bool,
) -> Gilbert:
    """Create Gilbert object from configuration with help of specified IO.

    Parameters
    ----------
    config : GilbertCfg
        Algorithm configuration.
    asset_io : GilbertIO
        IO manager to use for loading assets.
    is_debug : bool
        Debug mode flag.

    Returns
    -------
    Gilbert
        Initialized

    """
    state_config = config.get_state()

    initial_state = asset_io.load_state(state_config.expanded_file)

    if state_config.is_predefined_dimensions():
        depth = state_config.get_depth()
        quantity = state_config.get_quantity()
        logging.info("Using fixed dimensions depth=%r quantity=%r", depth, quantity)

    else:
        dimensions = ModeUtil.new(config.mode).get_dimensions(initial_state)
        depth = dimensions.depth
        quantity = dimensions.quantity
        logging.info("Deduced dimensions depth=%r quantity=%r", depth, quantity)

    symmetries = asset_io.load_symmetries(config.get_resources().symmetries)
    if symmetries:
        logging.info("Loaded symmetries:")
        for i, row in enumerate(symmetries):
            logging.info("Row %r: %r", i, [repr(sym.shape) for sym in row])
    else:
        logging.info("No symmetries provided.")

    projection = asset_io.load_projection(config.get_resources().projection)
    if projection is not None:
        logging.info("Loaded projection: %r", projection.shape)
    else:
        logging.info("No projection provided.")

    algorithm = Gilbert(
        initial=initial_state,
        depth=depth,
        quantity=quantity,
        mode=config.mode,
        backend=config.get_backend().name,
        precision=config.get_backend().precision,
        visibility=config.runtime.visibility,
        is_debug=is_debug,
    )
    algorithm.set_symmetries(symmetries)
    if projection is not None:
        algorithm.set_projection(projection)

    return algorithm

create_report_from

create_report_from(
    project_file_path: Path | str,
    task: str,
    reports: list[ReportType],
) -> Iterable[Report]

Load project (cssfproject.json) and create report for task selected by pattern.

Parameters:

Name	Type	Description	Default
project_file_path	Path | str	Path to cssfproject.json file or directory containing one.	required
task	str	Name or glob expression matching task name, expected to result in selection of single task.	required
reports	list[ReportType]	description	required

Returns:

Type	Description
Iterable[Report]	description

Yields:

Type	Description
Iterator[Iterable[Report]]	description

Source code in cssfinder/api.py

def create_report_from(
    project_file_path: Path | str,
    task: str,
    reports: list[ReportType],
) -> Iterable[Report]:
    """Load project (`cssfproject.json`) and create report for task selected by pattern.

    Parameters
    ----------
    project_file_path : Path | str
        Path to cssfproject.json file or directory containing one.
    task : str
        Name or glob expression matching task name, expected to result in selection of
        single task.
    reports : list[ReportType]
        _description_

    Returns
    -------
    Iterable[Report]
        _description_

    Yields
    ------
    Iterator[Iterable[Report]]
        _description_

    """
    project = CSSFProject.load_project(project_file_path)
    logging.info(
        "Loaded project %r by %r <%r>.",
        project.meta.name,
        project.meta.author,
        project.meta.email,
    )
    yield from create_report(project, task, reports)

create_report

create_report(
    project: CSSFProject,
    task: str,
    reports: list[ReportType],
) -> Iterable[Report]

Create report for task selected by pattern from project object.

Source code in cssfinder/api.py

def create_report(
    project: CSSFProject,
    task: str,
    reports: list[ReportType],
) -> Iterable[Report]:
    """Create report for task selected by pattern from project object."""
    tasks = project.select_tasks([task])

    for task_object in tasks:
        logging.info("Creating summary for task %s", task_object.task_name)

        manager = ReportManager(project, task_object)
        prepared_manager = manager.prepare()

        for report_type in reports:
            logging.info(
                "Report for task %s of type %s",
                task_object.task_name,
                report_type.name,
            )
            yield prepared_manager.request_report(report_type)