Add FIR generators

b_asic/sfg_generator.py → b_asic/sfg_generators.py

@@ -7,7 +7,12 @@ from typing import Dict, Optional, Sequence, Union
 
 import numpy as np
 
-from b_asic.core_operations import Name, SymmetricTwoportAdaptor
+from b_asic.core_operations import (
+    Addition,
+    ConstantMultiplication,
+    Name,
+    SymmetricTwoportAdaptor,
+)
 from b_asic.port import InputPort, OutputPort
 from b_asic.signal import Signal
 from b_asic.signal_flow_graph import SFG
@@ -109,3 +114,165 @@ def wdf_allpass(
 
     output << signal_out
     return SFG([input_op], [output], name=Name(name))
+
+
+def direct_form_fir(
+    coefficients: Sequence[complex],
+    input_op: Optional[Union[Input, Signal, InputPort]] = None,
+    output: Optional[Union[Output, Signal, OutputPort]] = None,
+    name: Optional[str] = None,
+    mult_properties: Optional[
+        Union[Dict[str, int], Dict[str, Dict[str, int]]]
+    ] = None,
+    add_properties: Optional[
+        Union[Dict[str, int], Dict[str, Dict[str, int]]]
+    ] = None,
+):
+    r"""
+    Generate a signal flow graph of a direct form FIR filter. The *coefficients* parameter is a
+    sequence of impulse response values::
+
+        coefficients = [h0, h1, h2, ..., hN]
+
+    Leading to the transfer function:
+
+    .. math:: \sum_{i=0}^N h_iz^{-i}
+
+    Parameters
+    ----------
+    coefficients : 1D-array
+        Coefficients to use for the FIR filter section
+
+    input_op : Input, optional
+        The Input to connect the SFG to. If not provided, one will be generated.
+
+    output : Output, optional
+        The Output to connect the SFG to. If not provided, one will be generated.
+
+    name : Name, optional
+        The name of the SFG. If None, "WDF allpass section".
+
+    mult_properties : dictionary, optional
+        Properties passed to :class:`~b_asic.core_operations.ConstantMultiplication`.
+
+    add_properties : dictionary, optional
+        Properties passed to :class:`~b_asic.core_operations.Addition`.
+
+    Returns
+    -------
+        Signal flow graph
+
+    See also
+    --------
+    transposed_direct_form_fir
+    """
+    np_coefficients = np.squeeze(np.asarray(coefficients))
+    if np_coefficients.ndim != 1:
+        raise TypeError("coefficients must be a 1D-array")
+    if input_op is None:
+        input_op = Input()
+    if output is None:
+        output = Output()
+    if name is None:
+        name = "Direct-form FIR filter"
+    if mult_properties is None:
+        mult_properties = {}
+    if add_properties is None:
+        add_properties = {}
+
+    taps = len(np_coefficients)
+    prev_delay = input_op
+    prev_add = None
+    for i, coeff in enumerate(np_coefficients):
+        tmp_mul = ConstantMultiplication(coeff, prev_delay, **mult_properties)
+        if prev_add is None:
+            prev_add = tmp_mul
+        else:
+            prev_add = Addition(tmp_mul, prev_add, **add_properties)
+        if i < taps - 1:
+            prev_delay = Delay(prev_delay)
+
+    output << prev_add
+
+    return SFG([input_op], [output], name=Name(name))
+
+
+def transposed_direct_form_fir(
+    coefficients: Sequence[complex],
+    input_op: Optional[Union[Input, Signal, InputPort]] = None,
+    output: Optional[Union[Output, Signal, OutputPort]] = None,
+    name: Optional[str] = None,
+    mult_properties: Optional[
+        Union[Dict[str, int], Dict[str, Dict[str, int]]]
+    ] = None,
+    add_properties: Optional[
+        Union[Dict[str, int], Dict[str, Dict[str, int]]]
+    ] = None,
+):
+    r"""
+    Generate a signal flow graph of a transposed direct form FIR filter. The *coefficients* parameter is a
+    sequence of impulse response values::
+
+        coefficients = [h0, h1, h2, ..., hN]
+
+    Leading to the transfer function:
+
+    .. math:: \sum_{i=0}^N h_iz^{-i}
+
+    Parameters
+    ----------
+    coefficients : 1D-array
+        Coefficients to use for the FIR filter section
+
+    input_op : Input, optional
+        The Input to connect the SFG to. If not provided, one will be generated.
+
+    output : Output, optional
+        The Output to connect the SFG to. If not provided, one will be generated.
+
+    name : Name, optional
+        The name of the SFG. If None, "WDF allpass section".
+
+    mult_properties : dictionary, optional
+        Properties passed to :class:`~b_asic.core_operations.ConstantMultiplication`.
+
+    add_properties : dictionary, optional
+        Properties passed to :class:`~b_asic.core_operations.Addition`.
+
+    Returns
+    -------
+        Signal flow graph
+
+    See also
+    --------
+    direct_form_fir
+    """
+    np_coefficients = np.squeeze(np.asarray(coefficients))
+    if np_coefficients.ndim != 1:
+        raise TypeError("coefficients must be a 1D-array")
+    if input_op is None:
+        input_op = Input()
+    if output is None:
+        output = Output()
+    if name is None:
+        name = "Transposed direct-form FIR filter"
+    if mult_properties is None:
+        mult_properties = {}
+    if add_properties is None:
+        add_properties = {}
+
+    taps = len(np_coefficients)
+    prev_delay = None
+    prev_add = None
+    for i, coeff in enumerate(reversed(np_coefficients)):
+        tmp_mul = ConstantMultiplication(coeff, input_op, **mult_properties)
+        if prev_delay is None:
+            tmp_add = tmp_mul
+        else:
+            tmp_add = Addition(tmp_mul, prev_delay, **add_properties)
+        if i < taps - 1:
+            prev_delay = Delay(tmp_add)
+
+    output << tmp_add
+
+    return SFG([input_op], [output], name=Name(name))

docs_sphinx/api/index.rst

@@ -11,7 +11,7 @@ API
     port.rst
     process.rst
     schedule.rst
-    sfg_generator.rst
+    sfg_generators.rst
     signal.rst
     signal_flow_graph.rst
     signal_generator.rst

docs_sphinx/api/sfg_generator.rst

-************************
-``b_asic.sfg_generator``
-************************
-
-.. automodule:: b_asic.sfg_generator
-   :members:

docs_sphinx/api/sfg_generators.rst

+*************************
+``b_asic.sfg_generators``
+*************************
+
+.. automodule:: b_asic.sfg_generators
+   :members:

test/test_sfg_generator.py

-from b_asic.core_operations import SymmetricTwoportAdaptor
-from b_asic.sfg_generator import wdf_allpass
-
-
-def test_wdf_allpass():
-    sfg = wdf_allpass([0.3, 0.5, 0.7])
-    assert (
-        len(
-            [
-                comp
-                for comp in sfg.components
-                if isinstance(comp, SymmetricTwoportAdaptor)
-            ]
-        )
-        == 3
-    )
-
-    sfg = wdf_allpass([0.3, 0.5, 0.7, 0.9])
-    assert (
-        len(
-            [
-                comp
-                for comp in sfg.components
-                if isinstance(comp, SymmetricTwoportAdaptor)
-            ]
-        )
-        == 4
-    )

test/test_sfg_generators.py

+from b_asic.core_operations import (
+    Addition,
+    ConstantMultiplication,
+    SymmetricTwoportAdaptor,
+)
+from b_asic.sfg_generators import (
+    direct_form_fir,
+    transposed_direct_form_fir,
+    wdf_allpass,
+)
+from b_asic.special_operations import Delay
+
+
+def test_wdf_allpass():
+    sfg = wdf_allpass([0.3, 0.5, 0.7])
+    assert (
+        len(
+            [
+                comp
+                for comp in sfg.components
+                if isinstance(comp, SymmetricTwoportAdaptor)
+            ]
+        )
+        == 3
+    )
+
+    sfg = wdf_allpass([0.3, 0.5, 0.7, 0.9])
+    assert (
+        len(
+            [
+                comp
+                for comp in sfg.components
+                if isinstance(comp, SymmetricTwoportAdaptor)
+            ]
+        )
+        == 4
+    )
+
+
+def test_direct_form_fir():
+    sfg = direct_form_fir([0.3, 0.5, 0.7])
+    assert (
+        len(
+            [
+                comp
+                for comp in sfg.components
+                if isinstance(comp, ConstantMultiplication)
+            ]
+        )
+        == 3
+    )
+    assert (
+        len([comp for comp in sfg.components if isinstance(comp, Addition)])
+        == 2
+    )
+    assert (
+        len([comp for comp in sfg.components if isinstance(comp, Delay)]) == 2
+    )
+
+
+def test_transposed_direct_form_fir():
+    sfg = transposed_direct_form_fir([0.3, 0.5, 0.7])
+    assert (
+        len(
+            [
+                comp
+                for comp in sfg.components
+                if isinstance(comp, ConstantMultiplication)
+            ]
+        )
+        == 3
+    )
+    assert (
+        len([comp for comp in sfg.components if isinstance(comp, Addition)])
+        == 2
+    )
+    assert (
+        len([comp for comp in sfg.components if isinstance(comp, Delay)]) == 2
+    )