Skip to content

rslearn.data_sources.wrs2

wrs2

Utilities for working with Landsat WRS-2 path/row polygons.

get_wrs2_polygons 

get_wrs2_polygons(cache_dir: UPath, buffer_degrees: float = 0.2) -> list[tuple[Geometry, str, str]]

Get polygons for each (path, row) in the WRS2 grid.

Parameters:

Name Type Description Default
cache_dir UPath

local directory to cache downloaded shapefile artifacts.

 required
buffer_degrees float

amount to buffer each WRS polygon. Landsat scenes often extend slightly beyond nominal WRS polygons.

 0.2

Returns:

Type Description
list[tuple[Geometry, str, str]]

List of (polygon, zero-padded path, zero-padded row).
Source code in rslearn/data_sources/wrs2.py 
25
26
27
28
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
def get_wrs2_polygons(
    cache_dir: UPath,
    buffer_degrees: float = 0.2,
) -> list[tuple[shapely.Geometry, str, str]]:
    """Get polygons for each (path, row) in the WRS2 grid.

    Args:
        cache_dir: local directory to cache downloaded shapefile artifacts.
        buffer_degrees: amount to buffer each WRS polygon. Landsat scenes often
            extend slightly beyond nominal WRS polygons.

    Returns:
        List of (polygon, zero-padded path, zero-padded row).
    """
    prefix = "WRS2_descending"
    shp_fname = cache_dir / f"{prefix}.shp"
    if not shp_fname.exists():
        zip_fname = cache_dir / f"{prefix}.zip"
        logger.info("downloading WRS2 shapefile from %s", WRS2_URL)
        with urllib.request.urlopen(WRS2_URL) as response:
            with zip_fname.open("wb") as f:
                shutil.copyfileobj(response, f)

        with zip_fname.open("rb") as f:
            with zipfile.ZipFile(f, "r") as zipf:
                member_names = zipf.namelist()
                for ext in SHAPEFILE_AUX_EXTENSIONS:
                    cur_fname = prefix + ext
                    if cur_fname not in member_names:
                        continue
                    with zipf.open(cur_fname) as memberf:
                        with (cache_dir / (prefix + ext)).open("wb") as out_f:
                            shutil.copyfileobj(memberf, out_f)

                with zipf.open(f"{prefix}.shp") as memberf:
                    with shp_fname.open("wb") as out_f:
                        shutil.copyfileobj(memberf, out_f)

    aux_files = [cache_dir / (prefix + ext) for ext in SHAPEFILE_AUX_EXTENSIONS]
    with get_upath_local(shp_fname, extra_paths=aux_files) as local_fname:
        with fiona.open(local_fname) as src:
            polygons = []
            for feat in src:
                shp = shapely.geometry.shape(feat["geometry"])
                shp = shp.buffer(buffer_degrees)
                path = str(feat["properties"]["PATH"]).zfill(3)
                row = str(feat["properties"]["ROW"]).zfill(3)
                polygons.append((shp, path, row))
            return polygons

build_wrs2_grid_index 

build_wrs2_grid_index(cache_dir: UPath) -> GridIndex

Build a GridIndex over buffered WRS2 polygons.

Source code in rslearn/data_sources/wrs2.py 
76
77
78
79
80
81
def build_wrs2_grid_index(cache_dir: UPath) -> GridIndex:
    """Build a GridIndex over buffered WRS2 polygons."""
    grid_index = GridIndex(WRS2_GRID_SIZE)
    for polygon, path, row in get_wrs2_polygons(cache_dir):
        grid_index.insert(polygon.bounds, (polygon, path, row))
    return grid_index

get_pathrows_for_geometry 

get_pathrows_for_geometry(wrs2_index: GridIndex, wgs84_geometry: STGeometry) -> set[tuple[str, str]]

Get WRS2 path/row pairs intersecting a geometry in WGS84.

Source code in rslearn/data_sources/wrs2.py 
84
85
86
87
88
89
90
91
92
93
94
def get_pathrows_for_geometry(
    wrs2_index: GridIndex,
    wgs84_geometry: STGeometry,
) -> set[tuple[str, str]]:
    """Get WRS2 path/row pairs intersecting a geometry in WGS84."""
    pathrows = set()
    for shp in flatten_shape(wgs84_geometry.shp):
        for polygon, path, row in wrs2_index.query(shp.bounds):
            if wgs84_geometry.shp.intersects(polygon):
                pathrows.add((path, row))
    return pathrows