xwr.rsp.numpy

Radar Signal Processing in Numpy.

Tip

We use pyfftw to perform FFTs, which wraps FFTW. In our testing for computing 64x3x4x256 range-Doppler frames, this provides a ~5x speedup over np.fft.fftn for single frames and a ~10x speedup for batches of 8 frames.

FFTW plans are also cached for efficiency: the first time a particular shape and axes are requested from RSPNumpy.fft, a copy of the array is provided to fftw to create a plan, which is saved by RSPNumpy.

Warning

This module is not automatically imported; you will need to explicitly import it:

from xwr.rsp import numpy as rsp

xwr.rsp.numpy.AWR1642Boost

Bases: RSPNumpy

Radar Signal Processing for the AWR1642 or AWR1843 with TX2 disabled.

Antenna Array

The TI AWR1642Boost (or AWR1843Boost with TX2 disabled) has a 1x8 linear MIMO array:

1-1 1-2 1-3 1-4 2-1 2-2 2-3 2-4

Parameters:

Name	Type	Description	Default
window	bool | dict[Literal['range', 'doppler', 'azimuth', 'elevation'], bool]	whether to apply a hanning window. If bool, the same option is applied to all axes. If dict, specify per axis with keys "range", "doppler", "azimuth", and "elevation".	False
size	dict[Literal['range', 'doppler', 'azimuth', 'elevation'], int]	target size for each axis after zero-padding, specified by axis. If an axis is not spacified, it is not padded.	{}

Source code in src/xwr/rsp/numpy/rsp.py

class AWR1642Boost(RSPNumpy):
    """Radar Signal Processing for the AWR1642 or AWR1843 with TX2 disabled.

    !!! info "Antenna Array"

        The TI AWR1642Boost (or AWR1843Boost with TX2 disabled) has a
        1x8 linear MIMO array:
        ```
        1-1 1-2 1-3 1-4 2-1 2-2 2-3 2-4
        ```

    Args:
        window: whether to apply a hanning window. If `bool`, the same option
            is applied to all axes. If `dict`, specify per axis with keys
            "range", "doppler", "azimuth", and "elevation".
        size: target size for each axis after zero-padding, specified by axis.
            If an axis is not spacified, it is not padded.
    """

    def mimo_virtual_array(
        self, rd: Complex64[np.ndarray, "#batch doppler tx rx range"]
    ) -> Complex64[np.ndarray, "#batch doppler el az range"]:
        batch, doppler, tx, rx, range = rd.shape

        # 1843Boost cast as 1642Boost
        if tx == 3:
            if rx != 4:
                raise ValueError(
                    f"Expected (tx, rx)=3x4 in 1843Boost -> 1642Boost "
                    f"emulation, got tx={tx} and rx={rx}.")
            rd = rd[:, :, [0, 2], :, :]
        else:
            if tx != 2 or rx != 4:
                raise ValueError(
                    f"Expected (tx, rx)=2x4, got tx={tx} and rx={rx}.")

        return rd.reshape(batch, doppler, 1, -1, range)

xwr.rsp.numpy.AWR1843AOP

Bases: RSPNumpy

Radar Signal Processing for AWR1843AOP.

Antenna Array

In the TI AWR1843AOP, the MIMO virtual array is arranged in a 2D grid:

1-1 2-1 3-1   ^
1-2 2-2 3-2   | Up
1-3 2-3 3-3
1-4 2-4 3-4 (TX-RX pairs)

Parameters:

Name	Type	Description	Default
window	bool | dict[Literal['range', 'doppler', 'azimuth', 'elevation'], bool]	whether to apply a hanning window. If bool, the same option is applied to all axes. If dict, specify per axis with keys "range", "doppler", "azimuth", and "elevation".	False
size	dict[Literal['range', 'doppler', 'azimuth', 'elevation'], int]	target size for each axis after zero-padding, specified by axis. If an axis is not spacified, it is not padded.	{}

Source code in src/xwr/rsp/numpy/rsp.py

class AWR1843AOP(RSPNumpy):
    """Radar Signal Processing for AWR1843AOP.

    !!! info "Antenna Array"

        In the TI AWR1843AOP, the MIMO virtual array is arranged in a 2D grid:
            ```
            1-1 2-1 3-1   ^
            1-2 2-2 3-2   | Up
            1-3 2-3 3-3
            1-4 2-4 3-4 (TX-RX pairs)
            ```

    Args:
        window: whether to apply a hanning window. If `bool`, the same option
            is applied to all axes. If `dict`, specify per axis with keys
            "range", "doppler", "azimuth", and "elevation".
        size: target size for each axis after zero-padding, specified by axis.
            If an axis is not spacified, it is not padded.
    """

    def mimo_virtual_array(
        self, rd: Complex64[np.ndarray, "#batch doppler tx rx range"]
    ) -> Complex64[np.ndarray, "#batch doppler el az range"]:
        _, _, tx, rx, _ = rd.shape
        if tx != 3 or rx != 4:
            raise ValueError(
                f"Expected (tx, rx)=3x4, got tx={tx} and rx={rx}.")

        return np.swapaxes(rd, 2, 3)

xwr.rsp.numpy.AWR1843Boost

Bases: RSPNumpy

Radar Signal Processing for AWR1843Boost.

Antenna Array

In the TI AWR1843Boost, the MIMO virtual array has resolution 2x8, with a single 1/2-wavelength elevated middle antenna element:

TX-RX:  2-1 2-2 2-3 2-4           ^
1-1 1-2 1-3 1-4 3-1 3-2 3-3 3-4   | Up

Parameters:

Name	Type	Description	Default
window	bool | dict[Literal['range', 'doppler', 'azimuth', 'elevation'], bool]	whether to apply a hanning window. If bool, the same option is applied to all axes. If dict, specify per axis with keys "range", "doppler", "azimuth", and "elevation".	False
size	dict[Literal['range', 'doppler', 'azimuth', 'elevation'], int]	target size for each axis after zero-padding, specified by axis. If an axis is not spacified, it is not padded.	{}

Source code in src/xwr/rsp/numpy/rsp.py

class AWR1843Boost(RSPNumpy):
    """Radar Signal Processing for AWR1843Boost.

    !!! info "Antenna Array"

        In the TI AWR1843Boost, the MIMO virtual array has resolution 2x8, with
        a single 1/2-wavelength elevated middle antenna element:
        ```
        TX-RX:  2-1 2-2 2-3 2-4           ^
        1-1 1-2 1-3 1-4 3-1 3-2 3-3 3-4   | Up
        ```

    Args:
        window: whether to apply a hanning window. If `bool`, the same option
            is applied to all axes. If `dict`, specify per axis with keys
            "range", "doppler", "azimuth", and "elevation".
        size: target size for each axis after zero-padding, specified by axis.
            If an axis is not spacified, it is not padded.
    """

    def mimo_virtual_array(
        self, rd: Complex64[np.ndarray, "#batch doppler tx rx range"]
    ) -> Complex64[np.ndarray, "#batch doppler el az range"]:
        batch, doppler, tx, rx, range = rd.shape
        if tx != 3 or rx != 4:
            raise ValueError(
                f"Expected (tx, rx)=3x4, got tx={tx} and rx={rx}.")

        mimo = np.zeros((batch, doppler, 2, 8, range), dtype=np.complex64)
        mimo[:, :, 0, 2:6, :] = rd[:, :, 1, :, :]
        mimo[:, :, 1, 0:4, :] = rd[:, :, 0, :, :]
        mimo[:, :, 1, 4:8, :] = rd[:, :, 2, :, :]
        return mimo

xwr.rsp.numpy.RSPNumpy

Bases: RSP[ndarray], ABC

Numpy Radar Signal Processing base class.

Parameters:

Name	Type	Description	Default
window	bool | dict[Literal['range', 'doppler', 'azimuth', 'elevation'], bool]	whether to apply a hanning window. If bool, the same option is applied to all axes. If dict, specify per axis with keys "range", "doppler", "azimuth", and "elevation".	False
size	dict[Literal['range', 'doppler', 'azimuth', 'elevation'], int]	target size for each axis after zero-padding, specified by axis. If an axis is not spacified, it is not padded.	{}

Source code in src/xwr/rsp/numpy/rsp.py

class RSPNumpy(RSP[np.ndarray], ABC):
    """Numpy Radar Signal Processing base class.

    Args:
        window: whether to apply a hanning window. If `bool`, the same option
            is applied to all axes. If `dict`, specify per axis with keys
            "range", "doppler", "azimuth", and "elevation".
        size: target size for each axis after zero-padding, specified by axis.
            If an axis is not spacified, it is not padded.
    """

    def __init__(
        self, window: bool | dict[
            Literal["range", "doppler", "azimuth", "elevation"], bool] = False,
        size: dict[
            Literal["range", "doppler", "azimuth", "elevation"], int] = {}
    ) -> None:
        super().__init__(window=window, size=size)
        self._fft_cache: dict[
            tuple[tuple[int, ...], tuple[int, ...]], FFTW] = {}

    def fft(
        self, array: Complex64[np.ndarray, "..."], axes: tuple[int, ...],
        size: tuple[int, ...] | None = None,
        shift: tuple[int, ...] | None = None
    ) -> Complex64[np.ndarray, "..."]:
        if size is not None:
            for axis, s in zip(axes, size):
                array = self.pad(array, axis, s)

        key = (array.shape, axes)
        if key not in self._fft_cache:
            self._fft_cache[key] = FFTW(
                np.copy(array), np.zeros_like(array), axes=axes)

        fftd = self._fft_cache[key](array)
        return np.fft.fftshift(fftd, axes=shift) if shift else fftd

    @staticmethod
    def pad(
        x: Shaped[np.ndarray, "..."], axis: int, size: int
    ) -> Shaped[np.ndarray, "..."]:
        if size == x.shape[axis]:
            return x
        elif size < x.shape[axis]:
            slices = [slice(None)] * x.ndim
            slices[axis] = slice(0, size)
            return x[tuple(slices)]
        else:
            shape = list(x.shape)
            shape[axis] = size - x.shape[axis]
            zeros = np.zeros(shape, dtype=x.dtype)
            return np.concatenate([x, zeros], axis=axis)

    @staticmethod
    def hann(
        iq: Complex64[np.ndarray, "..."], axis: int
    ) -> Complex64[np.ndarray, "..."]:
        hann = np.hanning(iq.shape[axis] + 2).astype(np.float32)[1:-1]
        broadcast: list[None | slice] = [None] * iq.ndim
        broadcast[axis] = slice(None)
        return iq * (hann / np.mean(hann))[tuple(broadcast)]