Skip to content

darts_export

darts_export

Dataset export for the DARTS dataset.

__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] = [],
    metadata: dict = {},
    debug: bool = False,
)

Export a tile into a inference dataset, consisting of multiple files.

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 [].

  • metadata (dict, default: {} ) –

    Metadata to include in the export.

  • debug (bool, default: False ) –

    Debug mode: will write a .netcdf with all of the tiles contents. Defaults to False.

Raises:

  • ValueError –

    If the band is not found in the tile.

Source code in darts-export/src/darts_export/export.py 
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
@stopwatch.f("Exporting tile", printer=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] = [],
    metadata: dict = {},
    debug: bool = False,
):
    """Export a tile into a inference dataset, consisting of multiple files.

    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 [].
        metadata (dict, optional): Metadata to include in the export.
        debug (bool, optional): Debug mode: will write a .netcdf with all of the tiles contents. Defaults to False.

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

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

    if len(metadata) > 0:
        metadata["DARTS_exportdate"] = str(datetime.now(UTC))

    raster_tags = metadata

    for band in bands:
        match band:
            case "polygonized":
                _export_polygonized(tile, out_dir, ensemble_subsets)
            case "binarized":
                _export_binarized(tile, out_dir, ensemble_subsets, tags=raster_tags)
            case "probabilities":
                _export_probabilities(tile, out_dir, ensemble_subsets, tags=raster_tags)
            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", tags=raster_tags)
            case "dem":
                _export_raster(tile, ["slope", "relative_elevation"], out_dir, fname="dem", tags=raster_tags)
            case "tcvis":
                _export_raster(
                    tile, ["tc_brightness", "tc_greenness", "tc_wetness"], out_dir, fname="tcvis", tags=raster_tags
                )
            case "metadata":
                _export_metadata(out_dir, metadata)
            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, tags=raster_tags)

    if debug:
        manager.to_netcdf(tile, out_dir / "darts_inference_debug.nc", crop=False)

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 
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
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"