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.