Add FIR generators

Closes #152 (closed) #153 (closed)

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,
+):
+    """
+    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))