diff --git a/python/packages/nisar/antenna/beamformer.py b/python/packages/nisar/antenna/beamformer.py index 4acc7cf9b..0ae36977d 100644 --- a/python/packages/nisar/antenna/beamformer.py +++ b/python/packages/nisar/antenna/beamformer.py @@ -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 @@ -16,6 +18,9 @@ DEFAULT_NUM_PULSE_SKIP = 12 +logger = logging.getLogger('beamformer') + + class BadHPACalWarning(Warning): """Warning for Bad HPA CAL values""" pass @@ -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 ---------- @@ -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, @@ -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): @@ -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 @@ -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] diff --git a/python/packages/nisar/antenna/pattern.py b/python/packages/nisar/antenna/pattern.py index e338da570..5f6caa3b6 100644 --- a/python/packages/nisar/antenna/pattern.py +++ b/python/packages/nisar/antenna/pattern.py @@ -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 @@ -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. """ @@ -176,7 +232,8 @@ 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() @@ -184,6 +241,8 @@ def __init__(self, raw: Raw, dem: DEMInterpolator, 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) @@ -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: @@ -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 @@ -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}" @@ -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,