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.signal import Signal
from b_asic.operation import Operation

PortId = NewType("PortId", int)


class Port(ABC):
    """Abstract port class.
    TODO: More info.
    """

    _port_id: PortId
    _operation: Operation

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

    @property
    def identifier(self) -> PortId:
        """Get the unique identifier."""
        return self._port_id

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

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

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

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

    @abstractmethod
    def connect(self, signal: Signal) -> None:
        """Connect a signal."""
        raise NotImplementedError

    @abstractmethod
    def disconnect(self, i: int = 0) -> None:
        """Disconnect a signal."""
        raise NotImplementedError


class InputPort(Port):
    """Input port.
    TODO: More info.
    """
    _source_signal: Optional[Signal]

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

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

    def signal(self, i: int = 0) -> Signal:
        assert 0 <= i < self.signal_count() # TODO: Error message.
        assert self._source_signal is not None # TODO: Error message.
        return self._source_signal

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

    def connect(self, signal: Signal) -> None:
        self._source_signal = signal
        signal.destination = self

    def disconnect(self, i: int = 0) -> None:
        assert 0 <= i < self.signal_count() # TODO: Error message.
        self._source_signal.disconnect_source()
        self._source_signal = None

    # TODO: More stuff.


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

    _destination_signals: List[Signal]

    def __init__(self, port_id: PortId, 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() # TODO: Error message.
        return self._destination_signals[i]

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

    def connect(self, signal: Signal) -> None:
        assert signal not in self._destination_signals # TODO: Error message.
        self._destination_signals.append(signal)
        signal.source = self

    def disconnect(self, i: int = 0) -> None:
        assert 0 <= i < self.signal_count() # TODO: Error message.
        del self._destination_signals[i]

    # TODO: More stuff.