hyperproc.registry

Source-derived reference

Generated from the current hyperproc 0.1.2 checkout.

Implementation: hyperproc/registry.py. Signatures, defaults, docstrings, and expandable source are extracted statically; the module is not imported or executed. Names beginning with _ are implementation details, not a stable public API.

Use the function signature as the authority for individual parameter defaults and return annotations. Original docstrings sometimes group parameter names or wrap return descriptions across lines; these descriptions are preserved rather than inferred or rewritten.

Which reader handles which (sensor, level), and how to guess from a filename.

One table, one place to edit. Adding a sensor means writing its reader module and flipping its loader entry here from None to the function.

REGISTRY: dict[tuple[str, str], Entry] = {} module-attribute

Entry dataclass

One supported (or planned) sensor/level combination.

Source code in hyperproc/registry.py

@dataclass(frozen=True)
class Entry:
    """One supported (or planned) sensor/level combination."""

    sensor: str
    level: str
    #: Called as ``loader(path, **kwargs) -> xr.Dataset``. None = not written yet.
    loader: Callable | None
    #: What the user should hand to :func:`hyperproc.open`.
    expects: str
    #: Regexes matched against the filename to auto-detect this entry.
    patterns: tuple[str, ...] = ()
    note: str = ""

    @property
    def implemented(self) -> bool:
        return self.loader is not None

    def matches(self, name: str) -> bool:
        return any(re.match(p, name, re.IGNORECASE) for p in self.patterns)

sensor: str instance-attribute

level: str instance-attribute

loader: Callable | None instance-attribute

expects: str instance-attribute

patterns: tuple[str, ...] = () class-attribute instance-attribute

note: str = '' class-attribute instance-attribute

implemented: bool property

__init__(sensor: str, level: str, loader: Callable | None, expects: str, patterns: tuple[str, ...] = (), note: str = '') -> None

matches(name: str) -> bool

Source code in hyperproc/registry.py

def matches(self, name: str) -> bool:
    return any(re.match(p, name, re.IGNORECASE) for p in self.patterns)

_tanager(path, **kw)

Source code in hyperproc/registry.py

def _tanager(path, **kw):
    from hyperproc.readers.tanager import open_tanager

    return open_tanager(path, **kw)

_pace(path, **kw)

Source code in hyperproc/registry.py

def _pace(path, **kw):
    from hyperproc.readers.pace import open_pace

    return open_pace(path, **kw)

_enmap(path, **kw)

Source code in hyperproc/registry.py

def _enmap(path, **kw):
    from hyperproc.readers.enmap import open_enmap

    return open_enmap(path, **kw)

_desis(path, **kw)

Source code in hyperproc/registry.py

def _desis(path, **kw):
    from hyperproc.readers.desis import open_desis

    return open_desis(path, **kw)

_neon(path, **kw)

Source code in hyperproc/registry.py

def _neon(path, **kw):
    from hyperproc.readers.neon import open_neon

    return open_neon(path, **kw)

_aviris(path, **kw)

Source code in hyperproc/registry.py

def _aviris(path, **kw):
    from hyperproc.readers.aviris import open_aviris

    return open_aviris(path, **kw)

_prisma(path, **kw)

Source code in hyperproc/registry.py

def _prisma(path, **kw):
    from hyperproc.readers.prisma import open_prisma

    return open_prisma(path, **kw)

_emit(path, **kw)

Source code in hyperproc/registry.py

def _emit(path, **kw):
    # Imported lazily so a missing optional dependency for one sensor never
    # breaks `import hyperproc` for the others.
    from hyperproc.readers.emit import open_emit

    return open_emit(path, **kw)

register(entry: Entry) -> None

Source code in hyperproc/registry.py

def register(entry: Entry) -> None:
    REGISTRY[(entry.sensor.upper(), entry.level.upper())] = entry

sniff(path: str | Path) -> tuple[str, str] | None

Guess (sensor, level) from a filename. None if nothing matches.

Source code in hyperproc/registry.py

def sniff(path: str | Path) -> tuple[str, str] | None:
    """Guess ``(sensor, level)`` from a filename. None if nothing matches."""
    name = Path(path).name
    for key, entry in REGISTRY.items():
        if entry.matches(name):
            return key
    return None

resolve(path: str | Path, sensor: str | None, level: str | None) -> Entry

Find the Entry for an explicit sensor/level, or work it out from path.

Source code in hyperproc/registry.py

def resolve(path: str | Path, sensor: str | None, level: str | None) -> Entry:
    """Find the Entry for an explicit sensor/level, or work it out from ``path``."""
    if sensor is None or level is None:
        guess = sniff(path)
        if guess is None:
            raise ValueError(
                f"Cannot tell what {Path(path).name!r} is. Pass sensor= and level= "
                f"explicitly, e.g. hyperproc.open(path, sensor='EMIT', level='L2A'). "
                f"{summary()}"
            )
        sensor = sensor or guess[0]
        level = level or guess[1]

    key = (sensor.upper(), level.upper())
    if key not in REGISTRY:
        known = sorted({s for s, _ in REGISTRY})
        levels = sorted(lv for s, lv in REGISTRY if s == sensor.upper())
        if levels:
            raise ValueError(
                f"No reader for {sensor.upper()} {level.upper()}. "
                f"Known levels for {sensor.upper()}: {', '.join(levels)}"
            )
        raise ValueError(f"Unknown sensor {sensor!r}. Known: {', '.join(known)}")

    entry = REGISTRY[key]
    if not entry.implemented:
        raise NotImplementedError(
            f"{entry.sensor} {entry.level} is registered but its reader is not written yet.\n"
            f"  expects: {entry.expects}\n"
            + (f"  note:    {entry.note}\n" if entry.note else "")
            + f"  to add:  write hyperproc/readers/{entry.sensor.lower()}.py, then set "
              f"loader= in hyperproc/registry.py"
        )
    return entry

summary() -> str

Source code in hyperproc/registry.py

def summary() -> str:
    done = sorted(f"{s} {lv}" for (s, lv), e in REGISTRY.items() if e.implemented)
    todo = sorted({s for (s, _), e in REGISTRY.items() if not e.implemented})
    return f"Implemented: {', '.join(done)}. Planned: {', '.join(todo)}."

list_readers() -> None

Print the dispatch table.

Source code in hyperproc/registry.py

def list_readers() -> None:
    """Print the dispatch table."""
    print(f"\n  {'sensor':<8} {'level':<6} {'status':<14} expects")
    print("  " + "-" * 76)
    for (s, lv), e in sorted(REGISTRY.items()):
        status = "ready" if e.implemented else "not written"
        print(f"  {s:<8} {lv:<6} {status:<14} {e.expects}")
    print()