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

PortIndex = NewType("PortIndex", int)

class Port(ABC):
    """Port Interface.

    TODO: More documentaiton?
    """

    @property
    @abstractmethod
    def operation(self) -> Operation:
        """Return the connected operation."""
        raise NotImplementedError

    @property
    @abstractmethod
    def index(self) -> PortIndex:
        """Return the unique PortIndex."""
        raise NotImplementedError

    @property
    @abstractmethod
    def signals(self) -> List[Signal]:
        """Return a list of all connected signals."""
        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(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 add_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(self, port: "Port") -> None:
        """Disconnect the entered port from the port by removing it from the ports signal.
        If the entered port is still connected to this ports signal then disconnect the entered
        port from the signal aswell.
        """
        raise NotImplementedError

    @abstractmethod
    def remove_signal(self, signal: Signal) -> None:
        """Remove the signal that was entered from the Ports 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.
        """
        raise NotImplementedError

    @abstractmethod
    def clear(self) -> None:
        """Removes all connected signals from the Port."""
        raise NotImplementedError


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

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

    _index: int
    _operation: Operation

    def __init__(self, index: int, operation: Operation):
        self._index = index
        self._operation = operation

    @property
    def operation(self) -> Operation:
        return self._operation

    @property
    def index(self) -> PortIndex:
        return self._index


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

    _source_signal: Optional[Signal]
    _value_length: Optional[int]

    def __init__(self, port_id: PortIndex, operation: Operation):
        super().__init__(port_id, operation)
        self._source_signal = None
        self._value_length = None

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

    @property
    def value_length(self) -> Optional[int]:
        return self._value_length

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

    @value_length.setter
    def value_length(self, bits: Optional[int]) -> None:
        assert (isinstance(bits, int) and bits >= 0) or bits is None, \
            "Value length on input port has to be a non-negative integer or None."
        self._value_length = bits

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

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

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

    def disconnect(self, port: "OutputPort") -> None:
        assert self._source_signal.source is port, "The entered port is not connected to this port."
        self._source_signal.remove_source()

    def remove_signal(self, signal: Signal) -> None:
        old_signal: Signal = self._source_signal
        self._source_signal = None
        if self is old_signal.destination:
            # Disconnect the dest of the signal if this inputport currently is the dest.
            old_signal.remove_destination()

    def clear(self) -> None:
        self.remove_signal(self._source_signal)


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

    _destination_signals: List[Signal]

    def __init__(self, port_id: PortIndex, operation: Operation):
        super().__init__(port_id, operation)
        self._destination_signals = []

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

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

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

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

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

    def add_signal(self, signal: Signal) -> None:
        assert signal not in self.signals, \
                "Attempting to connect to Signal already connected."
        self._destination_signals.append(signal)
        if self is not signal.source:
            # Connect this outputport to the signal if it isn't already.
            signal.set_source(self)

    def disconnect(self, port: InputPort) -> None:
        assert port in self.connected_ports, "Attempting to disconnect port that isn't connected."
        for sig in self._destination_signals:
            if sig.destination is port:
                sig.remove_destination()
                break

    def remove_signal(self, signal: Signal) -> None:
        i: int = self._destination_signals.index(signal)
        old_signal: Signal = self._destination_signals[i]
        del self._destination_signals[i]
        if self is old_signal.source:
            old_signal.remove_source()

    def clear(self) -> None:
        for signal in self._destination_signals:
            self.remove_signal(signal)