Skip to content

Pulse Extension in RX DBF of EAP #324

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Open
rad-eng-59 wants to merge 4 commits into
base: develop
Choose a base branch
from
Open

Pulse Extension in RX DBF of EAP #324

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
4 commits
Select commit Hold shift + click to select a range
9d24eb6
Introduce pulse extension for rx dbf in beamformer class
rad-eng-59 Jun 19, 2026
1b47879
Add pulse extension interface for rx dbf to pattern class
rad-eng-59 Jun 19, 2026
d856cf2
Add option to remove tx phase toggling in tx bmf of pattern class
rad-eng-59 Jun 19, 2026
f0f95a4
Replace median with mean for discarding toggling in tx bmf
rad-eng-59 Jun 21, 2026
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
95 changes: 90 additions & 5 deletions python/packages/nisar/antenna/beamformer.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -2,9 +2,11 @@
from abc import ABC, abstractmethod
import warnings
import numpy as np
import logging

from scipy.interpolate import interp1d
from scipy.signal import convolve

import isce3
from isce3.antenna import ant2rgdop, ant2geo
from isce3.core import Linspace, speed_of_light
from nisar.antenna import CalPath
Expand All @@ -16,6 +18,9 @@
DEFAULT_NUM_PULSE_SKIP = 12


logger = logging.getLogger('beamformer')


class BadHPACalWarning(Warning):
"""Warning for Bad HPA CAL values"""
pass
Expand Down Expand Up @@ -482,6 +487,9 @@ class RxDBF(ElevationBeamformer):
If both this parameter and the `rg_spacing_min` are set to None, all
input slant range values will be included in the interpolation process
of the antenna beam formation.
pulse_ext: float, optional
If not None, it will be convolved with RX patterns prior to forming
RX DBF pattern.

Attributes
----------
Expand All @@ -506,9 +514,11 @@ def __init__(self, orbit, attitude, dem_interp, el_ant_info, trm_info,
ref_epoch, *, el_lut=None, norm_weight=False,
num_pulse_skip=DEFAULT_NUM_PULSE_SKIP,
el_ofs_dbf=0.0, rg_spacing_min=None,
el_spacing_min=8.72665e-5):
el_spacing_min=8.72665e-5,
pulse_ext=None):
self.el_ofs_dbf = el_ofs_dbf
self.el_spacing_min = el_spacing_min
self.pulse_ext = pulse_ext
super().__init__(orbit, attitude, dem_interp, el_ant_info, trm_info,
ref_epoch, el_lut=el_lut, norm_weight=norm_weight,
num_pulse_skip=num_pulse_skip,
Expand All @@ -532,6 +542,36 @@ def __init__(self, orbit, attitude, dem_interp, el_ant_info, trm_info,
1.0
)
self.rg_spacing_min = abs(np.diff(sr)[0])
# get slant range spacing corresponding to half DBF angular resolution
# at nearest range from the very first channel
if pulse_ext is not None:
ela_first = trm_info.el_ang_dbf[self.active_channel_idx[0], 0]
ela_next = trm_info.el_ang_dbf[self.active_channel_idx[0], 1]
azt_mid = np.mean(self.trm_info.time)
pos_ecef, vel_ecef = self.orbit.interpolate(azt_mid)
q_ant2ecef = self.attitude.interpolate(azt_mid)
sr, _, _ = ant2rgdop(
[ela_first, ela_next],
self.el_ant_info.cut_angle,
pos_ecef,
vel_ecef,
q_ant2ecef,
1.0
)
rg_space_rect = 0.5 * abs(np.diff(sr)[0])
dsr_pw_ext = 0.5 * speed_of_light * pulse_ext
self._size_rect = max(1, round(dsr_pw_ext / rg_space_rect))
self._rg_space_rect = dsr_pw_ext / self._size_rect
if self._size_rect < 2:
warnings.warn(
f'The size of pulse ext {pulse_ext * 1e6} (us) is '
f'{self._size_rect} less than 2! No pulsewidth in RX DBF!'
)
logger.info(f'The size of pulse ext {pulse_ext * 1e6} (us) is'
f' {self._size_rect}! Apply pulsewidth to RX DBF!')
else:
self._size_rect = 0
self._rg_space_rect = None

@property
def slant_range_dbf(self):
Expand Down Expand Up @@ -640,6 +680,21 @@ def form_pattern(self, pulse_time, slant_range, channel_adj_factors=None):
rx_pat = np.zeros((len(pulse_time), slant_range.size), dtype='complex')
num_active_chanl = len(self.active_channel_idx)

# form uniform coarse slant range vector used in pulse ext if any
if self._size_rect > 1:
# preserve exactly the same coarse spacing in slant range!
rg_ext = self._size_rect * self._rg_space_rect
sr_first = sr[0] - rg_ext
sr_last = sr[-1] + rg_ext
size_sr_pw_ext = round(
(sr_last - sr_first) / self._rg_space_rect) + 1
sr_pw_ext = sr_first + self._rg_space_rect * np.arange(
size_sr_pw_ext)
ant_pat_el_pw = np.zeros(
(num_active_chanl, size_sr_pw_ext), dtype=ant_pat_el.dtype)
rect_ext = 1. / self._size_rect * np.ones((1, self._size_rect))
slice_ext = np.s_[0:size_sr_pw_ext]

# loop over pulses
for pp, tm in enumerate(pulse_time):
# Compute the respective slant range for beamformed antenna pattern
Expand All @@ -653,14 +708,44 @@ def form_pattern(self, pulse_time, slant_range, channel_adj_factors=None):
x = 0
if self.el_lut is None:
sr_ant = self._elaz2slantrange(tm)
# Apply pulse ext if any
if self._size_rect > 1:
for cc in range(num_active_chanl):
ant_pat_el_pw[cc] = np.interp(
sr_pw_ext, sr_ant, ant_pat_el[cc])
# do pulse convolution
ant_pat_el_pw[...] = convolve(
ant_pat_el_pw, rect_ext, mode='full'
)[:, slice_ext]
else: # no pulse ext
ant_pat_el_pw = ant_pat_el
sr_pw_ext = sr_ant
# form RX DBF pattern
for cc in range(num_active_chanl):
x += np.interp(
sr, sr_ant, ant_pat_el[cc, :]) * rx_wgt[cc, :]
sr, sr_pw_ext, ant_pat_el_pw[cc, :]
) * rx_wgt[cc, :]
else:
sr_angles = self.el_lut.eval(tm, sr)
# Apply pulse ext if any
if self._size_rect > 1:
el_pw_ext = self.el_lut.eval(tm, sr_pw_ext)
for cc in range(num_active_chanl):
ant_pat_el_pw[cc] = np.interp(
el_pw_ext,
self.el_ant_info.angle,
ant_pat_el[cc])
# do pulse convolution
ant_pat_el_pw[...] = convolve(
ant_pat_el_pw, rect_ext, mode='full'
)[:, slice_ext]
else: # no pulse ext
ant_pat_el_pw = ant_pat_el
el_pw_ext = self.el_ant_info.angle
# form RX DBF pattern
for cc in range(num_active_chanl):
x += np.interp(sr_angles, self.el_ant_info.angle,
ant_pat_el[cc, :]) * rx_wgt[cc, :]
x += np.interp(sr_angles, el_pw_ext,
ant_pat_el_pw[cc, :]) * rx_wgt[cc, :]

rx_pat[pp] = x.repeat(nrgb_skip)[:slant_range.size]

Expand Down
74 changes: 69 additions & 5 deletions python/packages/nisar/antenna/pattern.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -114,19 +114,70 @@ def get_dbf_timing(self, t):


def build_tx_trm(raw: Raw, pulse_times: np.ndarray, freq_band: str,
tx_pol: str):
"""Build TxTrmInfo object """
tx_pol: str, remove_toggling: bool = False):
"""Build TxTrmInfo object

Parameters
----------
raw : nisar.products.readers.raw.Raw
Raw L0B product parser.
pulse_times : np.ndarray(float)
Pulse times in seconds
freq_band : str
frequency band, 'A' or 'B'.
tx_pol : str
Transmit polarization such 'H', 'V', etc
remove_toggling : bool, default=False
Discard pulse-to-pulse TX phase toggling if True.

Returns
-------
TxTrmInfo
data class for all key calibration info on TX modules.

"""
# Parse Tx-related Cal stuff used for Tx BMF
tx_chanl = raw.getListOfTxTRMs(freq_band, tx_pol)
# get chirp correlator and cal type for co-pol product
chp_corr, cal_type = chirpcorrelator_caltype_from_raw(
raw, txrx_pol=2 * tx_pol)
corr_tap2 = chp_corr[..., 1]
if remove_toggling:
warn('Remove TX toggling, if any, for the second tap HPA!')
from nisar.antenna import get_calib_range_line_idx
i_hpa, _, _, _ = get_calib_range_line_idx(cal_type)
hpa_mean = np.nanmean(corr_tap2[i_hpa], axis=0)
corr_tap2[i_hpa] = hpa_mean

# build TxTRM from Tx Cal stuff w/o optional "tx_phase"
return TxTrmInfo(pulse_times, tx_chanl, corr_tap2,
cal_type)


def pulse_ext_from_raw(raw: Raw) -> float:
"""
Get pulse extension which is total duration of
sequentially transmitted chirps.

Parameters
----------
raw : nisar.products.readers.raw.Raw
Raw L0B product parser.

Returns
-------
float
Total pulse width in seconds transmitted sequentially.

"""
pw_ext = 0
for freq_band in raw.frequencies:
txrx_pol = sorted(raw.polarizations[freq_band])[0]
_, _, _, pw = raw.getChirpParameters(freq_band, txrx_pol[0])
pw_ext += pw
return pw_ext


class AntennaPattern:
"""
Convenience class for generating combined TX*RX elevation pattern for
Expand Down Expand Up @@ -165,6 +216,11 @@ class AntennaPattern:
delay_ofs_dbf: float, default=-2.1474e-6
Delay offset (seconds) in data window position of onboard DBF
process applied to all bands and polarizations.
remove_toggling_tx : bool, default=False
Discard pulse-to-pulse TX phase toggling if True while
foming TX BMF pattern. This is useful for synthesizing
antenna pattern during debugging and performance analysis.
For ops, this feature is off.

"""

Expand All @@ -176,14 +232,17 @@ def __init__(self, raw: Raw, dem: DEMInterpolator,
el_spacing_min=8.72665e-5,
freq_band=None,
caltone_freq=None,
delay_ofs_dbf=-2.1474e-6):
delay_ofs_dbf=-2.1474e-6,
remove_toggling_tx=False):

self.orbit = orbit.copy()
self.attitude = attitude.copy()
self.dem = dem
self.norm_weight = norm_weight
self.el_spacing_min = el_spacing_min
self.el_lut = el_lut
self.remove_toggling_tx = remove_toggling_tx
self.pw_ext = pulse_ext_from_raw(raw)

# get frequency band
freqs = np.sort(raw.frequencies)
Expand Down Expand Up @@ -303,6 +362,7 @@ def __init__(self, raw: Raw, dem: DEMInterpolator,
el_lut=self.el_lut,
norm_weight=self.norm_weight,
el_spacing_min=self.el_spacing_min,
pulse_ext=self.pw_ext
)
self.rg_spacing_min = self.rx_dbf[rx_p].rg_spacing_min
else:
Expand All @@ -312,6 +372,7 @@ def __init__(self, raw: Raw, dem: DEMInterpolator,
el_lut=self.el_lut,
norm_weight=self.norm_weight,
rg_spacing_min=self.rg_spacing_min,
pulse_ext=self.pw_ext
)

# build all TxBMFs for all possible TX linear polarizations
Expand All @@ -327,7 +388,9 @@ def __init__(self, raw: Raw, dem: DEMInterpolator,
# pairings of freq band and TX pol.
for tx_band, pols in raw.polarizations.items():
if tx_p in {pol[0] for pol in pols}:
tx_trm = build_tx_trm(raw, self.pulse_times, tx_band, tx_p)
tx_trm = build_tx_trm(
raw, self.pulse_times, tx_band, tx_p,
remove_toggling=self.remove_toggling_tx)
break
else:
assert False, f"couldn't find freq_id for tx_pol={tx_p}"
Expand Down Expand Up @@ -428,7 +491,8 @@ def form_pattern(self, tseq, slant_range: Linspace,
self.orbit, self.attitude, self.dem, self.el_pat_rx[rxp],
self.rx_trm[rxp], self.reference_epoch,
el_lut=self.el_lut,
norm_weight=self.rx_dbf[rxp].norm_weight)
norm_weight=self.rx_dbf[rxp].norm_weight,
pulse_ext=self.pw_ext)

pat = self.rx_dbf[rxp].form_pattern(
tgroup, slant_range,
Expand Down
Loading