diff --git a/b_asic/port.py b/b_asic/port.py index c22053df1928cf2f6ea32c4880816d551b85836a..64c56bc9aef2ca71c78d08ccf0b8c20c1c68fd61 100644 --- a/b_asic/port.py +++ b/b_asic/port.py @@ -35,15 +35,6 @@ class Port(ABC): """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"]: @@ -58,20 +49,23 @@ class Port(ABC): @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.""" + 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.""" + 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.""" + port from the signal aswell. + """ raise NotImplementedError @abstractmethod @@ -119,31 +113,38 @@ class InputPort(AbstractPort): """ _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] - def signal(self, i: int = 0) -> Signal: - assert 0 <= i < self.signal_count(), "Signal index out of bound." - assert self._source_signal is not None, "No Signal connect to InputPort." - return 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 + 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." @@ -160,12 +161,13 @@ class InputPort(AbstractPort): 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 + # 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. @@ -194,14 +196,14 @@ class OutputPort(AbstractPort): 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 + 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 + # Connect this outputport to the signal if it isn't already. signal.set_source(self) def disconnect(self, port: InputPort) -> None: diff --git a/test/test_inputport.py b/test/test_inputport.py index a43240693ac632b48461023536ff46b0ea379c5c..a0f24e4e90cb69553a200e0a8402ac248c14d3b9 100644 --- a/test/test_inputport.py +++ b/test/test_inputport.py @@ -24,19 +24,15 @@ def dangling_sig(): return Signal() @pytest.fixture -def s_w_source(): - out_port = OutputPort(0, None) +def s_w_source(out_port): return Signal(source=out_port) @pytest.fixture -def sig_with_dest(): - inp_port = InputPort(0, None) - return Signal(destination=out_port) +def sig_with_dest(inp_port): + return Signal(destination=inp_port) @pytest.fixture -def connected_sig(): - out_port = OutputPort(0, None) - inp_port = InputPort(0, None) +def connected_sig(inp_port, out_port): return Signal(source=out_port, destination=inp_port) def test_connect_then_disconnect(inp_port, out_port): @@ -83,13 +79,27 @@ def test_add_signal_then_disconnect(inp_port, s_w_source): assert s_w_source.source.signals == [s_w_source] assert s_w_source.destination is None -def test_connect_then_disconnect(inp_port, out_port): - """Can port be connected and then disconnected properly?""" - inp_port.connect(out_port) +def test_set_value_length_pos_int(inp_port): + inp_port.value_length = 10 + assert inp_port.value_length == 10 + +def test_set_value_length_zero(inp_port): + inp_port.value_length = 0 + assert inp_port.value_length == 0 + +def test_set_value_length_neg_int(inp_port): + with pytest.raises(Exception): + inp_port.value_length = -10 + +def test_set_value_length_complex(inp_port): + with pytest.raises(Exception): + inp_port.value_length = (2+4j) - inp_port.disconnect(out_port) +def test_set_value_length_float(inp_port): + with pytest.raises(Exception): + inp_port.value_length = 3.2 - print("outport signals:", out_port.signals, "count:", out_port.signal_count()) - assert inp_port.signal_count() == 1 - assert len(inp_port.connected_ports) == 0 - assert out_port.signal_count() == 0 +def test_set_value_length_pos_then_none(inp_port): + inp_port.value_length = 10 + inp_port.value_length = None + assert inp_port.value_length is None