Week 7 Interactive Session: Integration with Existing Models

Working with Prithvi, SatMAE, and other foundation models

Course Roadmap Mapping

This week’s work in the broader GFM plan.

Week Stage Focus You will build (geogfm) Library tools Outcome
7 Stage 2: Train Foundation Model Integration w/ Pretrained (light) core/registry.py; interoperability/huggingface.py stubs huggingface_hub, transformers optional Show mapping to Prithvi structure; plan switch

Weekly goals

  • Create a light registry for model variants
  • Add basic HF Hub glue to load model configs/weights
  • Compare MVP structure to Prithvi and define switch points

Session Outline (and Tangled Code)

  • Concepts β†’ Components mapping
    • Named factories for models/heads β†’ core/registry.py
    • External model interop hooks β†’ interoperability/huggingface.py

Package inits

# geogfm.interoperability
# registry will be added here

1) Lightweight Registry

from __future__ import annotations
from typing import Callable, Dict, Any

class Registry:
    """Minimal name β†’ builder registry for models/heads."""
    def __init__(self) -> None:
        self._fns: Dict[str, Callable[..., Any]] = {}

    def register(self, name: str) -> Callable[[Callable[..., Any]], Callable[..., Any]]:
        def wrapper(fn: Callable[..., Any]) -> Callable[..., Any]:
            key = name.lower()
            if key in self._fns:
                raise KeyError(f"Duplicate registration: {key}")
            self._fns[key] = fn
            return fn
        return wrapper

    def build(self, name: str, *args, **kwargs):
        key = name.lower()
        if key not in self._fns:
            raise KeyError(f"Unknown name: {name}")
        return self._fns[key](*args, **kwargs)

MODEL_REGISTRY = Registry()
HEAD_REGISTRY = Registry()

2) HuggingFace Interoperability Stubs

from __future__ import annotations
from typing import Any, Dict

try:
    from huggingface_hub import hf_hub_download  # optional
except Exception:  # pragma: no cover
    hf_hub_download = None  # type: ignore


def ensure_hf_available() -> None:
    if hf_hub_download is None:
        raise ImportError("huggingface_hub is not installed in this environment")


def download_config(repo_id: str, filename: str = "config.json") -> str:
    """Download a config file from HF Hub and return local path."""
    ensure_hf_available()
    return hf_hub_download(repo_id, filename)


def download_weights(repo_id: str, filename: str = "pytorch_model.bin") -> str:
    ensure_hf_available()
    return hf_hub_download(repo_id, filename)


def load_external_model(repo_id: str, config_loader) -> Dict[str, Any]:
    """Outline for loading external model configs/weights.
    Returns a dict with paths for downstream loading.
    """
    cfg_path = download_config(repo_id)
    w_path = download_weights(repo_id)
    return {"config_path": cfg_path, "weights_path": w_path}