dqm_ml_job.cli

Command-line interface for DQM job execution.

This module provides CLI functions for parsing arguments and running data quality assessment jobs from YAML configuration files.

logger = logging.getLogger(__name__) module-attribute

execute(arg_list: list[str] | None = None) -> None

Main CLI entry point for executing DQM jobs from YAML configurations. Args: arg_list: List of command line arguments (default: sys.argv[1:]).

Source code in packages/dqm-ml-job/src/dqm_ml_job/cli.py

def execute(arg_list: list[str] | None = None) -> None:
    """
    Main CLI entry point for executing DQM jobs from YAML configurations.
    Args:
        arg_list: List of command line arguments (default: sys.argv[1:]).
    """
    args = parse_args(arg_list)
    config: dict[str, Any] = {}

    for config_file in args.process_config:
        logger.debug("Executing job from config file: %s", config_file)

        config_path = Path(config_file).resolve()
        if not config_path.is_file():
            logger.error("Config file does not exist: %s", config_file)
            return

        with config_path.open() as stream:
            try:
                config_content = yaml.safe_load(stream)
                config.update(config_content)
            except yaml.YAMLError as exc:
                logger.error("Fail to part job configuration: %s", config_file)
                print(exc)
                return

    # if we succeed to load all config files, run the job

    # Optionally save the resolved configuration
    if args.save_config:
        logger.debug("Saving resolved configuration to: %s", args.save_config)
        save_path = Path(args.save_config).resolve()
        save_path.parent.mkdir(parents=True, exist_ok=True)
        with save_path.open("w") as stream:
            yaml.safe_dump(config, stream)

    run(config)

parse_args(arg_list: list[str] | None) -> Any

Parse command line arguments for the DQM job.

Parameters:

Name	Type	Description	Default
arg_list	list[str] | None	List of arguments (default: sys.argv[1:]).	required

Returns:

Type	Description
Any	The parsed Namespace object.

Source code in packages/dqm-ml-job/src/dqm_ml_job/cli.py

def parse_args(arg_list: list[str] | None) -> Any:
    """
    Parse command line arguments for the DQM job.

    Args:
        arg_list: List of arguments (default: sys.argv[1:]).

    Returns:
        The parsed Namespace object.
    """
    parser = argparse.ArgumentParser(
        prog="dqm-ml",
        description="DQM-ML Job client",
        epilog="for more informations see README",
    )

    parser.add_argument(
        "-p",
        "--process-config",
        type=str,
        nargs="+",
        required=True,
        help="configuration files to execute",
    )

    parser.add_argument(
        "--save-config",
        type=str,
        help="Path to save the resolved configuration",
    )

    # TODO add parameters to pass directly files / directory as inputs for loaders
    args = parser.parse_args(arg_list)

    return args

run(config: dict[str, Any]) -> None

Execute a job from a validated configuration dictionary.

The config is validated against JobConfig and must follow the v2 structure: - dataloaders: Contains loaders list and optional storage. - features: Optional interface with outputs and processors list. - metrics: Optional interface with outputs and processors list. - gap: Optional interface with outputs and processors list.

Source code in packages/dqm-ml-job/src/dqm_ml_job/cli.py

def run(config: dict[str, Any]) -> None:
    """
    Execute a job from a validated configuration dictionary.

    The config is validated against JobConfig and must follow the v2 structure:
    - dataloaders: Contains loaders list and optional storage.
    - features: Optional interface with outputs and processors list.
    - metrics: Optional interface with outputs and processors list.
    - gap: Optional interface with outputs and processors list.
    """
    if not config:
        raise ValueError("Job requires a configuration dictionary.")

    validated = JobConfig.model_validate(config)

    dataloaders_registry = PluginLoadedRegistry.get_dataloaders_registry()
    features_registry = PluginLoadedRegistry.get_features_registry()
    metrics_registry = PluginLoadedRegistry.get_metrics_registry()
    gap_registry = PluginLoadedRegistry.get_gap_registry()
    outputs_registry = PluginLoadedRegistry.get_outputwriter_registry()

    # Initialize dataloaders from list format
    dataloader_dicts = [loader.model_dump() for loader in validated.dataloaders.loaders]
    compute = _resolve_compute_config(validated)
    dl_storage = validated.dataloaders.storage.model_dump() if validated.dataloaders.storage else None
    for dl in dataloader_dicts:
        dl["threads"] = compute.threads
        if dl_storage and not dl.get("storage"):
            dl["storage"] = dl_storage
    dataloaders = _init_components_from_list(dataloader_dicts, dataloaders_registry, "dataloader")

    # Resolve storage config: interface override takes precedence over job-level
    def _resolve_storage(interface: Any) -> dict[str, Any] | None:
        if interface and interface.storage:
            result: dict[str, Any] = interface.storage.model_dump()
            return result
        if validated.storage:
            result = validated.storage.model_dump()
            return result
        return None

    # Initialize processors from all interfaces
    features_processors = _init_processors_from_interface(
        validated.features,
        features_registry,
        _resolve_storage(validated.features),
    )
    metrics_processors = _init_processors_from_interface(
        validated.metrics,
        metrics_registry,
        _resolve_storage(validated.metrics),
    )
    gap_processors = _init_processors_from_interface(
        validated.gap,
        gap_registry,
        _resolve_storage(validated.gap),
    )

    # Initialize output writers from interfaces
    features_output = _init_interface_outputs(
        validated.features,
        outputs_registry,
        "features",
        _resolve_storage(validated.features),
    )
    metrics_output = _init_interface_outputs(
        validated.metrics,
        outputs_registry,
        "metrics",
        _resolve_storage(validated.metrics),
    )
    delta_output = _init_interface_outputs(
        validated.gap,
        outputs_registry,
        "delta",
        _resolve_storage(validated.gap),
    )

    # Configure logging based on compute.log_level
    if compute.log_level:
        log_level = compute.log_level.upper()
        level = getattr(logging, log_level)
        logging.basicConfig(level=level)

    job = DatasetJob(
        dataloaders=dataloaders,
        features_processors=features_processors,
        metrics_processors=metrics_processors,
        gap_processors=gap_processors,
        features_output=features_output,
        progress_bar=compute.progress_bar,
        threads=compute.threads,
        errors_by_interface=_build_errors_by_interface(validated),
        compute_seed=compute.seed,
        compute_device=compute.device,
        compute_max_memory=compute.max_memory,
    )

    dataselection_metrics_list, delta_metrics_table = job.run()

    if metrics_output:
        metrics_output.write_metrics_dict(dataselection_metrics_list)
        _safe_flush(metrics_output)

    if delta_output and delta_metrics_table:
        delta_data = {col: delta_metrics_table.column(col) for col in delta_metrics_table.column_names}
        _enrich_delta_with_pairwise(validated, delta_data, delta_metrics_table)
        delta_output.write_table("delta", delta_data)
        _safe_flush(delta_output)