port.py

"""@package docstring
B-ASIC Port Module.
TODO: More info.
"""

from abc import ABC, abstractmethod
from typing import NewType, Optional, List

from b_asic.operation import Operation
from b_asic.signal import Signal

PortId = NewType("PortId", int)


class Port(ABC):
    """Abstract port class.

    Handles functionality for port id and saves the connection to the parent operation.
    """

    _port_id: PortId
    _operation: Operation

    def __init__(self, port_id: PortId, operation: Operation):
        self._port_id = port_id
        self._operation = operation

    @property
    def id(self) -> PortId:
        """Return the unique portid."""
        return self._port_id

    @property
    def operation(self) -> Operation:
        """Return the connected operation."""
        return self._operation

    @property
    @abstractmethod
    def signals(self) -> List[Signal]:
        """Return a list of all connected signals."""
        raise NotImplementedError

    @abstractmethod
    def signal(self, i: int = 0) -> Signal:
        """Return the connected signal at index i.

        Keyword argumens:
        i: integer index of the signal requsted.
        """
        raise NotImplementedError

    @property
    @abstractmethod
    def connected_ports(self) -> List["Port"]:
        """Return a list of all connected Ports."""
        raise NotImplementedError

    @abstractmethod
    def signal_count(self) -> int:
        """Return the number of connected signals."""
        raise NotImplementedError

    @abstractmethod
    def connect_port(self, port: "Port") -> Signal:
        """Create and return a signal that is connected to this port and the entered
        port and connect this port to the signal and the entered port to the signal."""
        raise NotImplementedError

    @abstractmethod
    def connect_signal(self, signal: Signal) -> None:
        """Connect this port to the entered signal. If the entered signal isn't connected to
        this port then connect the entered signal to the port aswell."""
        raise NotImplementedError

    @abstractmethod
    def disconnect_signal(self, i: int = 0) -> None:
        """Disconnect a signal from the port. If the port is still connected to the entered signal
        then the port is disconnected from the the entered signal aswell."""
        raise NotImplementedError

    @abstractmethod
    def is_connected_to_signal(self, signal: Signal) -> bool:
        """Return true if the port is connected to the entered signal else false."""
        raise NotImplementedError


class InputPort(Port):
    """Input port.
    TODO: More info.
    """

    _signal: Optional[Signal]

    def __init__(self, port_id: PortId, operation: Operation):
        super().__init__(port_id, operation)
        self._signal = None

    @property
    def signals(self) -> List[Signal]:
        return [] if self._signal is None else [self._signal]

    def signal(self, i: int = 0) -> Signal:
        assert 0 <= i < self.signal_count(), "Signal index out of bound."
        assert self._signal is not None, "No Signal connect to InputPort."
        return self._signal

    @property
    def connected_ports(self) -> List[Port]:
        return [] if self._signal is None else [self._signal.source]

    def signal_count(self) -> int:
        return 0 if self._signal is None else 1

    def connect_port(self, port: "OutputPort") -> Signal:
        assert self._signal is None, "Connecting new port to already connected input port."
        return Signal(port, self)        # self._signal is set by the signal constructor

    def connect_signal(self, signal: Signal) -> None:
        assert self._signal is None, "Connecting new port to already connected input port."
        self._signal = signal
        if self is not signal.destination:
            # Connect this inputport as destination for this signal if it isn't already.
            signal.connect_destination(self)

    def disconnect_signal(self, i: int = 0) -> None:
        assert 0 <= i < self.signal_count(), "Signal Index out of range."
        old_signal: Signal = self._signal
        self._signal = None
        if self is old_signal.destination:
            # Disconnect the dest of the signal if this inputport currently is the dest
            old_signal.disconnect_destination()
        old_signal.disconnect_destination()

    def is_connected_to_signal(self, signal: Signal) -> bool:
        return self._signal is signal


class OutputPort(Port):
    """Output port.
    TODO: More info.
    """

    _signals: List[Signal]

    def __init__(self, port_id: PortId, operation: Operation):
        super().__init__(port_id, operation)
        self._signals = []

    @property
    def signals(self) -> List[Signal]:
        return self._signals.copy()

    def signal(self, i: int = 0) -> Signal:
        assert 0 <= i < self.signal_count(), "Signal index out of bounds."
        return self._signals[i]

    @property
    def connected_ports(self) -> List[Port]:
        return [signal.destination for signal in self._signals \
            if signal.destination is not None]

    def signal_count(self) -> int:
        return len(self._signals)

    def connect_port(self, port: InputPort) -> Signal:
        return Signal(self, port)      # Signal is added to self._signals in signal constructor

    def connect_signal(self, signal: Signal) -> None:
        assert not self.is_connected_to_signal(signal), \
                "Attempting to connect to Signal already connected."
        self._signals.append(signal)
        if self is not signal.source:
            # Connect this outputport to the signal if it isn't already
            signal.connect_source(self)

    def disconnect_signal(self, i: int = 0) -> None:
        assert 0 <= i < self.signal_count(), "Signal index out of bounds."
        old_signal: Signal = self._signals[i]
        del self._signals[i]
        if self is old_signal.source:
            # Disconnect the source of the signal if this outputport currently is the source
            old_signal.disconnect_source()

    def disconnect_signal_by_ref(self, signal: Signal) -> None:
        """Remove the signal that was entered from the OutputPorts signals.
        If the entered signal still is connected to this port then disconnect the
        entered signal from the port aswell.

        Keyword arguments:
        - signal: Signal to remove.
        """
        i: int = self._signals.index(signal)
        self.disconnect_signal(i)

    def is_connected_to_signal(self, signal: Signal) -> bool:
        return signal in self._signals                  # O(n) complexity