Skip to content

darts_acquisition.s2.raw_data_store

Raw Data Store for Sentinel 2 data.

SceneItem module-attribute 

SceneItem = typing.TypeVar('SceneItem')

logger module-attribute 

logger = logging.getLogger(
    __name__.replace("darts_", "darts.")
)

StoreManager 

StoreManager(
    bands: list[str],
    store: str | pathlib.Path | None = None,
)

Bases: abc.ABC, typing.Generic[darts_acquisition.s2.raw_data_store.SceneItem]

Manager for storing raw sentinel 2 data.

This class is an abstract base class and should be extended to implement the respective downloading methods.

Usage:

1. "Normal" usage:

```python
    store_manager = StoreManager(store_path)
    ds_s2 = store_manager.load(identifier, bands)
```

2. Force download:

```python
    store_manager = StoreManager(store_path)
    ds_s2 = store_manager.load(identifier, force=True)
```

3. Download only (and only if missing) and store the scene:

```python
    store_manager = StoreManager(store_path)
    store_manager.download(identifier) # store_path must be not None
```

4. Offline mode:

```python
    store_manager = StoreManager(store_path)
    store_manager.open(identifier) # store_path must be not None, bands must be complete
```

Initialize the store manager.

Parameters:

  • bands (list[str]) –

    List of bands to manage

  • store (str | pathlib.Path | None, default: None ) –

    Directory path for storing raw sentinel 2 data

Source code in darts-acquisition/src/darts_acquisition/s2/raw_data_store.py 
54
55
56
57
58
59
60
61
62
63
def __init__(self, bands: list[str], store: str | Path | None = None):
    """Initialize the store manager.

    Args:
        bands (list[str]): List of bands to manage
        store (str | Path | None): Directory path for storing raw sentinel 2 data

    """
    self.bands = bands
    self.store = Path(store) if isinstance(store, str) else store

bands instance-attribute 

bands = darts_acquisition.s2.raw_data_store.StoreManager(
    bands
)

store instance-attribute 

store = (
    pathlib.Path(
        darts_acquisition.s2.raw_data_store.StoreManager(
            store
        )
    )
    if isinstance(
        darts_acquisition.s2.raw_data_store.StoreManager(
            store
        ),
        str,
    )
    else darts_acquisition.s2.raw_data_store.StoreManager(
        store
    )
)

complete 

complete(identifier: str) -> bool

Check if a scene in the store contains all requested bands.

Parameters:

  • identifier (str) –

    Unique identifier for the scene

Returns:

  • bool ( bool ) –

    True if all requested bands are present, False otherwise

Source code in darts-acquisition/src/darts_acquisition/s2/raw_data_store.py 
104
105
106
107
108
109
110
111
112
113
114
def complete(self, identifier: str) -> bool:
    """Check if a scene in the store contains all requested bands.

    Args:
        identifier (str): Unique identifier for the scene

    Returns:
        bool: True if all requested bands are present, False otherwise

    """
    return len(self.missing_bands(identifier)) == 0

download_and_store 

download_and_store(
    item: str
    | darts_acquisition.s2.raw_data_store.SceneItem,
)

Download a scene from the source and store it in the local store.

Store must be provided! Will do nothing if all required bands are already present.

Parameters:

Source code in darts-acquisition/src/darts_acquisition/s2/raw_data_store.py 
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
def download_and_store(self, item: str | SceneItem):
    """Download a scene from the source and store it in the local store.

    Store must be provided!
    Will do nothing if all required bands are already present.

    Args:
        item (str | SceneItem): Item or scene-id to open.

    """
    assert self.store is not None, "Store must be provided to download and store scenes!"
    identifier = self.identifier(item)
    missing_bands = self.missing_bands(identifier)
    if not missing_bands:
        return
    dataset = self.download_scene_from_source(item, missing_bands)
    self.save_to_store(dataset, identifier)

download_scene_from_source abstractmethod 

download_scene_from_source(
    item: str
    | darts_acquisition.s2.raw_data_store.SceneItem,
    bands: list[str],
) -> xarray.Dataset
Source code in darts-acquisition/src/darts_acquisition/s2/raw_data_store.py 
215
216
@abstractmethod
def download_scene_from_source(self, item: str | SceneItem, bands: list[str]) -> xr.Dataset: ...  # noqa: D102

encodings abstractmethod 

encodings(bands: list[str]) -> dict[str, dict[str, str]]
Source code in darts-acquisition/src/darts_acquisition/s2/raw_data_store.py 
212
213
@abstractmethod
def encodings(self, bands: list[str]) -> dict[str, dict[str, str]]: ...  # noqa: D102

exists 

exists(identifier: str) -> bool

Check if a scene already exists in the local raw data store.

Parameters:

  • identifier (str) –

    Unique identifier for the scene

Returns:

  • bool ( bool ) –

    True if the scene exists in the store, False otherwise

Source code in darts-acquisition/src/darts_acquisition/s2/raw_data_store.py 
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
def exists(self, identifier: str) -> bool:
    """Check if a scene already exists in the local raw data store.

    Args:
        identifier (str): Unique identifier for the scene

    Returns:
        bool: True if the scene exists in the store, False otherwise

    """
    if not self.store:
        return False

    scene_path = self.store / f"{identifier}.zarr"
    return scene_path.exists()

identifier abstractmethod 

identifier(
    item: str
    | darts_acquisition.s2.raw_data_store.SceneItem,
) -> str
Source code in darts-acquisition/src/darts_acquisition/s2/raw_data_store.py 
209
210
@abstractmethod
def identifier(self, item: str | SceneItem) -> str: ...  # noqa: D102

load 

load(
    item: str
    | darts_acquisition.s2.raw_data_store.SceneItem,
    force: bool = False,
) -> xarray.Dataset

Load a scene.

If force==True will download the scene from source even if present in store. Else, will try to open the scene from store first and only download missing bands. Will always store the downloaded scene in local store if store is provided, potentially overwriting existing.

Parameters:

Returns:

  • xarray.Dataset –

    xr.Dataset: The loaded scene as xarray Dataset

Source code in darts-acquisition/src/darts_acquisition/s2/raw_data_store.py 
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
def load(self, item: str | SceneItem, force: bool = False) -> xr.Dataset:
    """Load a scene.

    If `force==True` will download the scene from source even if present in store.
    Else, will try to open the scene from store first and only download missing bands.
    Will always store the downloaded scene in local store if store is provided, potentially overwriting existing.

    Args:
        item (str | SceneItem): Item or scene-id to open.
        force (bool, optional): If True, will download the scene even if present. Defaults to False.

    Returns:
        xr.Dataset: The loaded scene as xarray Dataset

    """
    identifier = self.identifier(item)
    if force:
        logger.debug(f"Force downloading scene {identifier} from source.")
        dataset = self.download_scene_from_source(item, self.bands)
        if self.store:
            self.save_to_store(dataset, identifier)
        return dataset

    missing_bands = self.missing_bands(identifier)
    if not missing_bands:
        logger.debug(f"Scene {identifier} is complete, opening from store.")
        return self.open(item)
    logger.debug(f"Scene {identifier} is missing bands {missing_bands}, downloading from source.")
    dataset = self.download_scene_from_source(item, missing_bands)
    if self.store:
        self.save_to_store(dataset, identifier)
    return dataset

missing_bands 

missing_bands(identifier: str) -> list[str]

Get the list of missing bands for a scene in the store.

Parameters:

  • identifier (str) –

    Unique identifier for the scene

Returns:

  • list[str] –

    list[str]: List of missing bands

Source code in darts-acquisition/src/darts_acquisition/s2/raw_data_store.py 
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
def missing_bands(self, identifier: str) -> list[str]:
    """Get the list of missing bands for a scene in the store.

    Args:
        identifier (str): Unique identifier for the scene

    Returns:
        list[str]: List of missing bands

    """
    if not self.store:
        return self.bands

    scene_path = self.store / f"{identifier}.zarr"
    if not scene_path.exists():
        return self.bands

    dataset = xr.open_zarr(scene_path, consolidated=False)
    required_bands = set(self.bands)
    present_bands = set(dataset.data_vars)
    missing = required_bands - present_bands
    return list(missing)

open 

open(
    item: str
    | darts_acquisition.s2.raw_data_store.SceneItem,
) -> xarray.Dataset

Open a scene from local store.

Store must be provided and the scene must be present in store!

Parameters:

Returns:

  • xarray.Dataset –

    xr.Dataset: The opened scene as xarray Dataset

Source code in darts-acquisition/src/darts_acquisition/s2/raw_data_store.py 
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
def open(self, item: str | SceneItem) -> xr.Dataset:
    """Open a scene from local store.

    Store must be provided and the scene must be present in store!

    Args:
        item (str | SceneItem): Item or scene-id to open

    Returns:
        xr.Dataset: The opened scene as xarray Dataset

    """
    identifier = self.identifier(item)
    assert self.complete(identifier), f"Scene {identifier} is incomplete in store!"
    scene_path = self.store / f"{identifier}.zarr"
    return xr.open_zarr(scene_path, consolidated=False).set_coords("spatial_ref").load()

save_to_store 

save_to_store(
    dataset: xarray.Dataset, identifier: str
) -> None

Save a scene dataset to the local raw data store.

Will append new bands to existing store if scene already exists. Will overwrite existing bands in an existing store if scene already exists.

Parameters:

  • dataset (xarray.Dataset) –

    Dataset to save

  • identifier (str) –

    Unique identifier for the scene

Source code in darts-acquisition/src/darts_acquisition/s2/raw_data_store.py 
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
def save_to_store(self, dataset: xr.Dataset, identifier: str) -> None:
    """Save a scene dataset to the local raw data store.

    Will append new bands to existing store if scene already exists.
    Will overwrite existing bands in an existing store if scene already exists.

    Args:
        dataset (xr.Dataset): Dataset to save
        identifier (str): Unique identifier for the scene

    """
    assert self.store is not None, "Store must be provided to save scenes!"
    scene_path = self.store / f"{identifier}.zarr"
    encoding = self.encodings(list(dataset.data_vars))
    if not scene_path.exists():
        dataset.to_zarr(scene_path, encoding=encoding, consolidated=False, mode="w")
    else:
        # Assert that the coordinates match
        existing_dataset = xr.open_zarr(scene_path, consolidated=False)
        xr.testing.assert_allclose(existing_dataset.coords, dataset.coords)
        # Overwrite dataset coords to avoid conflicts by floating point precision issues
        dataset["x"] = existing_dataset.x
        dataset["y"] = existing_dataset.y
        dataset.to_zarr(scene_path, encoding=encoding, consolidated=False, mode="a")