core_operations.py

        return a + b, a - b


class MAD(AbstractOperation):
    r"""
    Multiply-add operation.

    Gives the result of multiplying the first input by the second input and
    then adding the third input.

    .. math:: y = x_0 \times x_1 + x_2

    Parameters
    ==========

    src0, src1, src2 : SignalSourceProvider, optional
        The three signals to determine the multiply-add operation of.
    name : Name, optional
        Operation name.
    latency : int, optional
        Operation latency (delay from input to output in time units).
    latency_offsets : dict[str, int], optional
        Used if inputs have different arrival times or if the inputs should arrive
        after the operator has stared. For example, ``{"in0": 0, "in1": 0, "in2": 2}``
        which corresponds to *src2*, i.e., the term to be added, arriving two time
        units later than *src0* and *src1*. If not provided and *latency* is provided,
        set to zero. Hence, the previous example can be written as ``{"in1": 1}``
        only.
    execution_time : int, optional
        Operation execution time (time units before operator can be reused).

    See Also
    --------
    Multiplication
    Addition
    """

    __slots__ = (
        "_src0",
        "_src1",
        "_src2",
        "_name",
        "_latency",
        "_latency_offsets",
        "_execution_time",
    )
    _src0: Optional[SignalSourceProvider]
    _src1: Optional[SignalSourceProvider]
    _src2: Optional[SignalSourceProvider]
    _name: Name
    _latency: Optional[int]
    _latency_offsets: Optional[Dict[str, int]]
    _execution_time: Optional[int]

    is_swappable = True

    def __init__(
        self,
        src0: Optional[SignalSourceProvider] = None,
        src1: Optional[SignalSourceProvider] = None,
        src2: Optional[SignalSourceProvider] = None,
        name: Name = Name(""),
        latency: Optional[int] = None,
        latency_offsets: Optional[Dict[str, int]] = None,
        execution_time: Optional[int] = None,
    ):
        """Construct a MAD operation."""
        super().__init__(
            input_count=3,
            output_count=1,
            name=Name(name),
            input_sources=[src0, src1, src2],
            latency=latency,
            latency_offsets=latency_offsets,
            execution_time=execution_time,
        )

    @classmethod
    def type_name(cls) -> TypeName:
        return TypeName("mad")

    def evaluate(self, a, b, c):
        return a * b + c

    @property
    def is_linear(self) -> bool:
        return (
            self.input(0).connected_source.operation.is_constant
            or self.input(1).connected_source.operation.is_constant
        )

    def swap_io(self) -> None:
        self._input_ports = [
            self._input_ports[1],
            self._input_ports[0],
            self._input_ports[2],
        ]
        for i, p in enumerate(self._input_ports):
            p._index = i


class MADS(AbstractOperation):
    __slots__ = (
        "_is_add",
        "_override_zero_on_src0",
        "_src0",
        "_src1",
        "_src2",
        "_name",
        "_latency",
        "_latency_offsets",
        "_execution_time",
    )
    _is_add: Optional[bool]
    _override_zero_on_src0: Optional[bool]
    _src0: Optional[SignalSourceProvider]
    _src1: Optional[SignalSourceProvider]
    _src2: Optional[SignalSourceProvider]
    _name: Name
    _latency: Optional[int]
    _latency_offsets: Optional[Dict[str, int]]
    _execution_time: Optional[int]

    is_swappable = True

    def __init__(
        self,
        is_add: Optional[bool] = True,
        override_zero_on_src0: Optional[bool] = False,
        src0: Optional[SignalSourceProvider] = None,
        src1: Optional[SignalSourceProvider] = None,
        src2: Optional[SignalSourceProvider] = None,
        name: Name = Name(""),
        latency: Optional[int] = None,
        latency_offsets: Optional[Dict[str, int]] = None,
        execution_time: Optional[int] = None,
    ):
        """Construct a MADS operation."""
        super().__init__(
            input_count=3,
            output_count=1,
            name=Name(name),
            input_sources=[src0, src1, src2],
            latency=latency,
            latency_offsets=latency_offsets,
            execution_time=execution_time,
        )
        self.set_param("is_add", is_add)
        self.set_param("override_zero_on_src0", override_zero_on_src0)

    @classmethod
    def type_name(cls) -> TypeName:
        return TypeName("mads")

    def evaluate(self, a, b, c):
        if self.is_add:
            if self.override_zero_on_src0:
                return b * c
            else:
                return a + b * c
        else:
            if self.override_zero_on_src0:
                return -b * c
            else:
                return a - b * c

    @property
    def is_add(self) -> bool:
        """Get if operation is an addition."""
        return self.param("is_add")

    @is_add.setter
    def is_add(self, is_add: bool) -> None:
        """Set if operation is an addition."""
        self.set_param("is_add", is_add)

    @property
    def override_zero_on_src0(self) -> bool:
        """Get if operation is overriding a zero on port src0."""
        return self.param("override_zero_on_src0")

    @override_zero_on_src0.setter
    def override_zero_on_src0(self, override_zero_on_src0: bool) -> None:
        """Set if operation is overriding a zero on port src0."""
        self.set_param("override_zero_on_src0", override_zero_on_src0)

    @property
    def is_linear(self) -> bool:
        return (
            self.input(1).connected_source.operation.is_constant
            or self.input(2).connected_source.operation.is_constant
        )

    def swap_io(self) -> None:
        self._input_ports = [
            self._input_ports[0],
            self._input_ports[2],
            self._input_ports[1],
        ]
        for i, p in enumerate(self._input_ports):
            p._index = i


class SymmetricTwoportAdaptor(AbstractOperation):
    r"""
    Wave digital filter symmetric twoport-adaptor operation.

    .. math::
        \begin{eqnarray}
        y_0 & = & x_1 + \text{value}\times\left(x_1 - x_0\right)\\
        y_1 & = & x_0 + \text{value}\times\left(x_1 - x_0\right)
        \end{eqnarray}
    """

    __slots__ = (
        "_src0",
        "_src1",
        "_name",
        "_latency",
        "_latency_offsets",
        "_execution_time",
    )
    _src0: Optional[SignalSourceProvider]
    _src1: Optional[SignalSourceProvider]
    _name: Name
    _latency: Optional[int]
    _latency_offsets: Optional[Dict[str, int]]
    _execution_time: Optional[int]

    is_linear = True
    is_swappable = True

    def __init__(
        self,
        value: Num = 0,
        src0: Optional[SignalSourceProvider] = None,
        src1: Optional[SignalSourceProvider] = None,
        name: Name = Name(""),
        latency: Optional[int] = None,
        latency_offsets: Optional[Dict[str, int]] = None,
        execution_time: Optional[int] = None,
    ):
        """Construct a SymmetricTwoportAdaptor operation."""
        super().__init__(
            input_count=2,
            output_count=2,
            name=Name(name),
            input_sources=[src0, src1],
            latency=latency,
            latency_offsets=latency_offsets,
            execution_time=execution_time,
        )
        self.value = value

    @classmethod
    def type_name(cls) -> TypeName:
        return TypeName("sym2p")

    def evaluate(self, a, b):
        tmp = self.value * (b - a)
        return b + tmp, a + tmp

    @property
    def value(self) -> Num:
        """Get the constant value of this operation."""
        return self.param("value")

    @value.setter
    def value(self, value: Num) -> None:
        """Set the constant value of this operation."""
        if -1 <= value <= 1:
            self.set_param("value", value)
        else:
            raise ValueError('value must be between -1 and 1 (inclusive)')

    def swap_io(self) -> None:
        # Swap inputs and outputs and change sign of coefficient
        self._input_ports.reverse()
        for i, p in enumerate(self._input_ports):
            p._index = i
        self._output_ports.reverse()
        for i, p in enumerate(self._output_ports):
            p._index = i
        self.set_param("value", -self.value)


class Reciprocal(AbstractOperation):
    r"""
    Reciprocal operation.

    Gives the reciprocal of its input.

    .. math:: y = \frac{1}{x}

    Parameters
    ----------
    src0 : :class:`~b_asic.port.SignalSourceProvider`, optional
        The signal to compute the reciprocal of.
    name : Name, optional
        Operation name.
    latency : int, optional
        Operation latency (delay from input to output in time units).
    latency_offsets : dict[str, int], optional
        Used if input arrives later than when the operator starts, e.g.,
        ``{"in0": 0`` which corresponds to *src0* arriving one time unit after the
        operator starts. If not provided and *latency* is provided, set to zero.
    execution_time : int, optional
        Operation execution time (time units before operator can be reused).

    See also
    ========
    Division
    """

    __slots__ = ("_src0", "_name", "_latency", "_latency_offsets", "_execution_time")
    _src0: Optional[SignalSourceProvider]
    _name: Name
    _latency: Optional[int]
    _latency_offsets: Optional[Dict[str, int]]
    _execution_time: Optional[int]

    def __init__(
        self,
        src0: Optional[SignalSourceProvider] = None,
        name: Name = Name(""),
        latency: Optional[int] = None,
        latency_offsets: Optional[Dict[str, int]] = None,
        execution_time: Optional[int] = None,
    ):
        """Construct a Reciprocal operation."""
        super().__init__(
            input_count=1,
            output_count=1,
            name=Name(name),
            input_sources=[src0],
            latency=latency,
            latency_offsets=latency_offsets,
            execution_time=execution_time,
        )

    @classmethod
    def type_name(cls) -> TypeName:
        return TypeName("rec")

    def evaluate(self, a):
        return 1 / a


class RightShift(AbstractOperation):
    r"""
    Arithmetic right-shift operation.

    Shifts the input to the right assuming a fixed-point representation, so
    a multiplication by a power of two.

    .. math:: y = x \gg \text{value} = 2^{-\text{value}}x \text{ where value} \geq 0

    Parameters
    ----------
    value : int
        Number of bits to shift right.
    src0 : :class:`~b_asic.port.SignalSourceProvider`, optional
        The signal to shift right.
    name : Name, optional
        Operation name.
    latency : int, optional
        Operation latency (delay from input to output in time units).
    latency_offsets : dict[str, int], optional
        Used if input arrives later than when the operator starts, e.g.,
        ``{"in0": 0`` which corresponds to *src0* arriving one time unit after the
        operator starts. If not provided and *latency* is provided, set to zero.
    execution_time : int, optional
        Operation execution time (time units before operator can be reused).

    See Also
    --------
    LeftShift
    Shift
    """

    __slots__ = (
        "_value",
        "_src0",
        "_name",
        "_latency",
        "_latency_offsets",
        "_execution_time",
    )
    _value: Num
    _src0: Optional[SignalSourceProvider]
    _name: Name
    _latency: Optional[int]
    _latency_offsets: Optional[Dict[str, int]]
    _execution_time: Optional[int]

    is_linear = True

    def __init__(
        self,
        value: int = 0,
        src0: Optional[SignalSourceProvider] = None,
        name: Name = Name(""),
        latency: Optional[int] = None,
        latency_offsets: Optional[Dict[str, int]] = None,
        execution_time: Optional[int] = None,
    ):
        """Construct a RightShift operation with the given value."""
        super().__init__(
            input_count=1,
            output_count=1,
            name=Name(name),
            input_sources=[src0],
            latency=latency,
            latency_offsets=latency_offsets,
            execution_time=execution_time,
        )
        self.value = value

    @classmethod
    def type_name(cls) -> TypeName:
        return TypeName("rshift")

    def evaluate(self, a):
        return a * 2 ** (-self.param("value"))

    @property
    def value(self) -> int:
        """Get the constant value of this operation."""
        return self.param("value")

    @value.setter
    def value(self, value: int) -> None:
        """Set the constant value of this operation."""
        if not isinstance(value, int):
            raise TypeError("value must be an int")
        if value < 0:
            raise ValueError("value must be non-negative")
        self.set_param("value", value)


class LeftShift(AbstractOperation):
    r"""
    Arithmetic left-shift operation.

    Shifts the input to the left assuming a fixed-point representation, so
    a multiplication by a power of two.

    .. math:: y = x \ll \text{value} = 2^{\text{value}}x \text{ where value} \geq 0

    Parameters
    ----------
    value : int
        Number of bits to shift left.
    src0 : :class:`~b_asic.port.SignalSourceProvider`, optional
        The signal to shift left.
    name : Name, optional
        Operation name.
    latency : int, optional
        Operation latency (delay from input to output in time units).
    latency_offsets : dict[str, int], optional
        Used if input arrives later than when the operator starts, e.g.,
        ``{"in0": 0`` which corresponds to *src0* arriving one time unit after the
        operator starts. If not provided and *latency* is provided, set to zero.
    execution_time : int, optional
        Operation execution time (time units before operator can be reused).

    See Also
    --------
    RightShift
    Shift
    """

    __slots__ = (
        "_value",
        "_src0",
        "_name",
        "_latency",
        "_latency_offsets",
        "_execution_time",
    )
    _value: Num
    _src0: Optional[SignalSourceProvider]
    _name: Name
    _latency: Optional[int]
    _latency_offsets: Optional[Dict[str, int]]
    _execution_time: Optional[int]

    is_linear = True

    def __init__(
        self,
        value: int = 0,
        src0: Optional[SignalSourceProvider] = None,
        name: Name = Name(""),
        latency: Optional[int] = None,
        latency_offsets: Optional[Dict[str, int]] = None,
        execution_time: Optional[int] = None,
    ):
        """Construct a RightShift operation with the given value."""
        super().__init__(
            input_count=1,
            output_count=1,
            name=Name(name),
            input_sources=[src0],
            latency=latency,
            latency_offsets=latency_offsets,
            execution_time=execution_time,
        )
        self.value = value

    @classmethod
    def type_name(cls) -> TypeName:
        return TypeName("lshift")

    def evaluate(self, a):
        return a * 2 ** (self.param("value"))

    @property
    def value(self) -> int:
        """Get the constant value of this operation."""
        return self.param("value")

    @value.setter
    def value(self, value: int) -> None:
        """Set the constant value of this operation."""
        if not isinstance(value, int):
            raise TypeError("value must be an int")
        if value < 0:
            raise ValueError("value must be non-negative")
        self.set_param("value", value)


class Shift(AbstractOperation):
    r"""
    Arithmetic shift operation.

    Shifts the input to the left or right assuming a fixed-point representation, so
    a multiplication by a power of two. By definition a positive value is a shift to
    the left.

    .. math:: y = x \ll \text{value} = 2^{\text{value}}x

    Parameters
    ----------
    value : int
        Number of bits to shift. Positive *value* shifts to the left.
    src0 : :class:`~b_asic.port.SignalSourceProvider`, optional
        The signal to shift.
    name : Name, optional
        Operation name.
    latency : int, optional
        Operation latency (delay from input to output in time units).
    latency_offsets : dict[str, int], optional
        Used if input arrives later than when the operator starts, e.g.,
        ``{"in0": 0`` which corresponds to *src0* arriving one time unit after the
        operator starts. If not provided and *latency* is provided, set to zero.
    execution_time : int, optional
        Operation execution time (time units before operator can be reused).

    See Also
    --------
    LeftShift
    RightShift
    """

    __slots__ = (
        "_value",
        "_src0",
        "_name",
        "_latency",
        "_latency_offsets",
        "_execution_time",
    )
    _value: Num
    _src0: Optional[SignalSourceProvider]
    _name: Name
    _latency: Optional[int]
    _latency_offsets: Optional[Dict[str, int]]
    _execution_time: Optional[int]

    is_linear = True

    def __init__(
        self,
        value: int = 0,
        src0: Optional[SignalSourceProvider] = None,
        name: Name = Name(""),
        latency: Optional[int] = None,
        latency_offsets: Optional[Dict[str, int]] = None,
        execution_time: Optional[int] = None,
    ):
        """Construct a Shift operation with the given value."""
        super().__init__(
            input_count=1,
            output_count=1,
            name=Name(name),
            input_sources=[src0],
            latency=latency,
            latency_offsets=latency_offsets,
            execution_time=execution_time,
        )
        self.value = value

    @classmethod
    def type_name(cls) -> TypeName:
        return TypeName("shift")

    def evaluate(self, a):
        return a * 2 ** (self.param("value"))

    @property
    def value(self) -> int:
        """Get the constant value of this operation."""
        return self.param("value")

    @value.setter
    def value(self, value: int) -> None:
        """Set the constant value of this operation."""
        if not isinstance(value, int):
            raise TypeError("value must be an int")
        self.set_param("value", value)


class DontCare(AbstractOperation):
    r"""
    Dont-care operation

    Used for ignoring the input to another operation and thus avoiding dangling input nodes.

    Parameters
    ----------
    name : Name, optional
        Operation name.

    """

    __slots__ = "_name"
    _name: Name

    is_linear = True

    def __init__(self, name: Name = ""):
        """Construct a DontCare operation."""
        super().__init__(
            input_count=0,
            output_count=1,
            name=name,
            latency_offsets={"out0": 0},
        )

    @classmethod
    def type_name(cls) -> TypeName:
        return TypeName("dontcare")

    def evaluate(self):
        return 0

    @property
    def latency(self) -> int:
        return self.latency_offsets["out0"]

    def __repr__(self) -> str:
        return "DontCare()"

    def __str__(self) -> str:
        return "dontcare"


class Sink(AbstractOperation):
    r"""
    Sink operation.

    Used for ignoring the output from another operation to avoid dangling output nodes.

    Parameters
    ==========

    name : Name, optional
        Operation name.
    """

    __slots__ = "_name"
    _name: Name

    is_linear = True

    def __init__(self, name: Name = ""):
        """Construct a Sink operation."""
        super().__init__(
            input_count=1,
            output_count=0,
            name=name,
            latency_offsets={"in0": 0},
        )

    @classmethod
    def type_name(cls) -> TypeName:
        return TypeName("sink")

    def evaluate(self):
        raise NotImplementedError

    @property
    def latency(self) -> int:
        return self.latency_offsets["in0"]

    def __repr__(self) -> str:
        return "Sink()"

    def __str__(self) -> str:
        return "sink"