| """Search Hugging Face for marine feature datasets and write normalized manifests. |
| |
| The script does not download full datasets. It inspects dataset repository |
| metadata and file lists, then writes project-compatible manifests using hf:// |
| paths so large public datasets can be reviewed before any costly download. |
| """ |
|
|
| from __future__ import annotations |
|
|
| import argparse |
| import csv |
| import json |
| import os |
| import re |
| from dataclasses import asdict, dataclass |
| from datetime import datetime, timezone |
| from pathlib import Path |
| from typing import Iterable |
|
|
| from huggingface_hub import HfApi |
|
|
|
|
| IMAGE_EXTS = {".png", ".jpg", ".jpeg", ".tif", ".tiff", ".jp2", ".bmp", ".webp"} |
| METADATA_EXTS = {".csv", ".json", ".jsonl", ".parquet", ".txt"} |
| ARCHIVE_EXTS = {".zip", ".tar", ".gz", ".tgz", ".7z"} |
| TRACKED_EXTS = IMAGE_EXTS | METADATA_EXTS | ARCHIVE_EXTS |
| MASK_HINTS = ("mask", "label", "labels", "gt", "annotation", "annotations", "seg", "target") |
| TRAIN_SPLITS = ("train", "training", "val", "valid", "validation", "test") |
|
|
| QUERIES = [ |
| "seaweed", |
| "green tide", |
| "red tide", |
| "sargassum", |
| "aquaculture", |
| "ship", |
| "oil spill", |
| "sea ice", |
| "marine", |
| "ocean", |
| "remote sensing", |
| "sar ship", |
| "satellite imagery", |
| ] |
|
|
| ELEMENT_KEYWORDS = { |
| "green_tide": ("green_tide", "greentide", "green tide", "enteromorpha", "seaweed"), |
| "red_tide": ("red_tide", "redtide", "red tide", "harmful algal", "hab"), |
| "golden_tide": ("golden_tide", "goldentide", "sargassum", "sarg"), |
| "aquaculture": ("aquaculture", "oyster", "raft", "fish cage", "fish-cage"), |
| "ship": ("ship", "sar ship"), |
| "oil_spill": ("oilspill", "oil_spill", "oil spill", "oil-spill"), |
| "sea_ice": ("seaice", "sea_ice", "sea ice", "arctic ice"), |
| } |
|
|
| SHIP_CONTEXT_KEYWORDS = ( |
| "ship", |
| "sar", |
| "satellite", |
| "remote sensing", |
| "marine", |
| "ocean", |
| "sentinel", |
| "ais", |
| ) |
|
|
| EXCLUDE_KEYWORDS = ( |
| "medical", |
| "retinal", |
| "retina", |
| "miccai", |
| "flare", |
| "drive digital retinal", |
| "godot", |
| "shipping law", |
| "shipping orders", |
| "textual inversion", |
| "kantaicollection", |
| "waifu", |
| "anime", |
| "cancer", |
| "community health", |
| "plankton", |
| "mammal", |
| "legal", |
| ) |
|
|
| SATELLITE_PATTERN = re.compile(r"\b(GF\d+|HY\d+|Sentinel-?1|Sentinel-?2|Landsat-?\d*|SAR)\b", re.IGNORECASE) |
| PATCH_SIZE_PATTERN = re.compile(r"(?:^|[_/\-])(?:size)?(128|256|512|1024)(?:[_/\-]|$)") |
|
|
|
|
| @dataclass |
| class HfAssetRecord: |
| asset_id: str |
| repo_id: str |
| path: str |
| hf_path: str |
| filename: str |
| suffix: str |
| role: str |
| element: str |
| satellite: str | None |
| sensor: str | None |
| patch_size: int | None |
| source_project: str |
| source_dataset: str |
| size_bytes: int | None |
| downloads: int | None |
| likes: int | None |
| license: str | None |
| tags: list[str] |
| discovered_by: list[str] |
| quality_flags: list[str] |
|
|
|
|
| @dataclass |
| class HfSampleRecord: |
| sample_id: str |
| element: str |
| task_type: str |
| image_path: str |
| mask_path: str | None |
| label_encoding: dict[str, str] | None |
| satellite: str | None |
| sensor: str | None |
| resolution_m: float | None |
| patch_size: int | None |
| bands: list[str] | None |
| band_count: int | None |
| dtype: str | None |
| fusion: dict |
| acquired_at: str | None |
| source_project: str |
| source_dataset: str |
| split: str | None |
| quality_flags: list[str] |
| notes: str |
|
|
|
|
| def parse_args() -> argparse.Namespace: |
| parser = argparse.ArgumentParser(description=__doc__) |
| parser.add_argument("--output-root", required=True) |
| parser.add_argument("--author", default="cuibinge", help="HF namespace to always include.") |
| parser.add_argument("--limit-per-query", type=int, default=20) |
| parser.add_argument("--max-repos", type=int, default=80) |
| parser.add_argument("--max-files-per-repo", type=int, default=5000) |
| parser.add_argument("--include-low-confidence", action="store_true") |
| return parser.parse_args() |
|
|
|
|
| def normalize_text(value: str) -> str: |
| return value.replace("_", " ").replace("-", " ").replace("/", " ").lower() |
|
|
|
|
| def infer_element(*parts: str) -> str: |
| text = normalize_text(" ".join(part for part in parts if part)) |
| if "vessel" in text and all(keyword not in text for keyword in SHIP_CONTEXT_KEYWORDS): |
| return "unknown" |
| for element, keywords in ELEMENT_KEYWORDS.items(): |
| if any(keyword in text for keyword in keywords): |
| return element |
| if "vessel" in text and any(keyword in text for keyword in SHIP_CONTEXT_KEYWORDS): |
| return "ship" |
| return "unknown" |
|
|
|
|
| def infer_role(path: str) -> str: |
| lower = path.lower() |
| stem = Path(path).stem.lower() |
| suffix = Path(path).suffix.lower() |
| if suffix in ARCHIVE_EXTS: |
| return "archive" |
| if suffix in METADATA_EXTS: |
| if any(hint in lower or hint in stem for hint in MASK_HINTS) or stem in {"train", "val", "test"}: |
| return "annotation_table" |
| return "metadata" |
| if any(hint in lower or hint in stem for hint in MASK_HINTS): |
| return "mask" |
| return "image" |
|
|
|
|
| def infer_satellite(*parts: str) -> str | None: |
| match = SATELLITE_PATTERN.search(" ".join(parts)) |
| return match.group(1).upper().replace("-", "") if match else None |
|
|
|
|
| def infer_sensor(*parts: str) -> str | None: |
| upper = " ".join(parts).upper() |
| for sensor in ("PMS", "MUX", "MSS", "PAN", "WFV", "SAR", "MSI", "OLI"): |
| if sensor in upper: |
| return sensor |
| return None |
|
|
|
|
| def infer_patch_size(path: str) -> int | None: |
| match = PATCH_SIZE_PATTERN.search(path) |
| return int(match.group(1)) if match else None |
|
|
|
|
| def infer_split(path: str) -> str | None: |
| parts = {part.lower() for part in Path(path).parts} |
| for split in TRAIN_SPLITS: |
| if split in parts: |
| return "val" if split in {"valid", "validation"} else "train" if split == "training" else split |
| return None |
|
|
|
|
| def infer_license(tags: list[str]) -> str | None: |
| for tag in tags: |
| if tag.startswith("license:"): |
| return tag.split(":", 1)[1] |
| return None |
|
|
|
|
| def infer_fusion(repo_id: str, path: str, sensor: str | None) -> dict: |
| lower = f"{repo_id}/{path}".lower() |
| if "fuse" in lower or "fusion" in lower: |
| state = "fused_product" |
| method = "unknown_vendor_product" |
| persisted = True |
| elif sensor in {"SAR", "MSI", "OLI"}: |
| state = "none" |
| method = "none" |
| persisted = False |
| else: |
| state = "unknown" |
| method = "unknown" |
| persisted = False |
| return { |
| "state": state, |
| "method": method, |
| "sources": [{"role": "hf_dataset_file", "path": f"hf://datasets/{repo_id}/{path}", "resolution_m": None}], |
| "target_resolution_m": None, |
| "native_multispectral_resolution_m": None, |
| "persisted": persisted, |
| "reproducible": False, |
| "spectral_preservation": "unknown", |
| "notes": "Inferred from Hugging Face repository metadata; verify before training.", |
| } |
|
|
|
|
| def safe_id(value: str) -> str: |
| return re.sub(r"[^A-Za-z0-9]+", "_", value).strip("_").lower()[:180] |
|
|
|
|
| def dataset_search(api: HfApi, author: str, limit_per_query: int, max_repos: int) -> dict[str, dict]: |
| repos: dict[str, dict] = {} |
|
|
| def add_repo(dataset, reason: str) -> None: |
| entry = repos.setdefault(dataset.id, {"dataset": dataset, "reasons": []}) |
| if reason not in entry["reasons"]: |
| entry["reasons"].append(reason) |
|
|
| for dataset in api.list_datasets(author=author, full=True): |
| add_repo(dataset, f"author:{author}") |
|
|
| for query in QUERIES: |
| for dataset in api.list_datasets(search=query, limit=limit_per_query, full=True): |
| add_repo(dataset, f"query:{query}") |
| if len(repos) >= max_repos: |
| return repos |
| return repos |
|
|
|
|
| def relevant_repo(repo_id: str, tags: list[str], reasons: list[str]) -> bool: |
| text = normalize_text(" ".join([repo_id, " ".join(tags), " ".join(reasons)])) |
| if any(keyword in text for keyword in EXCLUDE_KEYWORDS): |
| return False |
| return any(keyword in text for keywords in ELEMENT_KEYWORDS.values() for keyword in keywords) or any( |
| token in text for token in ("marine", "ocean", "remote sensing", "satellite", "sar", "coast", "sea land") |
| ) |
|
|
|
|
| def iter_siblings(api: HfApi, repo_id: str, max_files: int) -> Iterable: |
| info = api.dataset_info(repo_id=repo_id, files_metadata=True) |
| for idx, sibling in enumerate(info.siblings or []): |
| if idx >= max_files: |
| break |
| yield sibling |
|
|
|
|
| def write_jsonl(path: Path, rows: Iterable[dict]) -> None: |
| with path.open("w", encoding="utf-8") as f: |
| for row in rows: |
| f.write(json.dumps(row, ensure_ascii=False) + "\n") |
|
|
|
|
| def main() -> None: |
| args = parse_args() |
| output_root = Path(args.output_root) |
| manifest_dir = output_root / "manifests" |
| report_dir = output_root / "reports" |
| manifest_dir.mkdir(parents=True, exist_ok=True) |
| report_dir.mkdir(parents=True, exist_ok=True) |
|
|
| api = HfApi(token=os.environ.get("HF_TOKEN")) |
| repos = dataset_search(api, args.author, args.limit_per_query, args.max_repos) |
|
|
| assets: list[HfAssetRecord] = [] |
| repo_rows: list[dict] = [] |
|
|
| for repo_id, entry in sorted(repos.items()): |
| dataset = entry["dataset"] |
| tags = list(getattr(dataset, "tags", []) or []) |
| reasons = entry["reasons"] |
| if not args.include_low_confidence and not relevant_repo(repo_id, tags, reasons): |
| continue |
|
|
| try: |
| siblings = list(iter_siblings(api, repo_id, args.max_files_per_repo)) |
| except Exception as exc: |
| repo_rows.append({"repo_id": repo_id, "status": "error", "error": str(exc), "reasons": ";".join(reasons)}) |
| continue |
|
|
| image_count = 0 |
| asset_count = 0 |
| for sibling in siblings: |
| rel_path = sibling.rfilename |
| suffix = Path(rel_path).suffix.lower() |
| if suffix not in TRACKED_EXTS: |
| continue |
| role = infer_role(rel_path) |
| if suffix in IMAGE_EXTS: |
| image_count += 1 |
| element = infer_element(repo_id, rel_path, " ".join(tags)) |
| sensor = infer_sensor(repo_id, rel_path, " ".join(tags)) |
| flags: list[str] = [] |
| if element == "unknown": |
| flags.append("unknown_element") |
| if role in {"mask", "annotation_table"}: |
| flags.append("mask_asset") |
| assets.append( |
| HfAssetRecord( |
| asset_id=safe_id(f"{repo_id}_{rel_path}"), |
| repo_id=repo_id, |
| path=rel_path, |
| hf_path=f"hf://datasets/{repo_id}/{rel_path}", |
| filename=Path(rel_path).name, |
| suffix=suffix, |
| role=role, |
| element=element, |
| satellite=infer_satellite(repo_id, rel_path, " ".join(tags)), |
| sensor=sensor, |
| patch_size=infer_patch_size(rel_path), |
| source_project=repo_id.split("/", 1)[-1], |
| source_dataset=repo_id, |
| size_bytes=getattr(sibling, "size", None), |
| downloads=getattr(dataset, "downloads", None), |
| likes=getattr(dataset, "likes", None), |
| license=infer_license(tags), |
| tags=tags, |
| discovered_by=reasons, |
| quality_flags=flags, |
| ) |
| ) |
| asset_count += 1 |
|
|
| repo_rows.append( |
| { |
| "repo_id": repo_id, |
| "status": "ok", |
| "reasons": ";".join(reasons), |
| "downloads": getattr(dataset, "downloads", None), |
| "likes": getattr(dataset, "likes", None), |
| "license": infer_license(tags), |
| "tags": ";".join(tags[:20]), |
| "files_scanned": len(siblings), |
| "image_assets": image_count, |
| "normalized_assets": asset_count, |
| } |
| ) |
|
|
| masks_by_key: dict[tuple[str, str], HfAssetRecord] = {} |
| for asset in assets: |
| if asset.role == "mask": |
| masks_by_key[(asset.repo_id, Path(asset.path).stem.lower())] = asset |
|
|
| samples: list[HfSampleRecord] = [] |
| for asset in assets: |
| if asset.role != "image": |
| continue |
| mask = masks_by_key.get((asset.repo_id, Path(asset.path).stem.lower())) |
| flags = list(asset.quality_flags) |
| if mask is None: |
| flags.append("unpaired_hf_image") |
| if asset.element == "unknown": |
| flags.append("needs_element_review") |
| samples.append( |
| HfSampleRecord( |
| sample_id=f"hf_{safe_id(asset.repo_id)}_{asset.asset_id}", |
| element=asset.element, |
| task_type="semantic_segmentation" if mask else "image_asset", |
| image_path=asset.hf_path, |
| mask_path=mask.hf_path if mask else None, |
| label_encoding={"0": "background", "1": asset.element} if mask and asset.element != "unknown" else None, |
| satellite=asset.satellite, |
| sensor=asset.sensor, |
| resolution_m=None, |
| patch_size=asset.patch_size, |
| bands=None, |
| band_count=None, |
| dtype=None, |
| fusion=infer_fusion(asset.repo_id, asset.path, asset.sensor), |
| acquired_at=None, |
| source_project=asset.source_project, |
| source_dataset=asset.source_dataset, |
| split=infer_split(asset.path), |
| quality_flags=flags, |
| notes="HF-discovered asset; inspect license, labels, georeferencing, and split before training.", |
| ) |
| ) |
|
|
| ready_samples = [ |
| sample |
| for sample in samples |
| if sample.element != "unknown" and sample.task_type == "semantic_segmentation" and sample.mask_path |
| ] |
| review_samples = [sample for sample in samples if sample not in ready_samples] |
|
|
| write_jsonl(manifest_dir / "hf_assets_raw.jsonl", (asdict(asset) for asset in assets)) |
| write_jsonl(manifest_dir / "samples.jsonl", (asdict(sample) for sample in samples)) |
| write_jsonl(manifest_dir / "samples_ready.jsonl", (asdict(sample) for sample in ready_samples)) |
| write_jsonl(manifest_dir / "samples_review.jsonl", (asdict(sample) for sample in review_samples)) |
|
|
| with (report_dir / "hf_dataset_inventory.csv").open("w", newline="", encoding="utf-8-sig") as f: |
| fieldnames = [ |
| "repo_id", |
| "status", |
| "reasons", |
| "downloads", |
| "likes", |
| "license", |
| "tags", |
| "files_scanned", |
| "image_assets", |
| "normalized_assets", |
| "error", |
| ] |
| writer = csv.DictWriter(f, fieldnames=fieldnames) |
| writer.writeheader() |
| for row in repo_rows: |
| writer.writerow({name: row.get(name, "") for name in fieldnames}) |
|
|
| summary = { |
| "created_at": datetime.now(timezone.utc).isoformat(), |
| "repo_count": len(repo_rows), |
| "asset_count": len(assets), |
| "sample_count": len(samples), |
| "ready_sample_count": len(ready_samples), |
| "review_sample_count": len(review_samples), |
| "by_element": {}, |
| "ready_by_element": {}, |
| "output_root": str(output_root), |
| "notes": [ |
| "hf:// paths are references; full dataset download is intentionally deferred.", |
| "Unknown elements and unpaired images require manual review before training.", |
| "No external coastline or land-mask vector is assumed.", |
| ], |
| } |
| for sample in samples: |
| summary["by_element"][sample.element] = summary["by_element"].get(sample.element, 0) + 1 |
| for sample in ready_samples: |
| summary["ready_by_element"][sample.element] = summary["ready_by_element"].get(sample.element, 0) + 1 |
| (report_dir / "hf_dataset_summary.json").write_text(json.dumps(summary, indent=2, ensure_ascii=False), encoding="utf-8") |
|
|
| md_lines = [ |
| "# Hugging Face Marine Dataset Discovery", |
| "", |
| f"- Created: {summary['created_at']}", |
| f"- Repositories reviewed: {summary['repo_count']}", |
| f"- Image assets indexed: {summary['asset_count']}", |
| f"- Standardized samples written: {summary['sample_count']}", |
| f"- Training-ready samples: {summary['ready_sample_count']}", |
| f"- Review samples: {summary['review_sample_count']}", |
| "", |
| "## Samples By Element", |
| "", |
| ] |
| for element, count in sorted(summary["by_element"].items()): |
| md_lines.append(f"- `{element}`: {count}") |
| md_lines.extend(["", "## Training-Ready Samples By Element", ""]) |
| for element, count in sorted(summary["ready_by_element"].items()): |
| md_lines.append(f"- `{element}`: {count}") |
| md_lines.extend( |
| [ |
| "", |
| "## Important Notes", |
| "", |
| "- Manifests use `hf://datasets/<repo>/<path>` references and do not imply files were downloaded.", |
| "- Licenses and label semantics must be checked before a dataset is used for training.", |
| "- Unknown or unpaired assets are kept for review, not treated as training-ready negatives.", |
| ] |
| ) |
| (report_dir / "hf_dataset_discovery.md").write_text("\n".join(md_lines) + "\n", encoding="utf-8") |
| print(json.dumps(summary, indent=2, ensure_ascii=False)) |
|
|
|
|
| if __name__ == "__main__": |
| main() |
|
|
