Observation-level manifest builder for MarsCLIP-style pretraining.
This module provides the first implementation slice for a tri-modal Mars
pretraining pipeline:
- one record per HiRISE observation,
- one preferred COLOR image path per record,
- observation text from
RATIONALE_DESC, and
- geospatial / viewing metadata derived from the cumulative index.
The builder intentionally operates at observation level rather than tile level,
because HiRISE text metadata are observation-level descriptions.
build_observation_manifest_from_index
build_observation_manifest_from_index(df: DataFrame, root: Path | str, *, bbox: tuple[float, float, float, float] | None = None, require_local_image: bool = True, prefer_cog: bool = True) -> pd.DataFrame
Build one COLOR observation record per observation from an index table.
Source code in src/clip/observation_manifest.py
| def build_observation_manifest_from_index(
df: pd.DataFrame,
root: pathlib.Path | str,
*,
bbox: tuple[float, float, float, float] | None = None,
require_local_image: bool = True,
prefer_cog: bool = True,
) -> pd.DataFrame:
"""Build one COLOR observation record per observation from an index table."""
root_path = pathlib.Path(root)
working = df.copy()
working["PRODUCT_ID"] = working["PRODUCT_ID"].astype(str).str.strip()
working["OBSERVATION_ID"] = working["OBSERVATION_ID"].astype(str).str.strip()
working["_product_type"] = working["PRODUCT_ID"].str.extract(
_PRODUCT_RE, expand=False
)
working = working[working["_product_type"] == "COLOR"].copy()
working = _apply_bbox_filter(working, bbox)
working = working.drop_duplicates("OBSERVATION_ID", keep="first").copy()
records: list[dict[str, object]] = []
for _, row in working.iterrows():
try:
record = _row_to_record(row, root_path, prefer_cog=prefer_cog)
except ValueError:
logger.warning(
"Observation %s straddles antimeridian. Skipping.",
row["OBSERVATION_ID"],
)
continue
if require_local_image and not record["has_local_image"]:
continue
records.append(record)
manifest = pd.DataFrame.from_records(records)
if manifest.empty:
return manifest
# Keep optional local-path fields as Python ``None`` rather than pandas
# NaN so downstream dataset code can use simple ``is None`` checks.
nullable_object_cols = [
"image_path",
"image_format",
"color_jp2_path",
"color_tif_path",
]
for col in nullable_object_cols:
manifest[col] = manifest[col].astype(object)
manifest[col] = manifest[col].where(pd.notna(manifest[col]), None)
manifest = manifest.sort_values("obs_id").reset_index(drop=True)
return manifest
|
build_observation_manifest
build_observation_manifest(root: Path | str, *, bbox: tuple[float, float, float, float] | None = None, require_local_image: bool = True, prefer_cog: bool = True) -> pd.DataFrame
Load the HiRISE cumulative index from disk and build a COLOR manifest.
Source code in src/clip/observation_manifest.py
| def build_observation_manifest(
root: pathlib.Path | str,
*,
bbox: tuple[float, float, float, float] | None = None,
require_local_image: bool = True,
prefer_cog: bool = True,
) -> pd.DataFrame:
"""Load the HiRISE cumulative index from disk and build a COLOR manifest."""
root_path = pathlib.Path(root)
lbl_path = root_path / f"{_INDEX_STEM}.LBL"
data = pdr.read(str(lbl_path))
data.load("all")
index_df: pd.DataFrame = data["RDR_INDEX_TABLE"]
return build_observation_manifest_from_index(
index_df,
root=root_path,
bbox=bbox,
require_local_image=require_local_image,
prefer_cog=prefer_cog,
)
|