rslearn.data_sources.utils

utils

Utilities shared by data sources.

MOSAIC_MIN_ITEM_COVERAGE module-attribute

MOSAIC_MIN_ITEM_COVERAGE = 0.1

Minimum fraction of area that item should cover when adding it to a mosaic group.

MOSAIC_REMAINDER_EPSILON module-attribute

MOSAIC_REMAINDER_EPSILON = 0.01

Fraction of original geometry area below which mosaic is considered to contain the entire geometry.

MatchedItemGroup dataclass

Bases: Generic[ItemType]

A matched item group carrying the request time range used to build it.

Source code in rslearn/data_sources/utils.py

@dataclass
class MatchedItemGroup(Generic[ItemType]):
    """A matched item group carrying the request time range used to build it."""

    items: list[ItemType]
    request_time_range: tuple[datetime, datetime] | None

PendingMosaic dataclass

A mosaic being created by match_candidate_items_to_window.

Parameters:

Name	Type	Description	Default
items	list	the list of items in the mosaic.	required
remainder	Polygon	the remainder of the geometry that is not covered by any of the items.	required
completed	bool	whether the mosaic is done (sufficient coverage of the geometry).	False

Source code in rslearn/data_sources/utils.py

@dataclass
class PendingMosaic:
    """A mosaic being created by match_candidate_items_to_window.

    Args:
        items: the list of items in the mosaic.
        remainder: the remainder of the geometry that is not covered by any of the
            items.
        completed: whether the mosaic is done (sufficient coverage of the geometry).
    """

    # Cannot use list[ItemType] here.
    items: list
    remainder: shapely.Polygon
    completed: bool = False

match_with_space_mode_contains

match_with_space_mode_contains(geometry: STGeometry, items: list[ItemType], item_shps: list[Geometry], query_config: QueryConfig) -> list[list[ItemType]]

Match items that fully contain the window geometry.

Parameters:

Name	Type	Description	Default
geometry	STGeometry	the window's geometry.	required
items	list[ItemType]	list of items.	required
item_shps	list[Geometry]	the item shapes projected to the window's projection.	required
query_config	QueryConfig	the query configuration.	required

Returns:

Type	Description
list[list[ItemType]]	list of matched item groups, where each group contains a single item.

Source code in rslearn/data_sources/utils.py

def match_with_space_mode_contains(
    geometry: STGeometry,
    items: list[ItemType],
    item_shps: list[shapely.Geometry],
    query_config: QueryConfig,
) -> list[list[ItemType]]:
    """Match items that fully contain the window geometry.

    Args:
        geometry: the window's geometry.
        items: list of items.
        item_shps: the item shapes projected to the window's projection.
        query_config: the query configuration.

    Returns:
        list of matched item groups, where each group contains a single item.
    """
    groups: list[list[ItemType]] = []
    for item, item_shp in zip(items, item_shps):
        if not item_shp.contains(geometry.shp):
            continue
        groups.append([item])
        if len(groups) >= query_config.max_matches:
            break
    return groups

match_with_space_mode_intersects

match_with_space_mode_intersects(geometry: STGeometry, items: list[ItemType], item_shps: list[Geometry], query_config: QueryConfig) -> list[list[ItemType]]

Match items that intersect any portion of the window geometry.

Parameters:

Name	Type	Description	Default
geometry	STGeometry	the window's geometry.	required
items	list[ItemType]	list of items.	required
item_shps	list[Geometry]	the item shapes projected to the window's projection.	required
query_config	QueryConfig	the query configuration.	required

Returns:

Type	Description
list[list[ItemType]]	list of matched item groups, where each group contains a single item.

Source code in rslearn/data_sources/utils.py

def match_with_space_mode_intersects(
    geometry: STGeometry,
    items: list[ItemType],
    item_shps: list[shapely.Geometry],
    query_config: QueryConfig,
) -> list[list[ItemType]]:
    """Match items that intersect any portion of the window geometry.

    Args:
        geometry: the window's geometry.
        items: list of items.
        item_shps: the item shapes projected to the window's projection.
        query_config: the query configuration.

    Returns:
        list of matched item groups, where each group contains a single item.
    """
    groups: list[list[ItemType]] = []
    for item, item_shp in zip(items, item_shps):
        if not shp_intersects(item_shp, geometry.shp):
            continue
        groups.append([item])
        if len(groups) >= query_config.max_matches:
            break
    return groups

match_with_space_mode_mosaic

match_with_space_mode_mosaic(geometry: STGeometry, items: list[ItemType], item_shps: list[Geometry], query_config: QueryConfig) -> list[list[ItemType]]

Match items into mosaic groups that cover the window geometry.

Creates groups of items that together cover the window geometry. The number of overlapping coverages in each group is controlled by mosaic_compositing_overlaps.

Parameters:

Name	Type	Description	Default
geometry	STGeometry	the window's geometry.	required
items	list[ItemType]	list of items.	required
item_shps	list[Geometry]	the item shapes projected to the window's projection.	required
query_config	QueryConfig	the query configuration.	required

Returns:

Type	Description
list[list[ItemType]]	list of matched item groups, where each group forms a mosaic covering the
list[list[ItemType]]	window.

Source code in rslearn/data_sources/utils.py

def match_with_space_mode_mosaic(
    geometry: STGeometry,
    items: list[ItemType],
    item_shps: list[shapely.Geometry],
    query_config: QueryConfig,
) -> list[list[ItemType]]:
    """Match items into mosaic groups that cover the window geometry.

    Creates groups of items that together cover the window geometry. The number of
    overlapping coverages in each group is controlled by mosaic_compositing_overlaps.

    Args:
        geometry: the window's geometry.
        items: list of items.
        item_shps: the item shapes projected to the window's projection.
        query_config: the query configuration.

    Returns:
        list of matched item groups, where each group forms a mosaic covering the
        window.
    """
    overlaps = query_config.mosaic_compositing_overlaps

    # Calculate how many single-coverage mosaics we need to create.
    # We need enough mosaics to consolidate into max_matches groups with the
    # desired number of overlaps per group.
    max_single_mosaics = query_config.max_matches * overlaps

    # Create single-coverage mosaics
    single_mosaics = _create_single_coverage_mosaics(
        geometry, items, item_shps, max_single_mosaics
    )

    # Consolidate into groups based on overlaps
    return _consolidate_mosaics_by_overlaps(
        single_mosaics, overlaps, query_config.max_matches
    )

match_with_space_mode_single_composite

match_with_space_mode_single_composite(geometry: STGeometry, items: list[ItemType], item_shps: list[Geometry], query_config: QueryConfig) -> list[list[ItemType]]

Match items for SINGLE_COMPOSITE.

All spatially-intersecting items go into a single group so that one composite can be created from all matching items.

Parameters:

Name	Type	Description	Default
geometry	STGeometry	the window's geometry.	required
items	list[ItemType]	list of items.	required
item_shps	list[Geometry]	the item shapes projected to the window's projection.	required
query_config	QueryConfig	the query configuration.	required

Returns:

Type	Description
list[list[ItemType]]	list containing a single item group of all spatially-intersecting items,
list[list[ItemType]]	or empty list if no items intersect.

Source code in rslearn/data_sources/utils.py

def match_with_space_mode_single_composite(
    geometry: STGeometry,
    items: list[ItemType],
    item_shps: list[shapely.Geometry],
    query_config: QueryConfig,
) -> list[list[ItemType]]:
    """Match items for SINGLE_COMPOSITE.

    All spatially-intersecting items go into a single group so that one composite can
    be created from all matching items.

    Args:
        geometry: the window's geometry.
        items: list of items.
        item_shps: the item shapes projected to the window's projection.
        query_config: the query configuration.

    Returns:
        list containing a single item group of all spatially-intersecting items,
        or empty list if no items intersect.
    """
    group_items: list[ItemType] = []
    for item, item_shp in zip(items, item_shps):
        if shp_intersects(item_shp, geometry.shp):
            group_items.append(item)
    return [group_items] if group_items else []

match_candidate_items_to_window

match_candidate_items_to_window(geometry: STGeometry, items: list[ItemType], query_config: QueryConfig) -> list[MatchedItemGroup[ItemType]]

Match candidate items to a window based on the query configuration.

If period_duration is set, the window time range is split into sub-periods and the handler is applied per-period with effective max_matches=1.

When period_duration is set and per_period_mosaic_reverse_time_order is True (the current default), the resulting groups are reversed so that the most recent period comes first. This default will change to False after 2026-04-01.

Parameters:

Name	Type	Description	Default
geometry	STGeometry	the window's geometry	required
items	list[ItemType]	all items from the data source that intersect spatially with the geometry	required
query_config	QueryConfig	the query configuration to use for matching	required

Returns:

Type	Description
list[MatchedItemGroup[ItemType]]	list of matched item groups.

Source code in rslearn/data_sources/utils.py

def match_candidate_items_to_window(
    geometry: STGeometry, items: list[ItemType], query_config: QueryConfig
) -> list[MatchedItemGroup[ItemType]]:
    """Match candidate items to a window based on the query configuration.

    If ``period_duration`` is set, the window time range is split into sub-periods
    and the handler is applied per-period with effective max_matches=1.

    When ``period_duration`` is set and ``per_period_mosaic_reverse_time_order``
    is True (the current default), the resulting groups are reversed so that the
    most recent period comes first. This default will change to False after
    2026-04-01.

    Args:
        geometry: the window's geometry
        items: all items from the data source that intersect spatially with the geometry
        query_config: the query configuration to use for matching

    Returns:
        list of matched item groups.
    """
    # PER_PERIOD_MOSAIC should default to 30-day periods in case period_duration is not
    # set, since period_duration previously applied only for PER_PERIOD_MOSAIC with
    # default 30 day duration.
    period_duration = query_config.period_duration
    if (
        query_config.space_mode == SpaceMode.PER_PERIOD_MOSAIC
        and period_duration is None
    ):
        period_duration = timedelta(days=30)

    # Filter items by time and project them into the geometry's projection.
    acceptable_items, acceptable_item_shps = _filter_and_project_items(
        geometry, items, query_config
    )

    handler = space_mode_handlers.get(query_config.space_mode)
    if handler is None:
        raise ValueError(f"invalid space mode {query_config.space_mode}")

    # Handle period_duration if set. This causes the space_mode_handler to be called
    # once for each period within the window time range. In this case max_matches
    # controls the number of periods, while each handler creates at most one item
    # group.
    if period_duration is not None:
        if geometry.time_range is None:
            raise ValueError("period_duration is set but geometry has no time_range")
        per_period_query_config = QueryConfig(
            space_mode=query_config.space_mode,
            max_matches=1,
            mosaic_compositing_overlaps=query_config.mosaic_compositing_overlaps,
        )

        # Iterate from most recent period backwards so that when max_matches
        # truncates, we keep the most recent periods.
        groups: list[MatchedItemGroup[ItemType]] = []
        period_end = geometry.time_range[1]
        while (
            period_end - period_duration >= geometry.time_range[0]
            and len(groups) < query_config.max_matches
        ):
            period_start = period_end - period_duration
            period_geom = STGeometry(
                geometry.projection, geometry.shp, (period_start, period_end)
            )
            period_end = period_start

            # Re-filter items to this period.
            period_items, period_shps = _filter_and_project_items(
                period_geom, items, per_period_query_config
            )
            period_groups = handler(
                period_geom, period_items, period_shps, per_period_query_config
            )
            if period_groups:
                groups.append(
                    MatchedItemGroup(
                        period_groups[0],
                        period_geom.time_range,
                    )
                )

        # Groups are in reverse chronological order. Reverse to chronological
        # unless the deprecated per_period_mosaic_reverse_time_order is True.
        if query_config.per_period_mosaic_reverse_time_order:
            warnings.warn(
                "QueryConfig.per_period_mosaic_reverse_time_order defaults to True, "
                "which returns item groups in reverse temporal order (most recent "
                "first) when period_duration is set. This default will change to "
                "False (chronological order) after 2026-04-01. To silence this "
                "warning, explicitly set per_period_mosaic_reverse_time_order=False.",
                FutureWarning,
                stacklevel=3,
            )
        else:
            groups.reverse()
    else:
        groups = [
            MatchedItemGroup(group, geometry.time_range)
            for group in handler(
                geometry, acceptable_items, acceptable_item_shps, query_config
            )
        ]

    # Enforce minimum matches if set.
    if len(groups) < query_config.min_matches:
        logger.warning(
            "Window rejected: found %d matches (required: %d) for time range %s",
            len(groups),
            query_config.min_matches,
            geometry.time_range if geometry.time_range else "unlimited",
        )
        return []

    return groups