Skip to content

xwr.rsp

Radar Signal Processing for batched 4D spectrum.

Image Axis Order

Elevation and azimuth axes are in "image order": increasing index is down and to the right, respectively.

Byte order: when do I sample_swap?

If you are using the xwr stack, use the default sample_swap=False, which corresponds to MSB_LSB_IQ, the only option supported by the source-available TI firmware. If processing data collected using other systems (in particular, mmWave studio, which has its own closed-source firmware which supports MSB_LSB_QI), you may need to set sample_swap=True if this option was enabled.

To use the RSP:

  1. Pick your backend. Currently, we support numpy, jax, and pytorch.

    Note

    Each RSP backend is not imported by default; you must explicitly import the backend you want to use, and make sure its dependencies are installed (pytorch, jax, etc). 

    from xwr.rsp import numpy as rsp
# or
from xwr.rsp import jax as rsp

  2. Select the appropriate radar model.

  3. Import the RSP class matching your backend and radar: 

    from xwr.rsp import RSP
from xwr.rsp.torch import AWR1843AOP

rsp: RSP = AWR1843AOP()

    Tip

    Use xwr.rsp.RSP as the type for a generic RSP, and RSP[np.ndarray], RSP[jax.Array], RSP[torch.Tensor], etc for a RSP with a specific backend.

  4. If point cloud processing is desired, use the matching CFAR class for your backend.

    Note

    We currently provide implementations of CA-CFAR and CFAR-CASO for each backend.

xwr.rsp.SignalCube module-attribute 

SignalCube = (
    Float[TArray, "batch doppler tx rx range"]
    | Float[TArray, "batch doppler range"]
    | Float[TArray, "batch doppler el az range"]
)

Accepted detector input. A virtual array cube, an angle spectrum, or an already-combined range-doppler image.

xwr.rsp.CACFAR

Bases: CFAR[TArray], ABC

Cell-averaging CFAR.

    ┌─────────────────┐ ▲ guard[0]+train[0]
    │    ┌───────┐    │ │
    │    │  ┌─┐  │    │ ▼
    │    │  └─┘  │    │ ▲ guard[0]
    │    └───────┘    │ ▼
    └─────────────────┘
guard[1] ◄──► ◄───────► guard[1]+train[1]

Implementation notes:

  • The noise floor is the mean of the training cells in the 2D ring.
  • train can be 0 on at most one axis.
  • A cell is detected when its integrated power exceeds snr_thresh * noise.
Type Parameters
  • TArray: Generic backend, e.g., np.ndarray, jax jax.Array, or torch Tensor.

Parameters:

Name Type Description Default
guard tuple[int, int]

guard cells on each side of the cell under test, for (range, doppler).

 (2, 2)
train tuple[int, int]

training cells on each side of the guard region, for (range, doppler).

 (2, 2)
snr_thresh float

detection threshold, as a linear power ratio (not dB).

 5.0
discard_range tuple[int, int]

range bins (close, far) to discard around DC.

 (10, 20)
Source code in src/xwr/rsp/spectrum.py 
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
class CACFAR(CFAR[TArray], ABC):
    """Cell-averaging CFAR.

    ```
        ┌─────────────────┐ ▲ guard[0]+train[0]
        │    ┌───────┐    │ │
        │    │  ┌─┐  │    │ ▼
        │    │  └─┘  │    │ ▲ guard[0]
        │    └───────┘    │ ▼
        └─────────────────┘
    guard[1] ◄──► ◄───────► guard[1]+train[1]
    ```

    Implementation notes:

    - The noise floor is the mean of the training cells in the 2D ring.
    - `train` can be `0` on at most one axis.
    - A cell is detected when its integrated power exceeds
        `snr_thresh * noise`.

    Type Parameters:
        - `TArray`: Generic backend, e.g., `np.ndarray`, jax `jax.Array`, or
            torch `Tensor`.

    Args:
        guard: guard cells on each side of the cell under test, for
            (range, doppler).
        train: training cells on each side of the guard region, for
            (range, doppler).
        snr_thresh: detection threshold, as a **linear power ratio** (not dB).
        discard_range: range bins (close, far) to discard around DC.
    """

    def __init__(
        self,
        guard: tuple[int, int] = (2, 2),
        train: tuple[int, int] = (2, 2),
        snr_thresh: float = 5.0,
        discard_range: tuple[int, int] = (10, 20),
    ) -> None:
        super().__init__(guard, train, discard_range)
        self.snr_thresh = snr_thresh

        g0, g1 = guard
        w0, w1 = g0 + train[0], g1 + train[1]

        mask = np.ones((2 * w0 + 1, 2 * w1 + 1), dtype=np.float32)
        mask[w0 - g0 : w0 + g0 + 1, w1 - g1 : w1 + g1 + 1] = 0.0
        if mask.sum() == 0:
            raise ValueError(
                f"CFAR mask is empty; check guard={guard} and train={train}.")
        self.mask: Float[np.ndarray, "kr kd"] = mask

xwr.rsp.CASOCFAR

Bases: CFAR[TArray], ABC

Cell-averaging Smallest of CFAR.

Info

Instead of the 2D kernel used in CACFAR, CASO uses a separate 1D kernel for the range and doppler axes, and reports a detection only where both axes fire.

           ┌─┐       ▲ train[0]
           │ │       │
           ├─┤       ▼
     ┌───┬─┼─┼─┬───┐
     └───┴─┼─┼─┴───┘ ▲ guard[0]
           ├─┤       ▼
           │ │
           └─┘
guard[1] ◄─►   ◄───► train[1]

Implementation notes:

  • On each axis, CASO takes the minimum of the two one-sided means, so a strong target on one side cannot inflate the noise floor and mask a weaker target on the other. As such, train must be >= 1 on both axes.
  • An axis fires where its integrated power exceeds that axis's snr_thresh * noise. Both axes must fire for a detection to be reported.
Type Parameters
  • TArray: Generic backend, e.g., np.ndarray, jax jax.Array, or torch Tensor.

Parameters:

Name Type Description Default
guard tuple[int, int]

guard cells on each side of the cell under test, for (range, doppler).

 (8, 0)
train tuple[int, int]

training cells on each side of the guard region, for (range, doppler).

 (8, 4)
snr_thresh tuple[float, float]

detection threshold for (range, doppler), as a linear power ratio (not dB).

 (5.0, 3.0)
discard_range tuple[int, int]

range bins (close, far) to discard around DC.

 (10, 20)
Source code in src/xwr/rsp/spectrum.py 
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
class CASOCFAR(CFAR[TArray], ABC):
    """Cell-averaging Smallest of CFAR.

    !!! info

        Instead of the 2D kernel used in [`CACFAR`][xwr.rsp.], CASO uses a
        separate 1D kernel for the range and doppler axes, and reports a
        detection only where **both** axes fire.

    ```
               ┌─┐       ▲ train[0]
               │ │       │
               ├─┤       ▼
         ┌───┬─┼─┼─┬───┐
         └───┴─┼─┼─┴───┘ ▲ guard[0]
               ├─┤       ▼
               │ │
               └─┘
    guard[1] ◄─►   ◄───► train[1]
    ```

    Implementation notes:

    - On each axis, CASO takes the **minimum** of the two one-sided means, so
        a strong target on one side cannot inflate the noise floor and mask a
        weaker target on the other. As such, `train` must be `>= 1` on both axes.
    - An axis fires where its integrated power exceeds that axis's
        `snr_thresh * noise`. Both axes must fire for a detection to be reported.

    Type Parameters:
        - `TArray`: Generic backend, e.g., `np.ndarray`, jax `jax.Array`, or
            torch `Tensor`.

    Args:
        guard: guard cells on each side of the cell under test, for
            (range, doppler).
        train: training cells on each side of the guard region, for
            (range, doppler).
        snr_thresh: detection threshold for (range, doppler), as a **linear
            power ratio** (not dB).
        discard_range: range bins (close, far) to discard around DC.
    """

    def __init__(
        self,
        guard: tuple[int, int] = (8, 0),
        train: tuple[int, int] = (8, 4),
        snr_thresh: tuple[float, float] = (5.0, 3.0),
        discard_range: tuple[int, int] = (10, 20),
    ) -> None:
        super().__init__(guard, train, discard_range)
        if any(t < 1 for t in train):
            raise ValueError(f"Train {train} must be >= 1 on each axis.")

        self.snr_r, self.snr_d = snr_thresh

        self.train_r, self.train_d = train
        self.pad_r = train[0] + guard[0]
        self.pad_d = train[1] + guard[1]

xwr.rsp.CFAR

Bases: ABC, Generic[TArray]

Abstract, backend-agnostic CFAR base class.

The following implementations are available:

Detector Noise estimate
CACFAR 2D cell-averaging ring
CASOCFAR Separate "smallest of" tests on the range and doppler axes
Type Parameters
  • TArray: Generic backend, e.g., np.ndarray, jax jax.Array, or torch Tensor.

Parameters:

Name Type Description Default
guard tuple[int, int]

guard cells on each side of the cell under test, for (range, doppler).

 required
train tuple[int, int]

training cells on each side of the guard region, for (range, doppler).

 required
discard_range tuple[int, int]

range bins (close, far) to discard around DC.

 (10, 20)
Source code in src/xwr/rsp/spectrum.py 
 41
 42
 43
 44
 45
 46
 47
 48
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
class CFAR(ABC, Generic[TArray]):
    """Abstract, backend-agnostic CFAR base class.

    The following implementations are available:

    | Detector | Noise estimate |
    |----------|----------------|
    | [`CACFAR`][xwr.rsp.] | 2D cell-averaging ring |
    | [`CASOCFAR`][xwr.rsp.] | Separate "smallest of" tests on the range and doppler axes |

    Type Parameters:
        - `TArray`: Generic backend, e.g., `np.ndarray`, jax `jax.Array`, or
            torch `Tensor`.

    Args:
        guard: guard cells on each side of the cell under test, for
            (range, doppler).
        train: training cells on each side of the guard region, for
            (range, doppler).
        discard_range: range bins (close, far) to discard around DC.
    """

    def __init__(
        self,
        guard: tuple[int, int],
        train: tuple[int, int],
        discard_range: tuple[int, int] = (10, 20),
    ) -> None:
        if any(g < 0 for g in guard):
            raise ValueError(f"Guard {guard} must be >= 0 on each axis.")
        if any(t < 0 for t in train):
            raise ValueError(f"Train {train} must be >= 0 on each axis.")
        if any(d < 0 for d in discard_range):
            raise ValueError(
                f"Discard range {discard_range} must be >= 0 on each side.")

        self.guard = guard
        self.train = train
        self.discard_r = discard_range

    @abstractmethod
    def _cfar(
        self, signal_cube: Float[TArray, "batch doppler channel range"]
    ) -> Detection[TArray]:
        """Run this detector on a batch of radar cubes.

        Args:
            signal_cube: batch of post range doppler FFT radar cubes in
                amplitude, with the channel axes already flattened into a
                single axis.

        Returns:
            The same detections as [`__call__`][xwr.rsp.CFAR.__call__].
        """
        ...

    def _flatten(
        self, signal_cube: SignalCube
    ) -> Float[TArray, "batch doppler channel range"]:
        """Collapse any axes between doppler and range into a channel axis.

        Args:
            signal_cube: batch of post range doppler FFT radar cubes in
                amplitude.

        Returns:
            The same values, with a single channel axis; a range-doppler
                image gains a channel axis of length 1.
        """
        batch, doppler, rng = (
            signal_cube.shape[0], signal_cube.shape[1], signal_cube.shape[-1])
        return cast(Any, signal_cube).reshape((batch, doppler, -1, rng))

    def __call__(self, signal_cube: SignalCube) -> Detection[TArray]:
        """Run 2D CFAR detection.

        !!! note

            The channel axes (the transmit and receive antennas of the
            virtual array, or the elevation and azimuth bins of an angle
            spectrum) are combined non-coherently, so their relative order
            does not matter.

        Args:
            signal_cube: batch of post range doppler FFT radar cubes in
                amplitude, or a range-doppler image which is already combined
                across the virtual array.

        Returns:
            The detection mask, the range-doppler spectrum it was computed
                from, and the signal to noise ratio.
        """
        return self._cfar(self._flatten(signal_cube))

__call__ 

__call__(signal_cube: SignalCube) -> Detection[TArray]

Run 2D CFAR detection.

Note

The channel axes (the transmit and receive antennas of the virtual array, or the elevation and azimuth bins of an angle spectrum) are combined non-coherently, so their relative order does not matter.

Parameters:

Name Type Description Default
signal_cube SignalCube

batch of post range doppler FFT radar cubes in amplitude, or a range-doppler image which is already combined across the virtual array.

 required

Returns:

Type Description
Detection[TArray]

The detection mask, the range-doppler spectrum it was computed from, and the signal to noise ratio.
Source code in src/xwr/rsp/spectrum.py 
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
def __call__(self, signal_cube: SignalCube) -> Detection[TArray]:
    """Run 2D CFAR detection.

    !!! note

        The channel axes (the transmit and receive antennas of the
        virtual array, or the elevation and azimuth bins of an angle
        spectrum) are combined non-coherently, so their relative order
        does not matter.

    Args:
        signal_cube: batch of post range doppler FFT radar cubes in
            amplitude, or a range-doppler image which is already combined
            across the virtual array.

    Returns:
        The detection mask, the range-doppler spectrum it was computed
            from, and the signal to noise ratio.
    """
    return self._cfar(self._flatten(signal_cube))

xwr.rsp.DensePoints dataclass

Bases: Generic[TArray]

A dense radar point cloud, and the points in it which are valid.

Type Parameters
  • TArray: Generic backend, e.g., np.ndarray, jax jax.Array, or torch Tensor.

Attributes:

Name Type Description
mask Bool[TArray, 'batch range doppler']

mask of valid points, i.e. the CFAR detection mask combined with the angular bounds set by angle_fov.
points Float32[TArray, 'batch range doppler 4']

all possible radar points, where the trailing axis holds (x, y, z, v).
Source code in src/xwr/rsp/aoa.py 
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
@dataclass
class DensePoints(Generic[TArray]):
    """A dense radar point cloud, and the points in it which are valid.

    Type Parameters:
        - `TArray`: Generic backend, e.g., `np.ndarray`, jax `jax.Array`, or
            torch `Tensor`.

    Attributes:
        mask: mask of valid points, i.e. the CFAR detection mask combined with
            the angular bounds set by `angle_fov`.
        points: all possible radar points, where the trailing axis holds
            `(x, y, z, v)`.
    """

    mask: Bool[TArray, "batch range doppler"]
    points: Float32[TArray, "batch range doppler 4"]

xwr.rsp.Detection dataclass

Bases: Generic[TArray]

Detected objects, and the statistics they were detected from.

Type Parameters
  • TArray: Generic backend, e.g., np.ndarray, jax jax.Array, or torch Tensor.

Attributes:

Name Type Description
mask Bool[TArray, 'batch range doppler']

cfar detected object mask.
signal Float[TArray, 'batch range doppler']

non-coherently integrated power across the channel axes, i.e. the range-doppler spectrum used for detection.
snr Float[TArray, 'batch range doppler']

signal to noise ratio, as a linear power ratio.
Source code in src/xwr/rsp/spectrum.py 
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
@dataclass
class Detection(Generic[TArray]):
    """Detected objects, and the statistics they were detected from.

    Type Parameters:
        - `TArray`: Generic backend, e.g., `np.ndarray`, jax `jax.Array`, or
            torch `Tensor`.

    Attributes:
        mask: cfar detected object mask.
        signal: non-coherently integrated power across the channel axes, i.e.
            the range-doppler spectrum used for detection.
        snr: signal to noise ratio, as a linear power ratio.
    """

    mask: Bool[TArray, "batch range doppler"]
    signal: Float[TArray, "batch range doppler"]
    snr: Float[TArray, "batch range doppler"]

xwr.rsp.PointCloud

Bases: ABC, Generic[TArray]

Get radar point cloud from post FFT cube.

To convert azimuth-elevation bin indices to azimuth-elevation angles, we use the property that the azimuth bin indices correspond to the sin of the angle 

angles = np.arcsin(
    np.clip(np.linspace(-1.0, 1.0, bin_size) / (2 * antenna_spacing),
            -1.0, 1.0)
)
where the corrected antenna spacing is calculated by 
0.5 * chirp_center_frequency / antenna_design_frequency

Info

The antenna design frequency here refers to the grid alignment of the antenna array, which are typically 0.5 wavelengths apart at some nominal design frequency. Thus, you must correct by a corresponding scale factor when the chirp center frequency differs.

Implementation notes:

  • With the default range and dopplerresolutions of 1.0, the point cloud is in range/doppler bins instead of meters and meters/second.
  • Radars have little resolving power close to the array plane in angle, and suffer from high noise at the edge of the main lobe. Reject these points with angle_fov.
Type Parameters
  • TArray: Generic backend, e.g., np.ndarray, jax jax.Array, or torch Tensor.

Parameters:

Name Type Description Default
range_res float

range resolution, i.e. meters per range bin; see XWRConfig.range_resolution. Defaults to 1.0, which leaves the range axis in bins.

 1.0
doppler_res float

doppler resolution, i.e. meters/second per doppler bin; see XWRConfig.doppler_resolution. Defaults to 1.0, which leaves the doppler axis in bins.

 1.0
angle_fov tuple[float, float]

(elevation, azimuth) field of view in degrees, where points outside +/-elevation and +/-azimuth are rejected.

 (20.0, 80.0)
antenna_spacing float

antenna spacing in terms of wavelength (default 0.5 for a half-wavelength grid).

 0.5
Source code in src/xwr/rsp/aoa.py 
 32
 33
 34
 35
 36
 37
 38
 39
 40
 41
 42
 43
 44
 45
 46
 47
 48
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
class PointCloud(ABC, Generic[TArray]):
    """Get radar point cloud from post FFT cube.

    To convert azimuth-elevation bin indices to azimuth-elevation angles,
    we use the property that the azimuth bin indices correspond to the sin of
    the angle
    ```
    angles = np.arcsin(
        np.clip(np.linspace(-1.0, 1.0, bin_size) / (2 * antenna_spacing),
                -1.0, 1.0)
    )
    ```
    where the *corrected* antenna spacing is calculated by
    ```
    0.5 * chirp_center_frequency / antenna_design_frequency
    ```

    !!! info

        The antenna design frequency here refers to the grid alignment of the
        antenna array, which are typically 0.5 wavelengths apart at some
        nominal design frequency. Thus, you must correct by a corresponding
        scale factor when the chirp center frequency differs.

    Implementation notes:

    -  With the default range and dopplerresolutions of `1.0`, the point cloud
        is in range/doppler bins instead of meters and meters/second.
    - Radars have little resolving power close to the array plane in angle,
        and suffer from high noise at the edge of the main lobe. Reject these
        points with `angle_fov`.

    Type Parameters:
        - `TArray`: Generic backend, e.g., `np.ndarray`, jax `jax.Array`, or
            torch `Tensor`.

    Args:
        range_res: range resolution, i.e. meters per range bin; see
            [`XWRConfig.range_resolution`][xwr.config.XWRConfig]. Defaults
            to `1.0`, which leaves the range axis in bins.
        doppler_res: doppler resolution, i.e. meters/second per doppler bin;
            see [`XWRConfig.doppler_resolution`][xwr.config.XWRConfig].
            Defaults to `1.0`, which leaves the doppler axis in bins.
        angle_fov: (elevation, azimuth) field of view **in degrees**, where points
            outside +/-elevation and +/-azimuth are rejected.
        antenna_spacing: antenna spacing in terms of wavelength (default 0.5
            for a half-wavelength grid).
    """

    def __init__(
        self,
        range_res: float = 1.0,
        doppler_res: float = 1.0,
        angle_fov: tuple[float, float] = (20.0, 80.0),
        antenna_spacing: float = 0.5,
    ) -> None:
        self.range_res = range_res
        self.doppler_res = doppler_res

        if antenna_spacing <= 0:
            raise ValueError("antenna_spacing must be > 0")
        self.antenna_spacing = antenna_spacing
        self.el_fov = float(np.deg2rad(angle_fov[0]))
        self.az_fov = float(np.deg2rad(angle_fov[1]))

    def _angle_table(self, n: int) -> Float32[np.ndarray, " n"]:
        """Bin-to-angle lookup for an angle axis of length `n`.

        Args:
            n: length of the angle axis, i.e. the angle fft size.

        Returns:
            Angle in radians for each bin, ascending.
        """
        sin = np.clip(
            np.linspace(-1.0, 1.0, n) / (2 * self.antenna_spacing), -1.0, 1.0)
        return np.arcsin(sin).astype(np.float32)

    @abstractmethod
    def aoa(
        self, cube: Float32[TArray, "batch range doppler el az"]
    ) -> Int[TArray, "batch range doppler 2"]:
        """Angle of arrival estimation.

        Takes the argmax over the (elevation, azimuth) angle spectrum of each
        range-doppler bin, yielding **bin indices** into the angle axes
        rather than angles.

        !!! warning

            This assumes at most one scatterer per range-doppler cell; if two
            targets share a range and velocity, only the stronger is reported.

        Args:
            cube: batch of post fft spectrum amplitudes, with the angle axes
                trailing.

        Returns:
            ang: detect angle index for every range doppler bin, as
                `(elevation, azimuth)` along the trailing axis.
        """
        ...

    @abstractmethod
    def __call__(
        self,
        cube: Float32[TArray, "batch doppler el az range"],
        mask: Bool[TArray, "batch range doppler"],
    ) -> DensePoints[TArray]:
        """Get point cloud from radar cube and detection mask.

        !!! note

            The returned point cloud is **dense**: every range-doppler bin
            yields a point, and the caller is expected to gather the valid
            ones using the returned mask (e.g. `pc[pc_mask]`).

        Implementation notes:

        - Return points are multiplied by `range_res` and `doppler_res` to convert
            from bins to meters and meters/second, respectively.
        - Points position compute as `x = r cos(-az) cos(el)`,
            `y = r sin(-az) cos(el)`, and `z = r sin(el)`

        Args:
            cube: batch of post fft spectrum amplitudes.
            mask: CFAR detection mask.

        Returns:
            The dense point cloud, and the mask of points in it which are
                valid; see [`DensePoints`][xwr.rsp.].
        """
        ...

__call__ abstractmethod 

__call__(
    cube: Float32[TArray, "batch doppler el az range"],
    mask: Bool[TArray, "batch range doppler"],
) -> DensePoints[TArray]

Get point cloud from radar cube and detection mask.

Note

The returned point cloud is dense: every range-doppler bin yields a point, and the caller is expected to gather the valid ones using the returned mask (e.g. pc[pc_mask]).

Implementation notes:

  • Return points are multiplied by range_res and doppler_res to convert from bins to meters and meters/second, respectively.
  • Points position compute as x = r cos(-az) cos(el), y = r sin(-az) cos(el), and z = r sin(el)

Parameters:

Name Type Description Default
cube Float32[TArray, 'batch doppler el az range']

batch of post fft spectrum amplitudes.

 required
mask Bool[TArray, 'batch range doppler']

CFAR detection mask.

 required

Returns:

Type Description
DensePoints[TArray]

The dense point cloud, and the mask of points in it which are valid; see DensePoints.
Source code in src/xwr/rsp/aoa.py 
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
@abstractmethod
def __call__(
    self,
    cube: Float32[TArray, "batch doppler el az range"],
    mask: Bool[TArray, "batch range doppler"],
) -> DensePoints[TArray]:
    """Get point cloud from radar cube and detection mask.

    !!! note

        The returned point cloud is **dense**: every range-doppler bin
        yields a point, and the caller is expected to gather the valid
        ones using the returned mask (e.g. `pc[pc_mask]`).

    Implementation notes:

    - Return points are multiplied by `range_res` and `doppler_res` to convert
        from bins to meters and meters/second, respectively.
    - Points position compute as `x = r cos(-az) cos(el)`,
        `y = r sin(-az) cos(el)`, and `z = r sin(el)`

    Args:
        cube: batch of post fft spectrum amplitudes.
        mask: CFAR detection mask.

    Returns:
        The dense point cloud, and the mask of points in it which are
            valid; see [`DensePoints`][xwr.rsp.].
    """
    ...

aoa abstractmethod 

aoa(
    cube: Float32[TArray, "batch range doppler el az"],
) -> Int[TArray, "batch range doppler 2"]

Angle of arrival estimation.

Takes the argmax over the (elevation, azimuth) angle spectrum of each range-doppler bin, yielding bin indices into the angle axes rather than angles.

Warning

This assumes at most one scatterer per range-doppler cell; if two targets share a range and velocity, only the stronger is reported.

Parameters:

Name Type Description Default
cube Float32[TArray, 'batch range doppler el az']

batch of post fft spectrum amplitudes, with the angle axes trailing.

 required

Returns:

Name Type Description
ang Int[TArray, 'batch range doppler 2']

detect angle index for every range doppler bin, as (elevation, azimuth) along the trailing axis.
Source code in src/xwr/rsp/aoa.py 
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
@abstractmethod
def aoa(
    self, cube: Float32[TArray, "batch range doppler el az"]
) -> Int[TArray, "batch range doppler 2"]:
    """Angle of arrival estimation.

    Takes the argmax over the (elevation, azimuth) angle spectrum of each
    range-doppler bin, yielding **bin indices** into the angle axes
    rather than angles.

    !!! warning

        This assumes at most one scatterer per range-doppler cell; if two
        targets share a range and velocity, only the stronger is reported.

    Args:
        cube: batch of post fft spectrum amplitudes, with the angle axes
            trailing.

    Returns:
        ang: detect angle index for every range doppler bin, as
            `(elevation, azimuth)` along the trailing axis.
    """
    ...

xwr.rsp.RSP

Bases: ABC, Generic[TArray]

Abstract, backend-agnostic Radar Signal Processing base class.

Info

This class documents the public interface for all radar signal processing (RSP) classes, except where otherwise noted.

Type Parameters
  • TArray: Generic backend, e.g., np.ndarray, jax jax.Array, or torch Tensor.

Parameters:

Name Type Description Default
window bool | Mapping[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 Mapping[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.

 {}
sample_swap bool

if True, swap the I and Q components when un-interleaving IIQQ data.

 False
Source code in src/xwr/rsp/generic.py 
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317
318
319
320
321
322
323
324
325
326
327
328
329
330
331
332
333
334
335
336
337
338
339
340
341
342
343
344
345
346
347
348
349
350
351
352
353
354
355
356
357
358
359
360
361
362
363
364
365
366
367
368
369
370
371
372
373
374
375
376
377
378
379
380
381
382
383
384
385
386
387
388
389
390
391
392
393
394
395
class RSP(ABC, Generic[TArray]):
    """Abstract, backend-agnostic Radar Signal Processing base class.

    !!! info

        This class documents the public interface for all radar signal
        processing (RSP) classes, except where otherwise noted.

    Type Parameters:
        - `TArray`: Generic backend, e.g., `np.ndarray`, jax `jax.Array`, or
            torch `Tensor`.

    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.
        sample_swap: if `True`, swap the I and Q components when
            un-interleaving IIQQ data.
    """

    SAMPLE_TYPE: Literal["IQ", "I"] = "IQ"

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

        if isinstance(window, bool):
            self.window = {}
            self._default_window = window
        else:
            self.window = dict(window)
            self._default_window = False

        self.size = size
        self.sample_swap = sample_swap

    @abstractmethod
    def fft(
        self, array: Complex64[TArray, "..."] | Float32[TArray, "..."],
        axes: tuple[int, ...],
        size: tuple[int, ...] | None = None,
        shift: tuple[int, ...] | None = None
    ) -> Complex64[TArray, "..."]:
        """Compute FFT on the specified axes of the array.

        Args:
            array: Input array.
            size: Target size for each axis after FFT (or `None` to use the
                input size).
            axes: Axes along which to compute the FFT.
            shift: Axes to shift after FFT, if any.

        Returns:
            FFT of the input array along the specified axes. If the input
                array is real-valued, the output is the non-negative frequency
                terms of the FFT along the specified axes (with length
                `n // 2 + 1`).
        """
        ...

    @staticmethod
    @abstractmethod
    def hann(
        x: Complex64[TArray, "..."] | Float32[TArray, "..."], axis: int
    ) -> Complex64[TArray, "..."] | Float32[TArray, "..."]:
        """Apply a Hann window to the specified axis of the time signal data.

        Args:
            x: time signal data.
            axis: Axis along which to apply the Hann window.

        Returns:
            Time signal data with the Hann window applied along the specified
                axis.
        """
        ...

    def doppler_range(
        self, x: Complex64[TArray, "#batch doppler tx rx range"]
            | Float32[TArray, "#batch doppler tx rx range"]
    ) -> Complex64[TArray, "#batch doppler2 tx rx range2"]:
        """Calculate range-doppler spectrum from time signal data.

        Args:
            x: IQ (complex64) or in-phase-only (float32) data.

        Returns:
            Computed range-doppler spectrum, with windowing if specified.
        """
        if self.window.get("range", self._default_window):
            x = self.hann(x, 4)
        if self.window.get("doppler", self._default_window):
            x = self.hann(x, 1)

        if self.SAMPLE_TYPE == "I":
            # Double range bins for in-phase-only.
            nrange = x.shape[-1] // 2
            range_bins = self.size.get("range", nrange) * 2
        else:
            range_bins = self.size.get("range", x.shape[4])

        rd = self.fft(
            x, axes=(1, 4), shift=(1,),
            size=(self.size.get("doppler", x.shape[1]), range_bins))

        return rd

    @abstractmethod
    def mimo_virtual_array(
        self, rd: Complex64[TArray, "#batch doppler tx rx range"]
    ) -> Complex64[TArray, "#batch doppler elevation azimuth range"]:
        """Set up MIMO virtual array from range-doppler spectrum.

        Args:
            rd: complex range-doppler spectrum.

        Returns:
            Computed MIMO virtual array, in elevation-azimuth order.
        """
        ...

    def elevation_azimuth(
        self, rd: Complex64[TArray, "#batch doppler tx rx range"]
    ) -> Complex64[TArray, "#batch doppler el az range"]:
        """Calculate elevation-azimuth spectrum from range-doppler spectrum.

        Args:
            rd: range-doppler spectrum.

        Returns:
            Computed elevation-azimuth spectrum, with windowing and padding if
                specified.
        """
        mimo = self.mimo_virtual_array(rd)

        if self.window.get("elevation", self._default_window):
            mimo = self.hann(mimo, 2)
        if self.window.get("azimuth", self._default_window):
            mimo = self.hann(mimo, 3)

        return self.fft(
            mimo, axes=(2, 3), shift=(2, 3),
            size=(
                self.size.get("elevation", mimo.shape[2]),
                self.size.get("azimuth", mimo.shape[3])))

    def __call__(
        self, x: Complex64[TArray, "#batch doppler tx rx _range"]
            | Float32[TArray, "#batch doppler tx rx _range"]
            | Int16[TArray, "#batch doppler tx rx _range"]
    ) -> Complex64[TArray, "#batch doppler2 el az _range"]:
        """Process time signal data to compute elevation-azimuth spectrum.

        Args:
            x: IQ data in complex or interleaved int16 IQ format, or
                in-phase-only data in float32 format.

        Returns:
            Computed doppler-elevation-azimuth-range spectrum.
        """
        for i, size in enumerate(x.shape):
            if size == 0:
                raise ValueError(
                    f"Input array has zero-length dimension {i}: {x.shape}")

        if self.SAMPLE_TYPE == "IQ":
            x = iq_from_iiqq(x, sample_swap=self.sample_swap)
        else:
            x = _to_float32(x)

        dr = self.doppler_range(x)
        drae = self.elevation_azimuth(dr)
        return drae

__call__ 

__call__(
    x: Complex64[TArray, "#batch doppler tx rx _range"]
    | Float32[TArray, "#batch doppler tx rx _range"]
    | Int16[TArray, "#batch doppler tx rx _range"],
) -> Complex64[TArray, "#batch doppler2 el az _range"]

Process time signal data to compute elevation-azimuth spectrum.

Parameters:

Name Type Description Default
x Complex64[TArray, '#batch doppler tx rx _range'] | Float32[TArray, '#batch doppler tx rx _range'] | Int16[TArray, '#batch doppler tx rx _range']

IQ data in complex or interleaved int16 IQ format, or in-phase-only data in float32 format.

 required

Returns:

Type Description
Complex64[TArray, '#batch doppler2 el az _range']

Computed doppler-elevation-azimuth-range spectrum.
Source code in src/xwr/rsp/generic.py 
369
370
371
372
373
374
375
376
377
378
379
380
381
382
383
384
385
386
387
388
389
390
391
392
393
394
395
def __call__(
    self, x: Complex64[TArray, "#batch doppler tx rx _range"]
        | Float32[TArray, "#batch doppler tx rx _range"]
        | Int16[TArray, "#batch doppler tx rx _range"]
) -> Complex64[TArray, "#batch doppler2 el az _range"]:
    """Process time signal data to compute elevation-azimuth spectrum.

    Args:
        x: IQ data in complex or interleaved int16 IQ format, or
            in-phase-only data in float32 format.

    Returns:
        Computed doppler-elevation-azimuth-range spectrum.
    """
    for i, size in enumerate(x.shape):
        if size == 0:
            raise ValueError(
                f"Input array has zero-length dimension {i}: {x.shape}")

    if self.SAMPLE_TYPE == "IQ":
        x = iq_from_iiqq(x, sample_swap=self.sample_swap)
    else:
        x = _to_float32(x)

    dr = self.doppler_range(x)
    drae = self.elevation_azimuth(dr)
    return drae

doppler_range 

doppler_range(
    x: Complex64[TArray, "#batch doppler tx rx range"]
    | Float32[TArray, "#batch doppler tx rx range"],
) -> Complex64[TArray, "#batch doppler2 tx rx range2"]

Calculate range-doppler spectrum from time signal data.

Parameters:

Name Type Description Default
x Complex64[TArray, '#batch doppler tx rx range'] | Float32[TArray, '#batch doppler tx rx range']

IQ (complex64) or in-phase-only (float32) data.

 required

Returns:

Type Description
Complex64[TArray, '#batch doppler2 tx rx range2']

Computed range-doppler spectrum, with windowing if specified.
Source code in src/xwr/rsp/generic.py 
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317
318
319
320
321
322
323
324
325
326
327
328
def doppler_range(
    self, x: Complex64[TArray, "#batch doppler tx rx range"]
        | Float32[TArray, "#batch doppler tx rx range"]
) -> Complex64[TArray, "#batch doppler2 tx rx range2"]:
    """Calculate range-doppler spectrum from time signal data.

    Args:
        x: IQ (complex64) or in-phase-only (float32) data.

    Returns:
        Computed range-doppler spectrum, with windowing if specified.
    """
    if self.window.get("range", self._default_window):
        x = self.hann(x, 4)
    if self.window.get("doppler", self._default_window):
        x = self.hann(x, 1)

    if self.SAMPLE_TYPE == "I":
        # Double range bins for in-phase-only.
        nrange = x.shape[-1] // 2
        range_bins = self.size.get("range", nrange) * 2
    else:
        range_bins = self.size.get("range", x.shape[4])

    rd = self.fft(
        x, axes=(1, 4), shift=(1,),
        size=(self.size.get("doppler", x.shape[1]), range_bins))

    return rd

elevation_azimuth 

elevation_azimuth(
    rd: Complex64[TArray, "#batch doppler tx rx range"],
) -> Complex64[TArray, "#batch doppler el az range"]

Calculate elevation-azimuth spectrum from range-doppler spectrum.

Parameters:

Name Type Description Default
rd Complex64[TArray, '#batch doppler tx rx range']

range-doppler spectrum.

 required

Returns:

Type Description
Complex64[TArray, '#batch doppler el az range']

Computed elevation-azimuth spectrum, with windowing and padding if specified.
Source code in src/xwr/rsp/generic.py 
344
345
346
347
348
349
350
351
352
353
354
355
356
357
358
359
360
361
362
363
364
365
366
367
def elevation_azimuth(
    self, rd: Complex64[TArray, "#batch doppler tx rx range"]
) -> Complex64[TArray, "#batch doppler el az range"]:
    """Calculate elevation-azimuth spectrum from range-doppler spectrum.

    Args:
        rd: range-doppler spectrum.

    Returns:
        Computed elevation-azimuth spectrum, with windowing and padding if
            specified.
    """
    mimo = self.mimo_virtual_array(rd)

    if self.window.get("elevation", self._default_window):
        mimo = self.hann(mimo, 2)
    if self.window.get("azimuth", self._default_window):
        mimo = self.hann(mimo, 3)

    return self.fft(
        mimo, axes=(2, 3), shift=(2, 3),
        size=(
            self.size.get("elevation", mimo.shape[2]),
            self.size.get("azimuth", mimo.shape[3])))

fft abstractmethod 

fft(
    array: Complex64[TArray, ...] | Float32[TArray, ...],
    axes: tuple[int, ...],
    size: tuple[int, ...] | None = None,
    shift: tuple[int, ...] | None = None,
) -> Complex64[TArray, ...]

Compute FFT on the specified axes of the array.

Parameters:

Name Type Description Default
array Complex64[TArray, ...] | Float32[TArray, ...]

Input array.

 required
size tuple[int, ...] | None

Target size for each axis after FFT (or None to use the input size).

 None
axes tuple[int, ...]

Axes along which to compute the FFT.

 required
shift tuple[int, ...] | None

Axes to shift after FFT, if any.

 None

Returns:

Type Description
Complex64[TArray, ...]

FFT of the input array along the specified axes. If the input array is real-valued, the output is the non-negative frequency terms of the FFT along the specified axes (with length n // 2 + 1).
Source code in src/xwr/rsp/generic.py 
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
@abstractmethod
def fft(
    self, array: Complex64[TArray, "..."] | Float32[TArray, "..."],
    axes: tuple[int, ...],
    size: tuple[int, ...] | None = None,
    shift: tuple[int, ...] | None = None
) -> Complex64[TArray, "..."]:
    """Compute FFT on the specified axes of the array.

    Args:
        array: Input array.
        size: Target size for each axis after FFT (or `None` to use the
            input size).
        axes: Axes along which to compute the FFT.
        shift: Axes to shift after FFT, if any.

    Returns:
        FFT of the input array along the specified axes. If the input
            array is real-valued, the output is the non-negative frequency
            terms of the FFT along the specified axes (with length
            `n // 2 + 1`).
    """
    ...

hann abstractmethod staticmethod 

hann(
    x: Complex64[TArray, ...] | Float32[TArray, ...], axis: int
) -> Complex64[TArray, ...] | Float32[TArray, ...]

Apply a Hann window to the specified axis of the time signal data.

Parameters:

Name Type Description Default
x Complex64[TArray, ...] | Float32[TArray, ...]

time signal data.

 required
axis int

Axis along which to apply the Hann window.

 required

Returns:

Type Description
Complex64[TArray, ...] | Float32[TArray, ...]

Time signal data with the Hann window applied along the specified axis.
Source code in src/xwr/rsp/generic.py 
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
@staticmethod
@abstractmethod
def hann(
    x: Complex64[TArray, "..."] | Float32[TArray, "..."], axis: int
) -> Complex64[TArray, "..."] | Float32[TArray, "..."]:
    """Apply a Hann window to the specified axis of the time signal data.

    Args:
        x: time signal data.
        axis: Axis along which to apply the Hann window.

    Returns:
        Time signal data with the Hann window applied along the specified
            axis.
    """
    ...

mimo_virtual_array abstractmethod 

mimo_virtual_array(
    rd: Complex64[TArray, "#batch doppler tx rx range"],
) -> Complex64[TArray, "#batch doppler elevation azimuth range"]

Set up MIMO virtual array from range-doppler spectrum.

Parameters:

Name Type Description Default
rd Complex64[TArray, '#batch doppler tx rx range']

complex range-doppler spectrum.

 required

Returns:

Type Description
Complex64[TArray, '#batch doppler elevation azimuth range']

Computed MIMO virtual array, in elevation-azimuth order.
Source code in src/xwr/rsp/generic.py 
330
331
332
333
334
335
336
337
338
339
340
341
342
@abstractmethod
def mimo_virtual_array(
    self, rd: Complex64[TArray, "#batch doppler tx rx range"]
) -> Complex64[TArray, "#batch doppler elevation azimuth range"]:
    """Set up MIMO virtual array from range-doppler spectrum.

    Args:
        rd: complex range-doppler spectrum.

    Returns:
        Computed MIMO virtual array, in elevation-azimuth order.
    """
    ...

xwr.rsp.iq_from_iiqq 

iq_from_iiqq(
    iiqq: Int16[TArray, "... n"] | Complex64[TArray, "... _n"],
    sample_swap: bool = False,
) -> Complex64[TArray, "... n2"]

Un-interleave IIQQ data.

Info

The default sample_swap = False corresponds to the MSB_LSB_IQ byte order used by xwr.

In this case, MSB_LSB_IQ means that I is the MSB and Q is the LSB. However, the data stream is little-endian, which means Q actually comes before I, leading to the actual physical layout being QQII and so on.

Type Parameters
  • TArray: This function is multi-backend, and supports numpy np.ndarray, jax jax.Array, and torch Tensor.

Parameters:

Name Type Description Default
iiqq Int16[TArray, '... n'] | Complex64[TArray, '... _n']

interleaved IIQQ data; see RadarFrame. If already complex, leave it as is.

 required
sample_swap bool

if True, swap the I and Q components so that the output is Q + j*I instead of I + j*Q.

 False

Returns:

Type Description
Complex64[TArray, '... n2']

Complex IQ data.
Source code in src/xwr/rsp/generic.py 
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
def iq_from_iiqq(
    iiqq: Int16[TArray, "... n"] | Complex64[TArray, "... _n"],
    sample_swap: bool = False,
) -> Complex64[TArray, "... n2"]:
    """Un-interleave IIQQ data.

    !!! info

        The default `sample_swap = False` corresponds to the
        [`MSB_LSB_IQ`][xwr.radar.defines.SampleSwap] byte order used by `xwr`.

        In this case, `MSB_LSB_IQ` means that I is the MSB and Q is the LSB.
        However, the data stream is little-endian, which means Q actually comes
        before I, leading to the actual physical layout being QQII and so on.

    Type Parameters:
        - `TArray`: This function is multi-backend, and supports numpy
            `np.ndarray`, jax `jax.Array`, and torch `Tensor`.

    Args:
        iiqq: interleaved IIQQ data; see [`RadarFrame`][xwr.capture.types.].
            If already complex, leave it as is.
        sample_swap: if `True`, swap the I and Q components so that the
            output is `Q + j*I` instead of `I + j*Q`.

    Returns:
        Complex IQ data.
    """
    shape = (*iiqq.shape[:-1], iiqq.shape[-1] // 2)

    backend = _check_backend(iiqq)
    if backend == "numpy":
        assert isinstance(iiqq, np.ndarray)

        if iiqq.dtype == np.complex64:
            return iiqq
        iq = np.zeros(shape, dtype=np.complex64)
        if sample_swap:
            iq[..., 0::2] = iiqq[..., 0::4] + 1j * iiqq[..., 2::4]
            iq[..., 1::2] = iiqq[..., 1::4] + 1j * iiqq[..., 3::4]
        else:
            iq[..., 0::2] = 1j * iiqq[..., 0::4] + iiqq[..., 2::4]
            iq[..., 1::2] = 1j * iiqq[..., 1::4] + iiqq[..., 3::4]
        return cast(Complex64[TArray, "... n/2"], iq)

    elif backend == "jax":
        from jax import numpy as jnp
        assert isinstance(iiqq, jnp.ndarray)

        if iiqq.dtype == jnp.complex64:
            return iiqq
        if sample_swap:
            iq = jnp.zeros(
                shape, dtype=jnp.complex64
            ).at[..., 0::2].set(iiqq[..., 0::4] + 1j * iiqq[..., 2::4]
            ).at[..., 1::2].set(iiqq[..., 1::4] + 1j * iiqq[..., 3::4])
        else:
            iq = jnp.zeros(
                shape, dtype=jnp.complex64
            ).at[..., 0::2].set(1j * iiqq[..., 0::4] + iiqq[..., 2::4]
            ).at[..., 1::2].set(1j * iiqq[..., 1::4] + iiqq[..., 3::4])
        return cast(Complex64[TArray, "... n/2"], iq)

    else: # backend == "torch"
        import torch
        assert isinstance(iiqq, torch.Tensor)

        if iiqq.dtype == torch.complex64:
            return iiqq
        iq = torch.zeros(shape, dtype=torch.complex64, device=iiqq.device)
        if sample_swap:
            iq[..., 0::2] = iiqq[..., 0::4] + 1j * iiqq[..., 2::4]
            iq[..., 1::2] = iiqq[..., 1::4] + 1j * iiqq[..., 3::4]
        else:
            iq[..., 0::2] = 1j * iiqq[..., 0::4] + iiqq[..., 2::4]
            iq[..., 1::2] = 1j * iiqq[..., 1::4] + iiqq[..., 3::4]
        return cast(Complex64[TArray, "... n/2"], iq)

xwr.rsp.iqiq_from_iiqq 

iqiq_from_iiqq(
    iiqq: Int16[TArray, "... n"], sample_swap: bool = False
) -> Int16[TArray, "... n/2 2"]

Un-interleave IIQQ data.

Type Parameters
  • TArray: This function is multi-backend, and supports numpy np.ndarray, jax jax.Array, and torch Tensor.

Parameters:

Name Type Description Default
iiqq Int16[TArray, '... n']

interleaved IIQQ data; see RadarFrame.

 required
sample_swap bool

if True, swap the I and Q components in the output.

 False

Returns:

Type Description
Int16[TArray, '... n/2 2']

IQ data in an uninterleaved format with a trailing I/Q axis.
Source code in src/xwr/rsp/generic.py 
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
def iqiq_from_iiqq(
    iiqq: Int16[TArray, "... n"],
    sample_swap: bool = False,
) -> Int16[TArray, "... n/2 2"]:
    """Un-interleave IIQQ data.

    Type Parameters:
        - `TArray`: This function is multi-backend, and supports numpy
            `np.ndarray`, jax `jax.Array`, and torch `Tensor`.

    Args:
        iiqq: interleaved IIQQ data; see [`RadarFrame`][xwr.capture.types.].
        sample_swap: if `True`, swap the I and Q components in the output.

    Returns:
        IQ data in an uninterleaved format with a trailing I/Q axis.
    """
    shape = (*iiqq.shape[:-1], iiqq.shape[-1] // 2)
    i_idx, q_idx = (1, 0) if sample_swap else (0, 1)

    backend = _check_backend(iiqq)
    if backend == "numpy":
        assert isinstance(iiqq, np.ndarray)

        iq = np.zeros((*shape, 2), dtype=np.int16)
        iq[..., 0::2, q_idx] = iiqq[..., 0::4]
        iq[..., 1::2, q_idx] = iiqq[..., 1::4]
        iq[..., 0::2, i_idx] = iiqq[..., 2::4]
        iq[..., 1::2, i_idx] = iiqq[..., 3::4]
        return cast(Int16[TArray, "... n/2 2"], iq)

    elif backend == "jax":
        from jax import numpy as jnp
        assert isinstance(iiqq, jnp.ndarray)

        iq = jnp.zeros(
            (*shape, 2), dtype=jnp.int16
        ).at[..., 0::2, q_idx].set(iiqq[..., 0::4]
        ).at[..., 1::2, q_idx].set(iiqq[..., 1::4]
        ).at[..., 0::2, i_idx].set(iiqq[..., 2::4]
        ).at[..., 1::2, i_idx].set(iiqq[..., 3::4])
        return cast(Int16[TArray, "... n/2 2"], iq)

    else:  # backend == "torch"
        import torch
        assert isinstance(iiqq, torch.Tensor)

        iq = torch.zeros((*shape, 2), dtype=torch.int16, device=iiqq.device)
        iq[..., 0::2, q_idx] = iiqq[..., 0::4]
        iq[..., 1::2, q_idx] = iiqq[..., 1::4]
        iq[..., 0::2, i_idx] = iiqq[..., 2::4]
        iq[..., 1::2, i_idx] = iiqq[..., 3::4]
        return cast(Int16[TArray, "... n/2 2"], iq)