rslearn.utils

utils

rslearn utilities.

Feature

A GeoJSON-like feature that contains one vector geometry.

Source code in rslearn/utils/feature.py

class Feature:
    """A GeoJSON-like feature that contains one vector geometry."""

    def __init__(self, geometry: STGeometry, properties: dict[str, Any] = {}):
        """Initialize a new Feature.

        Args:
            geometry: the STGeometry
            properties: properties of the feature
        """
        self.geometry = geometry
        self.properties = properties

    def to_geojson(self) -> dict[str, Any]:
        """Returns a GeoJSON dict corresponding to this feature."""
        return {
            "type": "Feature",
            "properties": self.properties,
            "geometry": json.loads(shapely.to_geojson(self.geometry.shp)),
        }

    def to_projection(self, projection: Projection) -> "Feature":
        """Converts this Feature to the target projection.

        Args:
            projection: the target projection

        Returns:
            a new Feature in the target projection
        """
        return Feature(self.geometry.to_projection(projection), self.properties)

    @staticmethod
    def from_geojson(projection: Projection, d: dict[str, Any]) -> "Feature":
        """Construct a Feature from a GeoJSON encoding.

        Args:
            projection: the projection of the GeoJSON feature
            d: the GeoJSON feature dict

        Returns:
            a Feature representing the specified geometry
        """
        shp = shapely.geometry.shape(d["geometry"])
        return Feature(STGeometry(projection, shp, None), d.get("properties", {}))

to_geojson

to_geojson() -> dict[str, Any]

Returns a GeoJSON dict corresponding to this feature.

Source code in rslearn/utils/feature.py

def to_geojson(self) -> dict[str, Any]:
    """Returns a GeoJSON dict corresponding to this feature."""
    return {
        "type": "Feature",
        "properties": self.properties,
        "geometry": json.loads(shapely.to_geojson(self.geometry.shp)),
    }

to_projection

to_projection(projection: Projection) -> Feature

Converts this Feature to the target projection.

Parameters:

Name	Type	Description	Default
projection	Projection	the target projection	required

Returns:

Type	Description
Feature	a new Feature in the target projection

Source code in rslearn/utils/feature.py

def to_projection(self, projection: Projection) -> "Feature":
    """Converts this Feature to the target projection.

    Args:
        projection: the target projection

    Returns:
        a new Feature in the target projection
    """
    return Feature(self.geometry.to_projection(projection), self.properties)

from_geojson staticmethod

from_geojson(projection: Projection, d: dict[str, Any]) -> Feature

Construct a Feature from a GeoJSON encoding.

Parameters:

Name	Type	Description	Default
projection	Projection	the projection of the GeoJSON feature	required
d	dict[str, Any]	the GeoJSON feature dict	required

Returns:

Type	Description
Feature	a Feature representing the specified geometry

Source code in rslearn/utils/feature.py

@staticmethod
def from_geojson(projection: Projection, d: dict[str, Any]) -> "Feature":
    """Construct a Feature from a GeoJSON encoding.

    Args:
        projection: the projection of the GeoJSON feature
        d: the GeoJSON feature dict

    Returns:
        a Feature representing the specified geometry
    """
    shp = shapely.geometry.shape(d["geometry"])
    return Feature(STGeometry(projection, shp, None), d.get("properties", {}))

Projection

A projection specifies a CRS, x resolution, and y resolution.

The coordinate reference system (CRS) defines the meaning of the coordinates. The resolutions specify the pixels per projection unit, and are used to map pixel coordinates to CRS coordinates.

Source code in rslearn/utils/geometry.py

class Projection:
    """A projection specifies a CRS, x resolution, and y resolution.

    The coordinate reference system (CRS) defines the meaning of the coordinates. The
    resolutions specify the pixels per projection unit, and are used to map pixel
    coordinates to CRS coordinates.
    """

    def __init__(self, crs: CRS, x_resolution: float, y_resolution: float) -> None:
        """Initialize a new Projection.

        Args:
            crs: the CRS
            x_resolution: the x resolution
            y_resolution: the y resolution
        """
        self.crs = crs
        self.x_resolution = x_resolution
        self.y_resolution = y_resolution

    def __eq__(self, other: Any) -> bool:
        """Returns whether this projection is the same as the other projection."""
        if not isinstance(other, Projection):
            return False
        if self.crs != other.crs:
            return False
        if not is_same_resolution(self.x_resolution, other.x_resolution):
            return False
        if not is_same_resolution(self.y_resolution, other.y_resolution):
            return False
        return True

    def __repr__(self) -> str:
        """Returns a string representation of this projection."""
        return (
            f"Projection(crs={self.crs}, "
            + f"x_resolution={self.x_resolution}, "
            + f"y_resolution={self.y_resolution})"
        )

    def __str__(self) -> str:
        """Returns a human-readable string summary of this projection."""
        return f"{self.crs}_{self.x_resolution}_{self.y_resolution}"

    def __hash__(self) -> int:
        """Returns a hash of this projection."""
        return hash((self.crs, self.x_resolution, self.y_resolution))

    def serialize(self) -> dict:
        """Serializes the projection to a JSON-encodable dictionary."""
        return {
            "crs": self.crs.to_string(),
            "x_resolution": self.x_resolution,
            "y_resolution": self.y_resolution,
        }

    @staticmethod
    def deserialize(d: dict) -> "Projection":
        """Deserializes a projection from a JSON-decoded dictionary."""
        return Projection(
            crs=CRS.from_string(d["crs"]),
            x_resolution=d["x_resolution"],
            y_resolution=d["y_resolution"],
        )

serialize

serialize() -> dict

Serializes the projection to a JSON-encodable dictionary.

Source code in rslearn/utils/geometry.py

def serialize(self) -> dict:
    """Serializes the projection to a JSON-encodable dictionary."""
    return {
        "crs": self.crs.to_string(),
        "x_resolution": self.x_resolution,
        "y_resolution": self.y_resolution,
    }

deserialize staticmethod

deserialize(d: dict) -> Projection

Deserializes a projection from a JSON-decoded dictionary.

Source code in rslearn/utils/geometry.py

@staticmethod
def deserialize(d: dict) -> "Projection":
    """Deserializes a projection from a JSON-decoded dictionary."""
    return Projection(
        crs=CRS.from_string(d["crs"]),
        x_resolution=d["x_resolution"],
        y_resolution=d["y_resolution"],
    )

STGeometry

A spatiotemporal geometry.

Specifiec crs and resolution and corresponding shape in pixel coordinates. Also specifies an optional time range (time range is unlimited if unset).

Source code in rslearn/utils/geometry.py

class STGeometry:
    """A spatiotemporal geometry.

    Specifiec crs and resolution and corresponding shape in pixel coordinates. Also
    specifies an optional time range (time range is unlimited if unset).
    """

    def __init__(
        self,
        projection: Projection,
        shp: shapely.Geometry,
        time_range: tuple[datetime, datetime] | None,
    ):
        """Creates a new spatiotemporal geometry.

        Args:
            projection: the projection
            shp: the shape in pixel coordinates
            time_range: optional start and end time (default unlimited)
        """
        self.projection = projection
        self.shp = shp
        self.time_range = time_range

    def contains_time(self, time: datetime) -> bool:
        """Returns whether this box contains the time."""
        if self.time_range is None:
            return True
        return time >= self.time_range[0] and time < self.time_range[1]

    def distance_to_time(self, time: datetime) -> timedelta:
        """Returns the distance from this box to the specified time.

        Args:
            time: the time to compute distance from

        Returns:
            the distance, which is 0 if the box contains the time
        """
        if self.time_range is None:
            return timedelta()
        if time < self.time_range[0]:
            return self.time_range[0] - time
        if time > self.time_range[1]:
            return time - self.time_range[1]
        return timedelta()

    def distance_to_time_range(
        self, time_range: tuple[datetime, datetime] | None
    ) -> timedelta:
        """Returns the distance from this geometry to the specified time range.

        Args:
            time_range: the time range to compute distance from

        Returns:
            the distance, which is 0 if the time ranges intersect
        """
        if self.time_range is None or time_range is None:
            return timedelta()
        if time_range[1] < self.time_range[0]:
            return self.time_range[0] - time_range[1]
        if self.time_range[1] < time_range[0]:
            return time_range[0] - self.time_range[1]
        return timedelta()

    def intersects_time_range(
        self, time_range: tuple[datetime, datetime] | None
    ) -> bool:
        """Returns whether this geometry intersects the other time range."""
        if self.time_range is None or time_range is None:
            return True
        if self.time_range[1] <= time_range[0]:
            return False
        if time_range[1] <= self.time_range[0]:
            return False
        return True

    def is_global(self) -> bool:
        """Returns whether this geometry has global spatial coverage.

        Global coverage is indicated by a special geometry with WGS84 projection and
        corners at (-180, -90, 180, 90) (see get_global_geometry).
        """
        if self.projection != WGS84_PROJECTION:
            return False
        if self.shp != shapely.box(*WGS84_BOUNDS):
            return False
        return True

    def is_too_large(self) -> bool:
        """Returns whether this geometry's spatial coverage is too large.

        This means that it will likely have issues during re-projections and such.
        """
        wgs84_bounds = self.to_projection(WGS84_PROJECTION).shp.bounds
        if wgs84_bounds[2] - wgs84_bounds[0] > MAX_GEOMETRY_DEGREES:
            return True
        if wgs84_bounds[3] - wgs84_bounds[1] > MAX_GEOMETRY_DEGREES:
            return True
        return False

    def to_wgs84(self) -> "STGeometry":
        """Convert to WGS84 with antimeridian splitting handling.

        For geometries already in WGS84, this is a no-op.
        """
        if self.projection.crs == WGS84_PROJECTION.crs:
            if self.projection == WGS84_PROJECTION:
                return self
            return self.to_projection(WGS84_PROJECTION)
        wgs84 = self.to_projection(WGS84_PROJECTION)
        return split_at_antimeridian(wgs84)

    def intersects(self, other: "STGeometry") -> bool:
        """Returns whether this box intersects the other box."""
        # Check temporal.
        if not self.intersects_time_range(other.time_range):
            return False

        # Check spatial.
        if self.is_global() or other.is_global():
            # One of the geometries indicates global coverage.
            return True

        # If projection is the same, it is an easy check.
        if other.projection == self.projection:
            return shp_intersects(self.shp, other.shp)

        # OK the projections are different. If the CRS is the same, then rescaling is
        # pretty safe so we can just rescale and compare shapely geometries. Otherwise,
        # we check the intersection in WGS84 so we can be sure to handle the
        # anti-meridian properly. Re-projection is required anyway so it shouldn't
        # be any less reliable to re-project to WGS84 (versus re-projecting one
        # geometry to the other's projection), and this allows us to have uniform
        # anti-meridian handling.
        if other.projection.crs == self.projection.crs:
            return shp_intersects(self.shp, other.to_projection(self.projection).shp)
        else:
            return shp_intersects(self.to_wgs84().shp, other.to_wgs84().shp)

    def to_projection(self, projection: Projection) -> "STGeometry":
        """Transforms this geometry to the specified projection.

        Note that this does not handle antimeridian splitting. Use to_wgs84 when
        re-projecting to WGS84 to get antimeridian splitting handling.
        """

        def apply_resolution(
            array: np.ndarray,
            x_resolution: float,
            y_resolution: float,
            forward: bool = True,
        ) -> np.ndarray:
            if forward:
                return np.stack(
                    [array[:, 0] / x_resolution, array[:, 1] / y_resolution], axis=1
                )
            else:
                return np.stack(
                    [array[:, 0] * x_resolution, array[:, 1] * y_resolution], axis=1
                )

        # Undo resolution.
        shp = shapely.transform(
            self.shp,
            lambda array: apply_resolution(
                array,
                self.projection.x_resolution,
                self.projection.y_resolution,
                forward=False,
            ),
        )
        # Change crs.
        # We only apply transform_geom if the CRS doesn't match, because even if we
        # call transform_geom with the same source and destination CRS, it takes
        # several milliseconds.
        if self.projection.crs != projection.crs:
            shp = rasterio.warp.transform_geom(self.projection.crs, projection.crs, shp)
            shp = shapely.geometry.shape(shp)

            # Sometimes the geometry becomes invalid after re-projection, e.g. if there
            # were near-duplicate vertices that crossed over or became the same after
            # transform_geom. So we apply make_valid to fix simple problems.
            if not shp.is_valid:
                shp = shapely.make_valid(shp)

        # Apply new resolution.
        shp = shapely.transform(
            shp,
            lambda array: apply_resolution(
                array, projection.x_resolution, projection.y_resolution, forward=True
            ),
        )

        return STGeometry(projection, shp, self.time_range)

    def __repr__(self) -> str:
        """Returns a string representation of this STGeometry."""
        return (
            f"STGeometry(projection={self.projection}, shp={self.shp}, "
            + f"time_range={self.time_range})"
        )

    def serialize(self) -> dict:
        """Serializes the geometry to a JSON-encodable dictionary."""
        return {
            "projection": self.projection.serialize(),
            "shp": self.shp.wkt,
            "time_range": (
                [self.time_range[0].isoformat(), self.time_range[1].isoformat()]
                if self.time_range
                else None
            ),
        }

    @staticmethod
    def deserialize(d: dict) -> "STGeometry":
        """Deserializes a geometry from a JSON-decoded dictionary."""
        return STGeometry(
            projection=Projection.deserialize(d["projection"]),
            shp=shapely.wkt.loads(d["shp"]),
            time_range=(
                (
                    datetime.fromisoformat(d["time_range"][0]),
                    datetime.fromisoformat(d["time_range"][1]),
                )
                if d["time_range"]
                else None
            ),
        )

contains_time

contains_time(time: datetime) -> bool

Returns whether this box contains the time.

Source code in rslearn/utils/geometry.py

def contains_time(self, time: datetime) -> bool:
    """Returns whether this box contains the time."""
    if self.time_range is None:
        return True
    return time >= self.time_range[0] and time < self.time_range[1]

distance_to_time

distance_to_time(time: datetime) -> timedelta

Returns the distance from this box to the specified time.

Parameters:

Name	Type	Description	Default
time	datetime	the time to compute distance from	required

Returns:

Type	Description
timedelta	the distance, which is 0 if the box contains the time

Source code in rslearn/utils/geometry.py

def distance_to_time(self, time: datetime) -> timedelta:
    """Returns the distance from this box to the specified time.

    Args:
        time: the time to compute distance from

    Returns:
        the distance, which is 0 if the box contains the time
    """
    if self.time_range is None:
        return timedelta()
    if time < self.time_range[0]:
        return self.time_range[0] - time
    if time > self.time_range[1]:
        return time - self.time_range[1]
    return timedelta()

distance_to_time_range

distance_to_time_range(time_range: tuple[datetime, datetime] | None) -> timedelta

Returns the distance from this geometry to the specified time range.

Parameters:

Name	Type	Description	Default
time_range	tuple[datetime, datetime] | None	the time range to compute distance from	required

Returns:

Type	Description
timedelta	the distance, which is 0 if the time ranges intersect

Source code in rslearn/utils/geometry.py

def distance_to_time_range(
    self, time_range: tuple[datetime, datetime] | None
) -> timedelta:
    """Returns the distance from this geometry to the specified time range.

    Args:
        time_range: the time range to compute distance from

    Returns:
        the distance, which is 0 if the time ranges intersect
    """
    if self.time_range is None or time_range is None:
        return timedelta()
    if time_range[1] < self.time_range[0]:
        return self.time_range[0] - time_range[1]
    if self.time_range[1] < time_range[0]:
        return time_range[0] - self.time_range[1]
    return timedelta()

intersects_time_range

intersects_time_range(time_range: tuple[datetime, datetime] | None) -> bool

Returns whether this geometry intersects the other time range.

Source code in rslearn/utils/geometry.py

def intersects_time_range(
    self, time_range: tuple[datetime, datetime] | None
) -> bool:
    """Returns whether this geometry intersects the other time range."""
    if self.time_range is None or time_range is None:
        return True
    if self.time_range[1] <= time_range[0]:
        return False
    if time_range[1] <= self.time_range[0]:
        return False
    return True

is_global

is_global() -> bool

Returns whether this geometry has global spatial coverage.

Global coverage is indicated by a special geometry with WGS84 projection and corners at (-180, -90, 180, 90) (see get_global_geometry).

Source code in rslearn/utils/geometry.py

def is_global(self) -> bool:
    """Returns whether this geometry has global spatial coverage.

    Global coverage is indicated by a special geometry with WGS84 projection and
    corners at (-180, -90, 180, 90) (see get_global_geometry).
    """
    if self.projection != WGS84_PROJECTION:
        return False
    if self.shp != shapely.box(*WGS84_BOUNDS):
        return False
    return True

is_too_large

is_too_large() -> bool

Returns whether this geometry's spatial coverage is too large.

This means that it will likely have issues during re-projections and such.

Source code in rslearn/utils/geometry.py

def is_too_large(self) -> bool:
    """Returns whether this geometry's spatial coverage is too large.

    This means that it will likely have issues during re-projections and such.
    """
    wgs84_bounds = self.to_projection(WGS84_PROJECTION).shp.bounds
    if wgs84_bounds[2] - wgs84_bounds[0] > MAX_GEOMETRY_DEGREES:
        return True
    if wgs84_bounds[3] - wgs84_bounds[1] > MAX_GEOMETRY_DEGREES:
        return True
    return False

to_wgs84

to_wgs84() -> STGeometry

Convert to WGS84 with antimeridian splitting handling.

For geometries already in WGS84, this is a no-op.

Source code in rslearn/utils/geometry.py

def to_wgs84(self) -> "STGeometry":
    """Convert to WGS84 with antimeridian splitting handling.

    For geometries already in WGS84, this is a no-op.
    """
    if self.projection.crs == WGS84_PROJECTION.crs:
        if self.projection == WGS84_PROJECTION:
            return self
        return self.to_projection(WGS84_PROJECTION)
    wgs84 = self.to_projection(WGS84_PROJECTION)
    return split_at_antimeridian(wgs84)

intersects

intersects(other: STGeometry) -> bool

Returns whether this box intersects the other box.

Source code in rslearn/utils/geometry.py

def intersects(self, other: "STGeometry") -> bool:
    """Returns whether this box intersects the other box."""
    # Check temporal.
    if not self.intersects_time_range(other.time_range):
        return False

    # Check spatial.
    if self.is_global() or other.is_global():
        # One of the geometries indicates global coverage.
        return True

    # If projection is the same, it is an easy check.
    if other.projection == self.projection:
        return shp_intersects(self.shp, other.shp)

    # OK the projections are different. If the CRS is the same, then rescaling is
    # pretty safe so we can just rescale and compare shapely geometries. Otherwise,
    # we check the intersection in WGS84 so we can be sure to handle the
    # anti-meridian properly. Re-projection is required anyway so it shouldn't
    # be any less reliable to re-project to WGS84 (versus re-projecting one
    # geometry to the other's projection), and this allows us to have uniform
    # anti-meridian handling.
    if other.projection.crs == self.projection.crs:
        return shp_intersects(self.shp, other.to_projection(self.projection).shp)
    else:
        return shp_intersects(self.to_wgs84().shp, other.to_wgs84().shp)

to_projection

to_projection(projection: Projection) -> STGeometry

Transforms this geometry to the specified projection.

Note that this does not handle antimeridian splitting. Use to_wgs84 when re-projecting to WGS84 to get antimeridian splitting handling.

Source code in rslearn/utils/geometry.py

def to_projection(self, projection: Projection) -> "STGeometry":
    """Transforms this geometry to the specified projection.

    Note that this does not handle antimeridian splitting. Use to_wgs84 when
    re-projecting to WGS84 to get antimeridian splitting handling.
    """

    def apply_resolution(
        array: np.ndarray,
        x_resolution: float,
        y_resolution: float,
        forward: bool = True,
    ) -> np.ndarray:
        if forward:
            return np.stack(
                [array[:, 0] / x_resolution, array[:, 1] / y_resolution], axis=1
            )
        else:
            return np.stack(
                [array[:, 0] * x_resolution, array[:, 1] * y_resolution], axis=1
            )

    # Undo resolution.
    shp = shapely.transform(
        self.shp,
        lambda array: apply_resolution(
            array,
            self.projection.x_resolution,
            self.projection.y_resolution,
            forward=False,
        ),
    )
    # Change crs.
    # We only apply transform_geom if the CRS doesn't match, because even if we
    # call transform_geom with the same source and destination CRS, it takes
    # several milliseconds.
    if self.projection.crs != projection.crs:
        shp = rasterio.warp.transform_geom(self.projection.crs, projection.crs, shp)
        shp = shapely.geometry.shape(shp)

        # Sometimes the geometry becomes invalid after re-projection, e.g. if there
        # were near-duplicate vertices that crossed over or became the same after
        # transform_geom. So we apply make_valid to fix simple problems.
        if not shp.is_valid:
            shp = shapely.make_valid(shp)

    # Apply new resolution.
    shp = shapely.transform(
        shp,
        lambda array: apply_resolution(
            array, projection.x_resolution, projection.y_resolution, forward=True
        ),
    )

    return STGeometry(projection, shp, self.time_range)

serialize

serialize() -> dict

Serializes the geometry to a JSON-encodable dictionary.

Source code in rslearn/utils/geometry.py

def serialize(self) -> dict:
    """Serializes the geometry to a JSON-encodable dictionary."""
    return {
        "projection": self.projection.serialize(),
        "shp": self.shp.wkt,
        "time_range": (
            [self.time_range[0].isoformat(), self.time_range[1].isoformat()]
            if self.time_range
            else None
        ),
    }

deserialize staticmethod

deserialize(d: dict) -> STGeometry

Deserializes a geometry from a JSON-decoded dictionary.

Source code in rslearn/utils/geometry.py

@staticmethod
def deserialize(d: dict) -> "STGeometry":
    """Deserializes a geometry from a JSON-decoded dictionary."""
    return STGeometry(
        projection=Projection.deserialize(d["projection"]),
        shp=shapely.wkt.loads(d["shp"]),
        time_range=(
            (
                datetime.fromisoformat(d["time_range"][0]),
                datetime.fromisoformat(d["time_range"][1]),
            )
            if d["time_range"]
            else None
        ),
    )

GridIndex

Bases: SpatialIndex

An index of temporal geometries using a grid.

Each cell in the grid contains a list of geometries that intersect it.

Source code in rslearn/utils/grid_index.py

class GridIndex(SpatialIndex):
    """An index of temporal geometries using a grid.

    Each cell in the grid contains a list of geometries that intersect it.
    """

    def __init__(self, size: float) -> None:
        """Initialize a new GridIndex.

        Args:
            size: the size of the grid cells
        """
        self.size = size
        self.grid: dict = {}
        self.items: list = []

    def insert(self, box: tuple[float, float, float, float], data: Any) -> None:
        """Insert a box into the index.

        Args:
            box: the bounding box of this item (minx, miny, maxx, maxy)
            data: arbitrary object
        """
        item_idx = len(self.items)
        self.items.append(data)

        def f(cell: tuple[int, int]) -> None:
            if cell not in self.grid:
                self.grid[cell] = []
            self.grid[cell].append(item_idx)

        self._each_cell(box, f)

    def _each_cell(
        self,
        box: tuple[float, float, float, float],
        f: Callable[[tuple[int, int]], None],
    ) -> None:
        """Call f for each cell intersecting a box.

        Args:
            box: the box (minx, miny, maxx, maxy)
            f: function to call for each cell
        """
        for i in range(
            int(math.floor(box[0] / self.size)), int(math.floor(box[2] / self.size)) + 1
        ):
            for j in range(
                int(math.floor(box[1] / self.size)),
                int(math.floor(box[3] / self.size)) + 1,
            ):
                f((i, j))

    def query(self, box: tuple[float, float, float, float]) -> list[Any]:
        """Query the index for objects intersecting a box.

        Args:
            box: the bounding box query (minx, miny, maxx, maxy)

        Returns:
            a list of objects in the index intersecting the box
        """
        matches = set()

        def f(cell: tuple[int, int]) -> None:
            if cell not in self.grid:
                return
            for item_idx in self.grid[cell]:
                matches.add(item_idx)

        self._each_cell(box, f)
        return [self.items[item_idx] for item_idx in matches]

insert

insert(box: tuple[float, float, float, float], data: Any) -> None

Insert a box into the index.

Parameters:

Name	Type	Description	Default
box	tuple[float, float, float, float]	the bounding box of this item (minx, miny, maxx, maxy)	required
data	Any	arbitrary object	required

Source code in rslearn/utils/grid_index.py

def insert(self, box: tuple[float, float, float, float], data: Any) -> None:
    """Insert a box into the index.

    Args:
        box: the bounding box of this item (minx, miny, maxx, maxy)
        data: arbitrary object
    """
    item_idx = len(self.items)
    self.items.append(data)

    def f(cell: tuple[int, int]) -> None:
        if cell not in self.grid:
            self.grid[cell] = []
        self.grid[cell].append(item_idx)

    self._each_cell(box, f)

query

query(box: tuple[float, float, float, float]) -> list[Any]

Query the index for objects intersecting a box.

Parameters:

Name	Type	Description	Default
box	tuple[float, float, float, float]	the bounding box query (minx, miny, maxx, maxy)	required

Returns:

Type	Description
list[Any]	a list of objects in the index intersecting the box

Source code in rslearn/utils/grid_index.py

def query(self, box: tuple[float, float, float, float]) -> list[Any]:
    """Query the index for objects intersecting a box.

    Args:
        box: the bounding box query (minx, miny, maxx, maxy)

    Returns:
        a list of objects in the index intersecting the box
    """
    matches = set()

    def f(cell: tuple[int, int]) -> None:
        if cell not in self.grid:
            return
        for item_idx in self.grid[cell]:
            matches.add(item_idx)

    self._each_cell(box, f)
    return [self.items[item_idx] for item_idx in matches]

get_global_raster_bounds

get_global_raster_bounds(projection: Projection) -> PixelBounds

Get very large pixel bounds for a global raster in the given projection.

This is useful for data sources that cover the entire world and don't want to compute exact bounds in arbitrary projections (which can fail for projections like UTM that only cover part of the world).

Parameters:

Name	Type	Description	Default
projection	Projection	the projection to get bounds in.	required

Returns:

Type	Description
PixelBounds	Pixel bounds that will intersect with any reasonable window. We assume that the
PixelBounds	absolute value of CRS coordinates is at most 2^32, and adjust it based on the
PixelBounds	resolution in the Projection in case very fine-grained resolutions are used.

Source code in rslearn/utils/geometry.py

def get_global_raster_bounds(projection: Projection) -> PixelBounds:
    """Get very large pixel bounds for a global raster in the given projection.

    This is useful for data sources that cover the entire world and don't want to
    compute exact bounds in arbitrary projections (which can fail for projections
    like UTM that only cover part of the world).

    Args:
        projection: the projection to get bounds in.

    Returns:
        Pixel bounds that will intersect with any reasonable window. We assume that the
        absolute value of CRS coordinates is at most 2^32, and adjust it based on the
        resolution in the Projection in case very fine-grained resolutions are used.
    """
    crs_bound = 2**32
    pixel_bound_x = int(crs_bound / abs(projection.x_resolution))
    pixel_bound_y = int(crs_bound / abs(projection.y_resolution))
    return (-pixel_bound_x, -pixel_bound_y, pixel_bound_x, pixel_bound_y)

is_same_resolution

is_same_resolution(res1: float, res2: float) -> bool

Returns whether the two resolutions are the same.

Source code in rslearn/utils/geometry.py

def is_same_resolution(res1: float, res2: float) -> bool:
    """Returns whether the two resolutions are the same."""
    return (max(res1, res2) / min(res1, res2) - 1) < RESOLUTION_EPSILON

shp_intersects

shp_intersects(shp1: Geometry, shp2: Geometry) -> bool

Returns whether the two shapes intersect.

Tries shp.intersects but falls back to shp.intersection which can be more reliable.

Source code in rslearn/utils/geometry.py

def shp_intersects(shp1: shapely.Geometry, shp2: shapely.Geometry) -> bool:
    """Returns whether the two shapes intersect.

    Tries shp.intersects but falls back to shp.intersection which can be more
    reliable.
    """
    try:
        return shp1.intersects(shp2)
    except shapely.GEOSException:
        return shp1.intersection(shp2).area > 0

daterange

daterange(start_time: datetime, end_time: datetime) -> Generator[datetime, None, None]

Generator that yields each day between start_time and end_time.

Source code in rslearn/utils/time.py

def daterange(
    start_time: datetime, end_time: datetime
) -> Generator[datetime, None, None]:
    """Generator that yields each day between start_time and end_time."""
    for n in range(int((end_time - start_time).days)):
        yield start_time + timedelta(n)