Container¶

class AppContainer:
    """Application container for MVP architecture."""

    __slots__ = (
        "_config",
        "_virtual_container",
        "_is_built",
        "_built_devices",
        "_devices_connected",
    )

    _device_components: ClassVar[dict[str, _DeviceComponent]] = {}
    _presenter_components: ClassVar[dict[str, _PresenterComponent]] = {}
    _view_components: ClassVar[dict[str, _ViewComponent]] = {}
    _config_path: ClassVar[Path | None] = None

    def __init_subclass__(
        cls,
        config: str | Path | None = None,
        **kwargs: Any,
    ) -> None:
        """Collect component wrappers from class attributes.

        Parameters
        ----------
        config : str | Path | None
            Path to a YAML configuration file for component kwargs.
        """
        super().__init_subclass__(**kwargs)

        if config is not None:
            cls._config_path = Path(config)

        devices: dict[str, _DeviceComponent] = {}
        presenters: dict[str, _PresenterComponent] = {}
        views: dict[str, _ViewComponent] = {}

        for base in cls.__bases__:
            if issubclass(base, AppContainer):
                devices.update(base._device_components)
                presenters.update(base._presenter_components)
                views.update(base._view_components)

        namespace = vars(cls)

        for attr_name, attr_value in namespace.items():
            if attr_name.startswith("_"):
                continue

            if isinstance(attr_value, _DeviceComponent):
                devices[attr_value.name] = attr_value
            elif isinstance(attr_value, _PresenterComponent):
                presenters[attr_value.name] = attr_value
            elif isinstance(attr_value, _ViewComponent):
                views[attr_value.name] = attr_value

        component_fields = {
            attr_name: value
            for attr_name, value in namespace.items()
            if not attr_name.startswith("_") and isinstance(value, _AnyField)
        }

        if component_fields:
            config_data: dict[str, Any] = {}
            if cls._config_path is not None:
                config_data = _load_yaml(cls._config_path)

            _section_key: dict[type, str] = {
                _DeviceField: "devices",
                _PresenterField: "presenters",
                _ViewField: "views",
            }

            for attr_name, field in component_fields.items():
                kw = field.kwargs
                if field.from_config is not None:
                    if not config_data:
                        raise TypeError(
                            f"Component field '{attr_name}' in {cls.__name__} has "
                            f"from_config set but no config path was "
                            f"provided to the container class"
                        )

                    section_key = _section_key[type(field)]
                    section_data: dict[str, Any] = config_data.get(section_key, {})
                    _sentinel = object()
                    cfg_section = section_data.get(field.from_config, _sentinel)

                    if cfg_section is _sentinel:
                        logger.warning(
                            f"No config section '{field.from_config}' found in "
                            f"'{section_key}' for component field '{attr_name}' in {cls.__name__}, "
                            f"using inline kwargs only"
                        )
                        kw = field.kwargs
                    else:
                        kw = {**(cfg_section or {}), **field.kwargs}

                comp_name = field.alias if field.alias is not None else attr_name

                wrapper: _DeviceComponent | _PresenterComponent | _ViewComponent
                if isinstance(field, _DeviceField):
                    wrapper = _DeviceComponent(field.cls, comp_name, **kw)
                    devices[comp_name] = wrapper
                elif isinstance(field, _PresenterField):
                    wrapper = _PresenterComponent(field.cls, comp_name, **kw)
                    presenters[comp_name] = wrapper
                else:
                    wrapper = _ViewComponent(field.cls, comp_name, **kw)
                    views[comp_name] = wrapper
                setattr(cls, attr_name, wrapper)

        cls._device_components = devices
        cls._presenter_components = presenters
        cls._view_components = views

        if devices or presenters or views:
            logger.debug(
                f"Collected from {cls.__name__}: "
                f"{len(devices)} devices, "
                f"{len(presenters)} presenters, "
                f"{len(views)} views"
            )

    def __init__(self, *, session: str = "Redsun", frontend: str = "pyqt") -> None:
        self._config: AppConfig = {
            "schema_version": 1.0,
            "session": session,
            "frontend": frontend,
        }
        self._virtual_container: VirtualContainer | None = None
        self._is_built: bool = False
        self._built_devices: dict[str, Device] = {}
        self._devices_connected: bool = False

        # In the declarative subclass path (class MyApp(QtAppContainer, config=...))
        # the metaclass loads the YAML only to resolve component kwargs and never
        # populates _config with top-level sections such as 'storage', 'session',
        # or 'schema_version'.  We read those here so that build() sees the same
        # state as the from_config() path, which sets them explicitly.
        config_path: Path | None = getattr(type(self), "_config_path", None)
        if config_path is not None:
            try:
                yaml_data = _load_yaml(config_path)
            except Exception as e:
                logger.warning(f"Could not read config file {config_path}: {e}")
                yaml_data = {}
            _COMPONENT_SECTIONS = frozenset({"devices", "presenters", "views"})
            for key, value in yaml_data.items():
                if key not in _COMPONENT_SECTIONS:
                    self._config[key] = value  # type: ignore[literal-required]

    @property
    def config(self) -> AppConfig:
        """Return the application configuration."""
        return self._config

    @property
    def devices(self) -> dict[str, Device]:
        """Return built device instances."""
        if not self._is_built:
            raise RuntimeError("Container not built. Call build() first.")
        return {name: comp.instance for name, comp in self._device_components.items()}

    @property
    def presenters(self) -> dict[str, Presenter]:
        """Return built presenter instances."""
        if not self._is_built:
            raise RuntimeError("Container not built. Call build() first.")
        return {
            name: comp.instance for name, comp in self._presenter_components.items()
        }

    @property
    def views(self) -> dict[str, View]:
        """Return built view instances."""
        if not self._is_built:
            raise RuntimeError("Container not built. Call build() first.")
        return {name: comp.instance for name, comp in self._view_components.items()}

    @property
    def virtual_container(self) -> VirtualContainer:
        """Return the virtual container instance."""
        if self._virtual_container is None:
            raise RuntimeError("Container not built. Call build() first.")
        return self._virtual_container

    @property
    def is_built(self) -> bool:
        """Return whether the container has been built."""
        return self._is_built

    def build(self) -> Self:
        """Instantiate all components in dependency order.

        Build order:

        1. VirtualContainer
        2. Devices
        3. Presenters (register their providers in the VirtualContainer)
        4. Views (inject dependencies from the VirtualContainer)
        """
        if self._is_built:
            logger.warning("Container already built, skipping rebuild")
            return self

        # ensure the background loop
        # is running
        _ = _loop_factory()

        logger.info("Building application container...")

        self._virtual_container = VirtualContainer()

        base_cfg: RedSunConfig = {
            "schema_version": self._config.get("schema_version", 1.0),
            "session": self._config.get("session", "Redsun"),
            "frontend": self._config.get("frontend", "pyqt"),
        }
        self._virtual_container._set_configuration(base_cfg)
        logger.debug("VirtualContainer created")

        # build devices
        built_devices: dict[str, Device] = {}
        for name, device_comp in self._device_components.items():
            try:
                built_devices[name] = device_comp.build()
                logger.debug(f"Device '{name}' built")
            except Exception as e:
                logger.error(f"Failed to build device '{name}': {e}")

        # build presenters
        for comp_name, presenter_component in self._presenter_components.items():
            try:
                presenter_component.build(built_devices)
            except Exception as e:
                logger.error(f"Failed to build presenter '{comp_name}': {e}")
                raise

        # build views
        for comp_name, view_component in self._view_components.items():
            try:
                view_component.build()
            except Exception as e:
                logger.error(f"Failed to build view '{comp_name}': {e}")
                raise

        # register providers from presenters and views
        all_components: dict[str, _PresenterComponent | _ViewComponent] = {
            **self._presenter_components,
            **self._view_components,
        }
        for comp_name, component in all_components.items():
            if isinstance(component.instance, IsProvider):
                component.instance.register_providers(self._virtual_container)

        # inject dependencies into presenters and views
        for comp_name, component in all_components.items():
            if isinstance(component.instance, IsInjectable):
                component.instance.inject_dependencies(self._virtual_container)

        self._built_devices = built_devices
        self._is_built = True
        logger.info(
            f"Container built: "
            f"{len(self._device_components)} devices, "
            f"{len(self._presenter_components)} presenters, "
            f"{len(self._view_components)} views"
        )

        return self

    def connect_devices(self, mock: bool = False) -> None:
        """Connect all devices via ophyd-async's async connect lifecycle.

        Call after [`build`][redsun.containers.container.AppContainer.build].
        Use ``mock=True`` in tests to skip hardware communication.

        Parameters
        ----------
        mock : bool
            If ``True``, connect using mock backends (no hardware required).

        Raises
        ------
        RuntimeError
            If called before [`build`][redsun.containers.container.AppContainer.build].
        """
        if not self._is_built:
            raise RuntimeError("Call build() before connect_devices()")

        async def _connect_all(mock: bool) -> None:
            await asyncio.gather(
                *[device.connect(mock=mock) for device in self._built_devices.values()]
            )

        run_coro(_connect_all(mock))
        self._devices_connected = True

    def shutdown(self) -> None:
        """Shutdown all presenters that implement ``HasShutdown``."""
        if not self._is_built:
            return

        for name, comp in self._presenter_components.items():
            if isinstance(comp.instance, HasShutdown):
                try:
                    comp.instance.shutdown()
                except Exception as e:
                    logger.error(f"Error shutting down presenter '{name}': {e}")

        self._is_built = False
        logger.info("Container shutdown complete")

    def run(self) -> None:
        """Build and connect devices if needed, then start the application."""
        if not self._is_built:
            self.build()
        if not self._devices_connected:
            self.connect_devices()

        frontend = self._config.get("frontend", "pyqt")
        logger.info(f"Starting application with frontend: {frontend}")

    @classmethod
    def from_config(cls, config_path: str) -> AppContainer:
        """Build a container dynamically from a YAML configuration file."""
        config, plugin_types = cls._load_configuration(config_path)

        namespace: dict[str, Any] = {}

        for name, device_class in plugin_types["devices"].items():
            cfg_kwargs = {
                k: v
                for k, v in config.get("devices", {}).get(name, {}).items()
                if k not in _PLUGIN_META_KEYS
            }
            namespace[name] = _DeviceComponent(device_class, name, **cfg_kwargs)

        for name, presenter_class in plugin_types["presenters"].items():
            cfg_kwargs = {
                k: v
                for k, v in config.get("presenters", {}).get(name, {}).items()
                if k not in _PLUGIN_META_KEYS
            }
            namespace[name] = _PresenterComponent(presenter_class, name, **cfg_kwargs)

        for name, view_class in plugin_types["views"].items():
            cfg_kwargs = {
                k: v
                for k, v in config.get("views", {}).get(name, {}).items()
                if k not in _PLUGIN_META_KEYS
            }
            namespace[name] = _ViewComponent(view_class, name, **cfg_kwargs)

        frontend = config.get("frontend", "pyqt")
        base_class = _resolve_frontend_container(frontend)

        DynamicApp: type[AppContainer] = type("DynamicApp", (base_class,), namespace)

        instance = DynamicApp(
            session=config.get("session", "Redsun"),
            frontend=frontend,
        )

        return instance

    @classmethod
    def _load_configuration(
        cls, config_path: str
    ) -> tuple[dict[str, Any], _PluginTypeDict]:
        """Load configuration and discover plugin classes from a YAML file."""
        with open(config_path, "r") as f:
            config: dict[str, Any] = yaml.safe_load(f)

        plugin_types: _PluginTypeDict = {"devices": {}, "presenters": {}, "views": {}}
        available_manifests = entry_points(group="redsun.plugins")

        groups: list[PLUGIN_GROUPS] = ["devices", "presenters", "views"]

        for group in groups:
            if group not in config:
                logger.debug(
                    "Group %s not found in the configuration file. Skipping", group
                )
                continue
            loaded = cls._load_plugins(
                group_cfg=config[group],
                group=group,
                available_manifests=available_manifests,
            )
            for name, plugin_cls in loaded:
                plugin_types[group][name] = plugin_cls  # type: ignore[assignment]

        return config, plugin_types

    @classmethod
    def _load_plugins(
        cls,
        *,
        group_cfg: dict[str, Any],
        group: PLUGIN_GROUPS,
        available_manifests: EntryPoints,
    ) -> list[tuple[str, PluginType]]:
        """Load plugin classes for a given group from manifests."""
        plugins: list[tuple[str, PluginType]] = []

        for name, info in group_cfg.items():
            plugin_name: str = info["plugin_name"]
            plugin_id: str = info["plugin_id"]

            iterator = (
                entry for entry in available_manifests if entry.name == plugin_name
            )
            plugin = next(iterator, None)

            if plugin is None:
                logger.error(
                    'Plugin "%s" not found in the installed plugins.', plugin_name
                )
                continue

            pkg_manifest = files(plugin.name.replace("-", "_")) / plugin.value
            with as_file(pkg_manifest) as manifest_path:
                with open(manifest_path, "r") as f:
                    manifest: dict[str, ManifestItems] = yaml.safe_load(f)

                if group not in manifest:
                    logger.error(
                        'Plugin "%s" manifest does not contain group "%s".',
                        plugin_name,
                        group,
                    )
                    continue

                items = manifest[group]
                if plugin_id not in items:
                    logger.error(
                        'Plugin "%s" does not contain the id "%s".',
                        plugin_name,
                        plugin_id,
                    )
                    continue

                class_path = items[plugin_id]
                try:
                    class_item_module, class_item_type = class_path.split(":")
                    imported_class = getattr(
                        import_module(class_item_module), class_item_type
                    )
                except (KeyError, ValueError):
                    logger.error(
                        'Plugin id "%s" of "%s" has invalid class path "%s". Skipping.',
                        plugin_id,
                        name,
                        class_path,
                    )
                    continue

                if not _check_plugin_protocol(imported_class, group):
                    logger.error(
                        "%s exists, but does not implement any known protocol.",
                        imported_class,
                    )
                    continue

                plugins.append((name, imported_class))

        return plugins

class AppConfig(RedSunConfig, total=False):
    """Extended configuration for Redsun application containers.

    Extends [`RedSunConfig`][redsun.virtual.RedSunConfig`] with component sections
    used by the application layer. These are **not** propagated to components.
    """

    devices: NotRequired[dict[str, Any]]
    presenters: NotRequired[dict[str, Any]]
    views: NotRequired[dict[str, Any]]

class QtAppContainer(AppContainer):
    """Application container for Qt-based frontends.

    Handles the full Qt lifecycle: ``QApplication`` creation, container
    build, ``QtMainView`` construction, virtual bus connection, and
    ``app.exec()``.

    Parameters
    ----------
    **config : Any
        Configuration options passed to :meth:`AppContainer.__init__`.
    """

    __slots__ = ("_qt_app", "_main_view")

    def __init__(self, **config: Any) -> None:
        super().__init__(**config)
        self._qt_app: QApplication | None = None
        self._main_view: QtMainView | None = None

    @property
    def main_view(self) -> QtMainView:
        """Return the main Qt window.

        Raises
        ------
        RuntimeError
            If the application has not been run yet.
        """
        if self._main_view is None:
            raise RuntimeError("Main view not built. Call run() first.")
        return self._main_view

    def build(self) -> QtAppContainer:
        """Ensure a ``QApplication`` exists, then build all components.

        If a ``QApplication`` is not yet running (e.g. when ``build()`` is
        called explicitly before ``run()``), one is created here so that
        view components that instantiate ``QWidget`` subclasses have a valid
        application object available.
        """
        if self._qt_app is None:
            self._qt_app = cast(
                "QApplication", QApplication.instance() or QApplication(sys.argv)
            )
        super().build()
        return self

    def run(self) -> NoReturn:
        """Build and launch the Qt application."""
        if self._qt_app is None:
            self._qt_app = cast(
                "QApplication", QApplication.instance() or QApplication(sys.argv)
            )

        if not self.is_built:
            self.build()

        assert self._qt_app is not None  # guaranteed by build() above
        session_name = self._config.get("session", "Redsun")
        self._main_view = QtMainView(
            virtual_container=self.virtual_container,
            session_name=session_name,
            views=cast("dict[str, QtView]", self.views),
        )

        # 4. Wire shutdown and start psygnal bridge
        self._qt_app.aboutToQuit.connect(self.shutdown)
        start_emitting_from_queue()

        # 6. Show and exec
        self._main_view.show()
        sys.exit(self._qt_app.exec())