Skip to main content

filemanager.FileManager

Manages file system operations for Sentinel-1 products.

Usage

Source

filemanager.FileManager(
    base_dir,
    slc_dir=None,
    prod_dir=None,
    comp_dir=None,
)

All filename generation logic is centralized here. Storage directories (base_dir, slc_dir, …) may point to either local paths or S3 locations via upath.UPath. A directory given as a string is read as a local path, with ~ expanded.

Parameters

base_dir: str | Path | UPath

Root directory for all products. May be a local Path or a UPath pointing to an S3 bucket.

slc_dir: str | Path | UPath | None = None

Directory for SLC burst files. Defaults to base_dir/s1-slc.

prod_dir: str | Path | UPath | None = None

Directory for processed products. Defaults to base_dir/s1-prod.

comp_dir: str | Path | UPath | None = None
Directory for composites. Defaults to base_dir/s1-comp.

Raises

ValueError
If a directory is given as an empty string, or as a URL string rather than a UPath.

Methods

Name Description
find() Find the products of one type on disk, optionally filtered.
get_product_path() Get the path of a product’s file, or of one layer of a composite.
get_slc_path() Generate the file path for an SLC burst.
open() Open a product’s file, or one layer of a composite, for reading.
product_exists() Check if a product’s file exists (works for both local and S3).

find()

Find the products of one type on disk, optionally filtered.

Usage

Source

find(self, product_type: type[Burst], time_range: TimeRange | None = None, filters: Unpack[_SLCFilters] = {}) -> list[Burst]
find(self, product_type: type[B], time_range: TimeRange | None = None, filters: Unpack[_BurstProductFilters] = {}) -> list[B]
find(self, product_type: type[CoherenceProduct], time_range: TimeRange | None = None, filters: Unpack[_CoherenceFilters] = {}) -> list[CoherenceProduct]
find(self, product_type: type[RTCLRWComposite], time_range: TimeRange | None = None, filters: Unpack[_CompositeFilters] = {}) -> list[RTCLRWComposite]
find(self, product_type: type[CohLRWComposite], time_range: TimeRange | None = None, filters: Unpack[_CohCompositeFilters] = {}) -> list[CohLRWComposite]
find(self, product_type: type[RTCPassComposite], time_range: TimeRange | None = None, filters: Unpack[_PassCompositeFilters] = {}) -> list[RTCPassComposite]
find(self, product_type: type[GTCPassComposite], time_range: TimeRange | None = None, filters: Unpack[_PassCompositeFilters] = {}) -> list[GTCPassComposite]
find(self, product_type: type[CohPassComposite], time_range: TimeRange | None = None, filters: Unpack[_CohPassCompositeFilters] = {}) -> list[CohPassComposite]

Each product type is filtered on the fields its file name records; a filter left out, or None, matches every value:

SLC file names record the relative orbit and the pass direction, so the products created from a burst found here are named without the burst-ID map. They do not record the footprint: a burst found here takes its geometry from the burst-ID map, which is loaded once per process; see s1bursts.burstmap.download_burst_map.

Parameters
product_type: type[T]

The type of product to find, such as RTCProduct, or Burst for SLC bursts.

time_range: TimeRange | None = None

Keep only the products acquired on a day of this window: by date for SLC bursts, RTC and GTC products, by date_primary for coherence products, as BurstPairCollection.starting_within() selects pairs. A composite is kept when its whole period, start_date through end_date, lies in the window.

**filters: Any
Field values to match, from the list above: rel_orbit_number (1-175), burst_id and baseline_days are integers, sub_swath a SubSwath such as "IW1", polarization a Polarization such as "VV", pass_direction "Ascending" or "Descending", crs a WGS 84 / UTM zone as a CRS object, authority string or EPSG code, and mgrs_code an upper-case MGRS reference such as "32TMT64".
Returns
list
The products found, sorted by file name. A composite is returned once per identity, holding the layers whose files were found, so it may lack its measurement when only QA layers are on disk.
Raises
ValueError

If product_type is not a type FileManager names, or a filter is not one of the type’s or has a value no file name holds.

TypeError
If time_range is not a TimeRange.
Warns
UserWarning
Once per FileManager and directory, when finding SLC bursts or per-burst products, if slc_dir or prod_dir holds files named without their pass direction, which are not found.

get_product_path()

Get the path of a product’s file, or of one layer of a composite.

Usage

Source

get_product_path(product, layer=Layer.MEASUREMENT)
Parameters
product: BaseProduct

Product to get the path for.

layer: Layer | str = Layer.MEASUREMENT
The layer whose file to locate, as a Layer or its value such as "QA-NUM"; the measurement by default. The path of a composite’s layer does not depend on whether product.layers holds it. A per-burst product has only its measurement.
Returns
Path
Path to the file.
Raises
ValueError
If the product type is unknown, or it has no layer layer.

get_slc_path()

Generate the file path for an SLC burst.

Usage

Source

get_slc_path(burst)
Parameters
burst: Burst
Burst to generate the path for.
Returns
Path
Expected path to the SLC zip file.
Warns
UserWarning
Once per FileManager, if slc_dir holds SLC bursts named without their pass direction, which BurstCatalog.download() downloads again under the new name.

open()

Open a product’s file, or one layer of a composite, for reading.

Usage

Source

open(product, layer=Layer.MEASUREMENT)

Works for both local and S3 storage: an S3 UPath hands its endpoint and credentials to GDAL, which rasterio.open() does not.

Parameters
product: BaseProduct

Product to open.

layer: Layer | str = Layer.MEASUREMENT
The layer whose file to open, as a Layer or its value such as "QA-NUM"; the measurement by default.
Returns
AbstractContextManager[DatasetReader]
Context manager yielding the read-only rasterio dataset.
Raises
ValueError
If the product type is unknown, or it has no layer layer.

product_exists()

Check if a product’s file exists (works for both local and S3).

Usage

Source

product_exists(product, layer=Layer.MEASUREMENT)
Parameters
product: BaseProduct

Product to check.

layer: Layer | str = Layer.MEASUREMENT
The layer whose file to check, as a Layer or its value; the measurement by default.
Returns
bool
True if the file exists.
Raises
ValueError
If the product type is unknown, or it has no layer layer.