Source code for astra.observatory_loader

import importlib.util
import logging
from pathlib import Path
from typing import List, Optional, Type

from astra import Config
from astra.observatory import Observatory




class ObservatoryLoader:
    """
    Load an Observatory subclass based on the observatory name.

    Examples:
    >>> from astra.observatory_loader import ObservatoryLoader
    >>> ObservatoryLoader(observatory_name="MyObservatory").load()
    """

    def __init__(self, observatory_name: Optional[str] = None):
        self.observatory_name = observatory_name
        self.custom_observatories = Config().paths.custom_observatories



    def load(self) -> Type[Observatory]:
        """Return an Observatory class: plugin-provided subclass if available, else default."""
        if self.observatory_name is None:
            return Observatory

        target = self.observatory_name.lower()

        for observatory_path in self.custom_observatories.glob("*.py"):
            classes = self._try_load_from_path(observatory_path)
            for cls in classes:
                if not isinstance(cls, type) or not issubclass(cls, Observatory):
                    continue

                # Match against class name and aliases
                candidates = {cls.__name__.lower()} | {
                    item.lower()
                    for item in getattr(cls, "OBSERVATORY_ALIASES", [])
                    if isinstance(item, str)
                }

                if target in candidates:
                    return cls

        return Observatory




    def _try_load_from_path(self, path: Path) -> List[Type[Observatory]]:
        """Attempt to load all Observatory subclasses from the specified path."""
        found: List[Type[Observatory]] = []
        try:
            spec = importlib.util.spec_from_file_location(path.stem, path)
            if spec and spec.loader:
                module = importlib.util.module_from_spec(spec)
                spec.loader.exec_module(module)
                for attr_name in dir(module):
                    attr = getattr(module, attr_name)
                    if (
                        isinstance(attr, type)
                        and issubclass(attr, Observatory)
                        and attr is not Observatory
                    ):
                        found.append(attr)
        except Exception as e:
            logging.error(f"Error loading module from {path}: {e}")
        return found