Source code for qudi.core.connector

# -*- coding: utf-8 -*-
"""
Connector object to establish connections between qudi modules.

Copyright (c) 2021, the qudi developers. See the AUTHORS.md file at the top-level directory of this
distribution and on <https://github.com/Ulm-IQO/qudi-core/>

This file is part of qudi.

Qudi is free software: you can redistribute it and/or modify it under the terms of
the GNU Lesser General Public License as published by the Free Software Foundation,
either version 3 of the License, or (at your option) any later version.

Qudi is distributed in the hope that it will be useful, but WITHOUT ANY WARRANTY;
without even the implied warranty of MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.
See the GNU Lesser General Public License for more details.

You should have received a copy of the GNU Lesser General Public License along with qudi.
If not, see <https://www.gnu.org/licenses/>.
"""

__all__ = ['Connector']

import weakref
from typing import Any, Type, Union
from qudi.util.overload import OverloadProxy




class Connector:
    """A connector used to connect qudi modules with each other.
    """



    def __init__(
        self, interface: Union[str, Type], name: str = None, optional: bool = False
    ):
        """Initialize a Connector instance.

        Parameters
        ----------
        interface : Union[str, Type]
            Name of the interface class to connect to or the interface class itself.
        name : str, optional
            Name of the connector in qudi config. Will set attribute name if omitted.
        optional : bool, optional
            Flag indicating if the connection is mandatory (False by default).

        Raises
        ------
        AssertionError
            If `interface` is not a string or a type.
            If `name` is not `None` or a non-empty string.
            If `optional` is not a boolean.
        """
        assert isinstance(interface, (str, type)), \
            'Parameter "interface" must be an interface class or the class name as str.'
        assert name is None or (isinstance(name, str) and name), \
            'Parameter "name" must be non-empty str or None.'
        assert isinstance(optional, bool), 'Parameter "optional" must be bool type.'
        self.interface = interface if isinstance(interface, str) else interface.__name__
        self.name = name
        self.optional = optional
        self._obj_proxy = None
        self._obj_ref = lambda: None


    def __set_name__(self, owner, name):
        if self.name is None:
            self.name = name

    def __call__(self) -> Any:
        """Return reference to the module that this connector is connected to."""
        if self.is_connected:
            return self._obj_proxy
        if self.optional:
            return None
        raise RuntimeError(
            f'Connector "{self.name}" (interface "{self.interface}") is not connected.'
        )

    def __copy__(self):
        return self.copy()

    def __deepcopy__(self, memodict={}):
        return self.copy()

    def __repr__(self):
        return f'{self.__module__}.Connector("{self.interface}", "{self.name}", {self.optional})'

    def __module_died_callback(self, ref=None):
        self.disconnect()

    @property
    def is_connected(self) -> bool:
        """Read-only property to check if the Connector instance is connected to a target module.

        Returns
        -------
        bool
            Connection status flag.
            - True: Connected
            - False: Disconnected
        """
        return self._obj_proxy is not None

    def connect(self, target: Any) -> None:
        """Check if target is connectible by this connector and connect.
        """
        bases = {cls.__name__ for cls in target.__class__.mro()}
        if self.interface not in bases:
            raise RuntimeError(
                f'Module "{target}" connected to connector "{self.name}" does not implement '
                f'interface "{self.interface}".'
            )
        self._obj_proxy = OverloadProxy(target, self.interface)
        self._obj_ref = weakref.ref(target, self.__module_died_callback)

    def disconnect(self) -> None:
        """Disconnect connector.
        """
        self._obj_proxy = None

    def copy(self, **kwargs):
        """Create a new instance of Connector with copied values and update
        """
        return Connector(kwargs.get('interface', self.interface),
                         kwargs.get('name', self.name),
                         kwargs.get('optional', self.optional))