"""eomatch.generate_previews - generate and catalogue preview thumbnails for matchups"""
from __future__ import annotations
import argparse
import datetime as dt
import logging
import os
from typing import List, Optional
import matplotlib.pyplot as plt
import pystac
from eomatch.add_asset import generate_missing_assets
from eomatch.context import EOMatchContext
from eomatch.domain import _matchup_collection_id, _matchup_events_collection_id
from eomatch.mu_stac import MatchupCatalogue
from eomatch.preview import BuildMUPreview, build_event_geo_thumbnail, preview_event
log = logging.getLogger(__name__)
__author__ = "Sam Hunt <sam.hunt@npl.co.uk>"
__all__ = ["generate_previews"]
def _get_event_item(mu_cat: MatchupCatalogue, event) -> pystac.Item:
"""Return *event*'s STAC Item from the catalogue, or raise if not found."""
events_collection_id = _matchup_events_collection_id(event.collections, event.platforms)
events_col = mu_cat.catalog.get_child(events_collection_id)
item = next(events_col.get_items(event.stac_id), None) if events_col is not None else None
if item is None:
raise ValueError(f"Event item {event.stac_id!r} not found in collection {events_collection_id!r}")
return item
def _get_matchup_item(mu_cat: MatchupCatalogue, matchup) -> pystac.Item:
"""Return *matchup*'s STAC Item from the catalogue, or raise if not found."""
collection_id = _matchup_collection_id(
[p.collection for p in matchup.products],
[p.platform for p in matchup.products],
)
col = mu_cat.catalog.get_child(collection_id)
if col is None:
raise ValueError(f"Collection {collection_id!r} not found in catalogue")
item = next(col.get_items(matchup.stac_id), None)
if item is None:
raise ValueError(f"Matchup item {matchup.stac_id!r} not found in collection {collection_id!r}")
return item
def _event_asset_path(mu_cat: MatchupCatalogue, event, filename: str) -> str:
"""Local path to generate an *event*-level asset at, mirroring the ``data/`` date hierarchy."""
assert mu_cat.path is not None
events_collection_id = _matchup_events_collection_id(event.collections, event.platforms)
directory = os.path.join(
mu_cat.path,
"thumbnails",
events_collection_id,
event.start_time.strftime("%Y"),
event.start_time.strftime("%m"),
event.start_time.strftime("%d"),
)
return os.path.join(directory, f"{event.stac_id}_{filename}")
def _matchup_asset_path(mu_cat: MatchupCatalogue, matchup, filename: str) -> str:
"""Local path to generate a *matchup*-level asset at, mirroring the ``data/`` date hierarchy."""
assert mu_cat.path is not None
collection_id = _matchup_collection_id(
[p.collection for p in matchup.products],
[p.platform for p in matchup.products],
)
time_bounds = matchup.product_time_bounds
assert time_bounds is not None
start_time = time_bounds[0]
directory = os.path.join(
mu_cat.path,
"thumbnails",
collection_id,
start_time.strftime("%Y"),
start_time.strftime("%m"),
start_time.strftime("%d"),
)
return os.path.join(directory, f"{matchup.stac_id}_{filename}")
def _event_asset_href(mu_cat: MatchupCatalogue, event, path: str) -> str:
"""Href for an event asset at *path*, relative to the event item's own directory if resolvable."""
self_href = _get_event_item(mu_cat, event).get_self_href()
return os.path.relpath(path, os.path.dirname(self_href)) if self_href is not None else path
def _matchup_asset_href(mu_cat: MatchupCatalogue, matchup, path: str) -> str:
"""Href for a matchup asset at *path*, relative to the matchup item's own directory if resolvable."""
self_href = _get_matchup_item(mu_cat, matchup).get_self_href()
return os.path.relpath(path, os.path.dirname(self_href)) if self_href is not None else path
[docs]
def generate_previews(
catalogue_path: str,
collections: Optional[List[str]] = None,
platforms: Optional[List[str]] = None,
start_time: Optional[dt.datetime] = None,
stop_time: Optional[dt.datetime] = None,
bbox: Optional[List[float]] = None,
max_pixels: int = 512,
overwrite: bool = False,
generate_thumbnail: bool = True,
generate_overview: bool = True,
generate_event_thumbnail: bool = True,
generate_event_overview: bool = True,
context: Optional[EOMatchContext] = None,
) -> int:
"""Generate preview thumbnails for matchups and events in a STAC catalogue.
For each matchup item, optionally generates:
- A side-by-side PNG thumbnail (``"thumbnail"`` asset) showing imagery from
each sensor.
- A georeferenced RGBA GeoTIFF overview (``"overview"`` asset) that STAC
Browser renders on the slippy map.
For each matchup event item, optionally generates:
- A PNG thumbnail (``"thumbnail"`` asset) showing product footprints coloured
by collection.
- A georeferenced RGBA GeoTIFF overview (``"overview"`` asset) with
semi-transparent polygon footprints, suitable for overlay on the slippy map.
Only matchup assets for events whose source products have already been
downloaded are generated. Event assets are generated whenever a matchup set
is present. Assets that already exist are skipped unless ``overwrite=True``.
Each asset is saved to the ``thumbnails/`` tree inside the catalogue and the
item JSON is updated on disk immediately.
Intended as a periodic job: running it repeatedly on the same catalogue is
safe — already-generated items are left untouched.
Uses static ``"thumbnail"``/``"overview"`` keys rather than
:py:func:`~eomatch.add_asset.register_analysis`'s dated-snapshot +
``:latest`` convention: a thumbnail is a refreshed view of the same
matchup, not a new analysis result to keep a history of, so there is
nothing worth accumulating dated copies of — ``overwrite=True``
regenerates it in place instead.
Example usage::
from eomatch.generate_previews import generate_previews
n = generate_previews("/path/to/catalogue", max_pixels=256)
n = generate_previews("/path/to/catalogue", generate_overview=True)
:param catalogue_path: path to the root ``catalog.json`` of an existing
:py:class:`~eomatch.mu_stac.MatchupCatalogue`.
:param collections: restrict to one sensor pair,
e.g. ``["S2_MSI_L1C", "LANDSAT_C2L1"]``.
:param platforms: return only events that include at least one of these
platforms.
:param start_time: lower bound on event time (inclusive).
:param stop_time: upper bound on event time (inclusive).
:param bbox: spatial filter as ``[lon_min, lat_min, lon_max, lat_max]``.
:param max_pixels: maximum pixel extent in either spatial dimension of the
generated assets.
:param overwrite: if ``True``, regenerate assets even when they already
exist.
:param generate_thumbnail: if ``True`` (default), generate the side-by-side
PNG thumbnail for matchups and register it as a ``"thumbnail"`` asset.
:param generate_overview: if ``True`` (default), generate a georeferenced
RGBA GeoTIFF overview for matchups and register it as an ``"overview"``
asset. STAC Browser renders this asset on the slippy map.
:param generate_event_thumbnail: if ``True`` (default), generate a polygon
footprint PNG thumbnail for events and register it as a ``"thumbnail"``
asset.
:param generate_event_overview: if ``True`` (default), generate a
georeferenced RGBA GeoTIFF polygon overview for events and register it
as an ``"overview"`` asset.
:param context: optional :py:class:`~eomatch.context.EOMatchContext`
supplying the ``preview.read`` config; defaults to
``EOMatchContext()``.
:return: number of assets generated (all asset types combined).
"""
mu_cat = MatchupCatalogue.open(path=catalogue_path)
assert mu_cat.path is not None
events = mu_cat.get_events(
collections=collections,
platforms=platforms,
start_time=start_time,
stop_time=stop_time,
bbox=bbox,
products_downloaded=True,
)
builder = BuildMUPreview(context=context)
events_with_matchups = [event for event in events if event.matchup_set is not None]
matchups = [matchup for event in events_with_matchups for matchup in event.matchup_set]
count = 0
if generate_event_thumbnail:
def _make_event_thumbnail(event, path):
plt.close(preview_event(event, output_path=path))
count += generate_missing_assets(
events_with_matchups,
"thumbnail",
output_path=lambda event: _event_asset_path(mu_cat, event, "thumbnail.png"),
generate=_make_event_thumbnail,
add_asset=mu_cat.add_event_asset,
asset_href=lambda event, path: _event_asset_href(mu_cat, event, path),
media_type=pystac.MediaType.PNG,
title="Event footprint thumbnail",
roles=["thumbnail"],
overwrite=overwrite,
)
if generate_event_overview:
count += generate_missing_assets(
events_with_matchups,
"overview",
output_path=lambda event: _event_asset_path(mu_cat, event, "overview.tif"),
generate=lambda event, path: build_event_geo_thumbnail(event, path, max_pixels=max_pixels),
add_asset=mu_cat.add_event_asset,
asset_href=lambda event, path: _event_asset_href(mu_cat, event, path),
media_type="image/tiff; application=geotiff",
title="Event georeferenced overview",
roles=["overview"],
overwrite=overwrite,
)
if generate_thumbnail:
def _make_matchup_thumbnail(matchup, path):
plt.close(builder.run(matchup, output_path=path, max_pixels=max_pixels))
count += generate_missing_assets(
matchups,
"thumbnail",
output_path=lambda matchup: _matchup_asset_path(mu_cat, matchup, "thumbnail.png"),
generate=_make_matchup_thumbnail,
add_asset=mu_cat.add_matchup_asset,
asset_href=lambda matchup, path: _matchup_asset_href(mu_cat, matchup, path),
media_type=pystac.MediaType.PNG,
title="Preview thumbnail",
roles=["thumbnail"],
overwrite=overwrite,
)
if generate_overview:
count += generate_missing_assets(
matchups,
"overview",
output_path=lambda matchup: _matchup_asset_path(mu_cat, matchup, "overview.tif"),
generate=lambda matchup, path: builder.build_geo_thumbnail(matchup, path, max_pixels=max_pixels),
add_asset=mu_cat.add_matchup_asset,
asset_href=lambda matchup, path: _matchup_asset_href(mu_cat, matchup, path),
media_type="image/tiff; application=geotiff",
title="Georeferenced overview",
roles=["overview"],
overwrite=overwrite,
)
log.info("Generated %d thumbnail(s).", count)
return count
def main() -> None:
"""Entry point for the ``eomatch-preview`` console script."""
parser = argparse.ArgumentParser(
description=(
"Generate preview thumbnails for downloaded matchups in a STAC catalogue "
"and register them as thumbnail assets."
)
)
parser.add_argument(
"catalogue",
metavar="CATALOGUE",
help="Path to the root catalog.json of the matchup catalogue.",
)
parser.add_argument(
"--collections",
metavar="ID",
nargs="+",
help="Restrict to this sensor pair, e.g. S2_MSI_L1C LANDSAT_C2L1.",
)
parser.add_argument(
"--platforms",
metavar="NAME",
nargs="+",
help="Restrict to events that include at least one of these platforms.",
)
parser.add_argument(
"--start",
metavar="DATETIME",
help="Lower bound on event time (ISO-8601, e.g. 2022-01-01).",
)
parser.add_argument(
"--stop",
metavar="DATETIME",
help="Upper bound on event time (ISO-8601, e.g. 2022-12-31).",
)
parser.add_argument(
"--bbox",
metavar="FLOAT",
nargs=4,
type=float,
help="Spatial filter: lon_min lat_min lon_max lat_max.",
)
parser.add_argument(
"--max-pixels",
metavar="N",
type=int,
default=512,
help="Maximum pixel extent in either dimension (default: 512).",
)
parser.add_argument(
"--overwrite",
action="store_true",
help="Regenerate assets even when they already exist.",
)
parser.add_argument(
"--no-thumbnail",
action="store_true",
help="Skip generating the side-by-side PNG thumbnail.",
)
parser.add_argument(
"--no-overview",
action="store_true",
help="Skip generating the georeferenced RGBA GeoTIFF overview.",
)
parser.add_argument(
"--no-event-thumbnail",
action="store_true",
help="Skip generating the event footprint PNG thumbnail.",
)
parser.add_argument(
"--no-event-overview",
action="store_true",
help="Skip generating the event georeferenced RGBA GeoTIFF overview.",
)
parser.add_argument(
"--config",
metavar="PATH",
help="Path to a YAML config file (merged on top of the user config).",
)
parser.add_argument(
"--verbose",
"-v",
action="store_true",
help="Enable debug logging.",
)
args = parser.parse_args()
logging.basicConfig(
level=logging.DEBUG if args.verbose else logging.INFO,
format="%(asctime)s %(levelname)s %(message)s",
datefmt="%H:%M:%S",
)
start_time = dt.datetime.fromisoformat(args.start) if args.start else None
stop_time = dt.datetime.fromisoformat(args.stop) if args.stop else None
context = EOMatchContext(args.config) if args.config else EOMatchContext()
generate_previews(
catalogue_path=args.catalogue,
collections=args.collections,
platforms=args.platforms,
start_time=start_time,
stop_time=stop_time,
bbox=args.bbox,
max_pixels=args.max_pixels,
overwrite=args.overwrite,
generate_thumbnail=not args.no_thumbnail,
generate_overview=not args.no_overview,
generate_event_thumbnail=not args.no_event_thumbnail,
generate_event_overview=not args.no_event_overview,
context=context,
)
if __name__ == "__main__":
main()