"""@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