darts_export

Dataset export for the DARTS dataset.

Functions:

• export_tile –

  Export a tile to a file.

• missing_outputs –

  Check for missing output files in the given directory.

Attributes:

• __version__ –

__version__ module-attribute

__version__ = importlib.metadata.version('darts-nextgen')

export_tile

export_tile(
    tile: xarray.Dataset,
    out_dir: pathlib.Path,
    bands: list[str] = [
        "probabilities",
        "binarized",
        "polygonized",
        "extent",
        "thumbnail",
    ],
    ensemble_subsets: list[str] = [],
)

Export a tile to a file.

Parameters:

• tile (xarray.Dataset) –

  The tile to export.

• out_dir (pathlib.Path) –

  The path where to export to.

• bands (list[str], default: ['probabilities', 'binarized', 'polygonized', 'extent', 'thumbnail'] ) –

  The bands to export. Defaults to ["probabilities"].

• ensemble_subsets (list[str], default: [] ) –

  The ensemble subsets to export. Defaults to [].

Raises:

• ValueError –

  If the band is not found in the tile.

Source code in darts-export/src/darts_export/export.py

@stopuhr.funkuhr("Exporting tile", logger.debug, print_kwargs=["bands", "ensemble_subsets"])
def export_tile(  # noqa: C901
    tile: xr.Dataset,
    out_dir: Path,
    bands: list[str] = ["probabilities", "binarized", "polygonized", "extent", "thumbnail"],
    ensemble_subsets: list[str] = [],
):
    """Export a tile to a file.

    Args:
        tile (xr.Dataset): The tile to export.
        out_dir (Path): The path where to export to.
        bands (list[str], optional): The bands to export. Defaults to ["probabilities"].
        ensemble_subsets (list[str], optional): The ensemble subsets to export. Defaults to [].

    Raises:
        ValueError: If the band is not found in the tile.

    """
    out_dir.mkdir(parents=True, exist_ok=True)

    for band in bands:
        match band:
            case "polygonized":
                _export_polygonized(tile, out_dir, ensemble_subsets)
            case "binarized":
                _export_binarized(tile, out_dir, ensemble_subsets)
            case "probabilities":
                _export_probabilities(tile, out_dir, ensemble_subsets)
            case "extent":
                _export_vector(tile, "extent", out_dir, fname="prediction_extent")
            case "thumbnail":
                _export_thumbnail(tile, out_dir)
            case "optical":
                _export_raster(tile, ["red", "green", "blue", "nir"], out_dir, fname="optical")
            case "dem":
                _export_raster(tile, ["slope", "relative_elevation"], out_dir, fname="dem")
            case "tcvis":
                _export_raster(tile, ["tc_brightness", "tc_greenness", "tc_wetness"], out_dir, fname="tcvis")
            case _:
                if band not in tile.data_vars:
                    raise ValueError(
                        f"Band {band} not found in tile for export. Available bands are: {list(tile.data_vars.keys())}"
                    )
                # Export the band as a raster
                _export_raster(tile, band, out_dir)

missing_outputs

missing_outputs(
    out_dir: pathlib.Path,
    bands: list[str] = [
        "probabilities",
        "binarized",
        "polygonized",
        "extent",
        "thumbnail",
    ],
    ensemble_subsets: list[str] = [],
) -> typing.Literal["all", "some", "none"]

Check for missing output files in the given directory.

Parameters:

• out_dir (pathlib.Path) –

  The directory to check for missing files.

• bands (list[str], default: ['probabilities', 'binarized', 'polygonized', 'extent', 'thumbnail'] ) –

  The bands to export. Defaults to ["probabilities"].

• ensemble_subsets (list[str], default: [] ) –

  The ensemble subsets to export. Defaults to [].

Returns:

• typing.Literal['all', 'some', 'none'] –

  Literal["all", "some", "none"]: A string indicating the status of missing files: - "none": No files are missing. - "some": Some files are missing, which one will be logged to debug. - "all": All files are missing.

Raises:

• ValueError –

  If the output path is not a directory.

Source code in darts-export/src/darts_export/check.py

def missing_outputs(  # noqa: C901
    out_dir: Path,
    bands: list[str] = ["probabilities", "binarized", "polygonized", "extent", "thumbnail"],
    ensemble_subsets: list[str] = [],
) -> Literal["all", "some", "none"]:
    """Check for missing output files in the given directory.

    Args:
        out_dir (Path): The directory to check for missing files.
        bands (list[str], optional): The bands to export. Defaults to ["probabilities"].
        ensemble_subsets (list[str], optional): The ensemble subsets to export. Defaults to [].

    Returns:
        Literal["all", "some", "none"]: A string indicating the status of missing files:
            - "none": No files are missing.
            - "some": Some files are missing, which one will be logged to debug.
            - "all": All files are missing.

    Raises:
        ValueError: If the output path is not a directory.

    """
    if not out_dir.exists():
        return []
    if not out_dir.is_dir():
        raise ValueError(f"Output path {out_dir} is not a directory.")
    expected_files = []
    for band in bands:
        match band:
            case "polygonized":
                expected_files += ["prediction_segments.gpkg"] + [
                    f"prediction_segments-{es}.gpkg" for es in ensemble_subsets
                ]
                expected_files += ["prediction_segments.parquet"] + [
                    f"prediction_segments-{es}.parquet" for es in ensemble_subsets
                ]
            case "binarized":
                expected_files += ["binarized.tif"] + [f"binarized-{es}.tif" for es in ensemble_subsets]
            case "probabilities":
                expected_files += ["probabilities.tif"] + [f"probabilities-{es}.tif" for es in ensemble_subsets]
            case "extent":
                expected_files += ["extent.gpkg", "extent.parquet"]
            case "thumbnail":
                expected_files += ["thumbnail.jpg"]
            case _:
                expected_files += [f"{band}.tif"]

    missing_files = _missing_files(out_dir, expected_files)
    if len(missing_files) == 0:
        return "none"
    elif len(missing_files) == len(expected_files):
        return "all"
    else:
        logger.debug(
            f"Missing files in {out_dir}: {', '.join(missing_files)}. Expected files: {', '.join(expected_files)}."
        )
        return "some"

• export_tile
• missing_outputs