Skip to content

Logging & Verbosity

log

filter_function

filter_function(record)

Determine which logs to include based on the verbosity level.

Source code in rekx/log.py
def filter_function(record):
    """Determine which logs to include based on the verbosity level."""
    # Assuming record["level"].name gives the log level like "INFO", "DEBUG", etc.
    # return verbose
    return record["level"].name == "INFO" if verbosity_level == 1 else True

progress

messages

print

print_metadata_series_long_table

print_metadata_series_long_table(
    metadata_series: dict, group_metadata=False
)
Source code in rekx/print.py
def print_metadata_series_long_table(
    metadata_series: dict,
    group_metadata=False,
):
    """ """
    console = Console()
    columns = []
    columns.append("Name")
    columns.append("Size")
    columns.append("Dimensions")
    metadata_series_level_one = next(iter(metadata_series.values()))
    variables_metadata = metadata_series_level_one.get("Variables", {})
    columns.append("Variable")
    # Add columns from the first variable's metadata dictionary
    for key in next(iter(variables_metadata.values())).keys():
        columns.append(key.replace("_", " ").title())
    dimensions = metadata_series_level_one.get("Dimensions", {})
    dimensions_sort_order = ["bnds", "time", "lon", "lat"]
    dimension_attributes_sorted = {
        key for key in dimensions_sort_order if key in dimensions
    }
    dimension_attributes = " x ".join(
        [f"[bold]{dimension}[/bold]" for dimension in dimension_attributes_sorted]
    )
    caption = f"Dimensions: {dimension_attributes} | "
    caption += f"Cache [bold]size[/bold] in bytes | "
    caption += f"[bold]Number of elements[/bold] | "
    caption += f"[bold]Preemption strategy[/bold] ranging in [0, 1] | "
    repetitions = metadata_series_level_one.get("Repetitions", None)
    caption += (
        f"Average time of [bold]{repetitions}[/bold] reads in [bold]seconds[/bold]"
    )
    table = Table(
        *columns,
        caption=caption,
        show_header=True,
        header_style="bold magenta",
        box=SIMPLE_HEAD,
    )

    # Process each file's metadata
    for filename, metadata in metadata_series.items():
        filename = metadata.get("File name", NOT_AVAILABLE)
        file_size = metadata.get("File size", NOT_AVAILABLE)

        dimensions = metadata.get("Dimensions", {})
        dimensions_sorted = {
            key: dimensions[key] for key in dimensions_sort_order if key in dimensions
        }
        dimension_shape = " x ".join(
            [f"{size}" for dimension, size in dimensions_sorted.items()]
        )

        variables_metadata = metadata.get("Variables", {})
        # row = []
        for variable, details in variables_metadata.items():
            if "Compression" in details:
                compression_details = format_compression(details["Compression"])
                details["Compression"] = compression_details["Filters"]
                details["Level"] = compression_details["Level"]

            row = [filename, str(file_size), dimension_shape, variable]
            row += [str(details.get(key, NOT_AVAILABLE)) for key in details.keys()]
            table.add_row(*row)

        if group_metadata:
            table.add_row("")  # Add an empty line between 'files' for clarity

    console.print(table)

print_metadata_series_table

print_metadata_series_table(
    metadata_series: dict, group_metadata=False
)
Source code in rekx/print.py
def print_metadata_series_table(
    metadata_series: dict,
    group_metadata=False,
):
    """ """
    for filename, metadata in metadata_series.items():
        filename = metadata.get("File name", "N/A")
        file_size = metadata.get("File size", "N/A")
        dimensions = metadata.get("Dimensions", {})
        dimensions_string = ", ".join(
            [f"{dimension}: {size}" for dimension, size in dimensions.items()]
        )
        caption = f"File size: {file_size} bytes, Dimensions: {dimensions_string}"
        caption += f"\n* Cache: Size in bytes, Number of elements, Preemption ranging in [0, 1]"
        variables_metadata = metadata.get("Variables")
        if variables_metadata:
            table = Table(
                title=f"[bold]{filename}[/bold]",
                caption=caption,
                show_header=True,
                header_style="bold magenta",
                box=SIMPLE_HEAD,
            )
            # table = Table(caption=caption, show_header=True, header_style="bold magenta", box=SIMPLE_HEAD)
            table.add_column("Variable", style="dim", no_wrap=True)

            # Expectedly all variables feature the same keys
            for key in next(iter(variables_metadata.values())).keys():
                table.add_column(key.replace("_", " ").title(), no_wrap=True)

            for variable, details in variables_metadata.items():
                if "Compression" in details:
                    compression_details = format_compression(details["Compression"])
                    details["Compression"] = compression_details["Filters"]
                    details["Level"] = compression_details["Level"]

                row = [variable] + [
                    str(details.get(key, ""))
                    for key in next(iter(variables_metadata.values())).keys()
                ]
                table.add_row(*row)

            console = Console()
            console.print(table)
            if group_metadata:
                console.print("\n")  # Add an empty line between groups for clarity

print_metadata_table

print_metadata_table(metadata)
Source code in rekx/print.py
def print_metadata_table(metadata):
    """ """
    filename = metadata.get("File name", "N/A")
    file_size = metadata.get("File size", "N/A")
    dimensions = metadata.get("Dimensions", {})
    dimensions_string = ", ".join(
        [f"{dimension}: {size}" for dimension, size in dimensions.items()]
    )
    caption = f"File size: {file_size} bytes, Dimensions: {dimensions_string}"
    caption += (
        f"\n* Cache: Size in bytes, Number of elements, Preemption ranging in [0, 1]"
    )

    variables_metadata = metadata.get("Variables")
    if variables_metadata:
        table = Table(
            title=filename,
            caption=caption,
            show_header=True,
            header_style="bold magenta",
            box=SIMPLE_HEAD,
        )
        table.add_column("Variable", style="dim", no_wrap=True)

        # Dynamically add columns based on the keys of the nested dictionaries
        # Assuming all variables have the same set of keys
        for key in next(iter(variables_metadata.values())).keys():
            table.add_column(key.replace("_", " ").title(), no_wrap=True)

        for variable, details in variables_metadata.items():
            # Format compression dictionary into a readable string
            if "Compression" in details:
                compression_details = format_compression(details["Compression"])
                details["Compression"] = compression_details["Filters"]
                details["Level"] = compression_details["Level"]

            row = [variable] + [
                str(details.get(key, ""))
                for key in next(iter(variables_metadata.values())).keys()
            ]
            table.add_row(*row)

        console = Console()
        console.print(table)