core_operations.py

"""@package docstring
B-ASIC Core Operations Module.
TODO: More info.
"""

from numbers import Number
from typing import Any
from numpy import conjugate, sqrt, abs as np_abs
from b_asic.port import InputPort, OutputPort
from b_asic.graph_id import GraphIDType
from b_asic.operation import AbstractOperation
from b_asic.graph_component import Name, TypeName


class Input(AbstractOperation):
    """Input operation.
    TODO: More info.
    """

    # TODO: Implement all functions.

    @property
    def type_name(self) -> TypeName:
        return "in"


class Constant(AbstractOperation):
    """Constant value operation.
    TODO: More info.
    """

    def __init__(self, value: Number = 0, name: Name = ""):
        super().__init__(name)

        self._output_ports = [OutputPort(0, self)]
        self._parameters["value"] = value

    def evaluate(self) -> Any:
        return self.param("value")

    @property
    def type_name(self) -> TypeName:
        return "c"


class Addition(AbstractOperation):
    """Binary addition operation.
    TODO: More info.
    """

    def __init__(self, source1: OutputPort = None, source2: OutputPort = None, name: Name = ""):
        super().__init__(name)

        self._input_ports = [InputPort(0, self), InputPort(1, self)]
        self._output_ports = [OutputPort(0, self)]

        if source1 is not None:
            self._input_ports[0].connect_to_port(source1)
        if source2 is not None:
            self._input_ports[1].connect_to_port(source2)

    def evaluate(self, a, b) -> Any:
        return a + b

    @property
    def type_name(self) -> TypeName:
        return "add"


class Subtraction(AbstractOperation):
    """Binary subtraction operation.
    TODO: More info.
    """

    def __init__(self, source1: OutputPort = None, source2: OutputPort = None, name: Name = ""):
        super().__init__(name)
        self._input_ports = [InputPort(0, self), InputPort(1, self)]
        self._output_ports = [OutputPort(0, self)]

        if source1 is not None:
            self._input_ports[0].connect_to_port(source1)
        if source2 is not None:
            self._input_ports[1].connect_to_port(source2)

    def evaluate(self, a, b) -> Any:
        return a - b

    @property
    def type_name(self) -> GraphIDType:
        return "sub"


class Multiplication(AbstractOperation):
    """Binary multiplication operation.
    TODO: More info.
    """

    def __init__(self, source1: OutputPort = None, source2: OutputPort = None, name: Name = ""):
        super().__init__(name)
        self._input_ports = [InputPort(0, self), InputPort(1, self)]
        self._output_ports = [OutputPort(0, self)]

        if source1 is not None:
            self._input_ports[0].connect_to_port(source1)
        if source2 is not None:
            self._input_ports[1].connect_to_port(source2)

    def evaluate(self, a, b) -> Any:
        return a * b

    @property
    def type_name(self) -> GraphIDType:
        return "mul"


class Division(AbstractOperation):
    """Binary division operation.
    TODO: More info.
    """

    def __init__(self, source1: OutputPort = None, source2: OutputPort = None, name: Name = ""):
        super().__init__(name)
        self._input_ports = [InputPort(0, self), InputPort(1, self)]
        self._output_ports = [OutputPort(0, self)]

        if source1 is not None:
            self._input_ports[0].connect_to_port(source1)
        if source2 is not None:
            self._input_ports[1].connect_to_port(source2)

    def evaluate(self, a, b) -> Any:
        return a / b

    @property
    def type_name(self) -> GraphIDType:
        return "div"


class SquareRoot(AbstractOperation):
    """Unary square root operation.
    TODO: More info.
    """

    def __init__(self, source1: OutputPort = None, name: Name = ""):
        super().__init__(name)
        self._input_ports = [InputPort(0, self)]
        self._output_ports = [OutputPort(0, self)]

        if source1 is not None:
            self._input_ports[0].connect_to_port(source1)


    def evaluate(self, a) -> Any:
        return sqrt((complex)(a))

    @property
    def type_name(self) -> GraphIDType:
        return "sqrt"


class ComplexConjugate(AbstractOperation):
    """Unary complex conjugate operation.
    TODO: More info.
    """

    def __init__(self, source1: OutputPort = None, name: Name = ""):
        super().__init__(name)
        self._input_ports = [InputPort(0, self)]
        self._output_ports = [OutputPort(0, self)]

        if source1 is not None:
            self._input_ports[0].connect_to_port(source1)


    def evaluate(self, a) -> Any:
        return conjugate(a)

    @property
    def type_name(self) -> GraphIDType:
        return "conj"


class Max(AbstractOperation):
    """Binary max operation.
    TODO: More info.
    """

    def __init__(self, source1: OutputPort = None, source2: OutputPort = None, name: Name = ""):
        super().__init__(name)
        self._input_ports = [InputPort(0, self), InputPort(1, self)]
        self._output_ports = [OutputPort(0, self)]

        if source1 is not None:
            self._input_ports[0].connect_to_port(source1)
        if source2 is not None:
            self._input_ports[1].connect_to_port(source2)

    def evaluate(self, a, b) -> Any:
        assert not isinstance(a, complex) and not isinstance(b, complex), \
            ("core_operation.Max does not support complex numbers.")
        return a if a > b else b

    @property
    def type_name(self) -> GraphIDType:
        return "max"


class Min(AbstractOperation):
    """Binary min operation.
    TODO: More info.
    """

    def __init__(self, source1: OutputPort = None, source2: OutputPort = None, name: Name = ""):
        super().__init__(name)
        self._input_ports = [InputPort(0, self), InputPort(1, self)]
        self._output_ports = [OutputPort(0, self)]

        if source1 is not None:
            self._input_ports[0].connect_to_port(source1)
        if source2 is not None:
            self._input_ports[1].connect_to_port(source2)

    def evaluate(self, a, b) -> Any:
        assert not isinstance(a, complex) and not isinstance(b, complex), \
            ("core_operation.Min does not support complex numbers.")
        return a if a < b else b

    @property
    def type_name(self) -> GraphIDType:
        return "min"


class Absolute(AbstractOperation):
    """Unary absolute value operation.
    TODO: More info.
    """

    def __init__(self, source1: OutputPort = None, name: Name = ""):
        super().__init__(name)
        self._input_ports = [InputPort(0, self)]
        self._output_ports = [OutputPort(0, self)]

        if source1 is not None:
            self._input_ports[0].connect_to_port(source1)


    def evaluate(self, a) -> Any:
        return np_abs(a)

    @property
    def type_name(self) -> GraphIDType:
        return "abs"


class ConstantMultiplication(AbstractOperation):
    """Unary constant multiplication operation.
    TODO: More info.
    """

    def __init__(self, coefficient: Number, source1: OutputPort = None, name: Name = ""):
        super().__init__(name)
        self._input_ports = [InputPort(0, self)]
        self._output_ports = [OutputPort(0, self)]
        self._parameters["coefficient"] = coefficient

        if source1 is not None:
            self._input_ports[0].connect_to_port(source1)

    def evaluate(self, a) -> Any:
        return a * self.param("coefficient")

    @property
    def type_name(self) -> TypeName:
        return "cmul"


class ConstantAddition(AbstractOperation):
    """Unary constant addition operation.
    TODO: More info.
    """

    def __init__(self, coefficient: Number, source1: OutputPort = None, name: Name = ""):
        super().__init__(name)
        self._input_ports = [InputPort(0, self)]
        self._output_ports = [OutputPort(0, self)]
        self._parameters["coefficient"] = coefficient

        if source1 is not None:
            self._input_ports[0].connect_to_port(source1)

    def evaluate(self, a) -> Any:
        return a + self.param("coefficient")

    @property
    def type_name(self) -> GraphIDType:
        return "cadd"


class ConstantSubtraction(AbstractOperation):
    """Unary constant subtraction operation.
    TODO: More info.
    """

    def __init__(self, coefficient: Number, source1: OutputPort = None, name: Name = ""):
        super().__init__(name)
        self._input_ports = [InputPort(0, self)]
        self._output_ports = [OutputPort(0, self)]
        self._parameters["coefficient"] = coefficient

        if source1 is not None:
            self._input_ports[0].connect_to_port(source1)

    def evaluate(self, a) -> Any:
        return a - self.param("coefficient")

    @property
    def type_name(self) -> GraphIDType:
        return "csub"


class ConstantDivision(AbstractOperation):
    """Unary constant division operation.
    TODO: More info.
    """

    def __init__(self, coefficient: Number, source1: OutputPort = None, name: Name = ""):
        super().__init__(name)
        self._input_ports = [InputPort(0, self)]
        self._output_ports = [OutputPort(0, self)]
        self._parameters["coefficient"] = coefficient

        if source1 is not None:
            self._input_ports[0].connect_to_port(source1)

    def evaluate(self, a) -> Any:
        return a / self.param("coefficient")

    @property
    def type_name(self) -> GraphIDType:
        return "cdiv"