antenna_pairs.py

Go to the documentation of this file.

 
import numpy as np
import polars as pl
 
 
def generate_MI_antenna_pairs(antennas: pl.DataFrame,
                              phi_sector_group_size: int = 3) -> pl.DataFrame:
    r"""! Pairs up neighboring antennas in the main instrument.
    @ingroup AAA
    @param[in] antennas                Column `AntIdx` is **required**
    @param[in] phi_sector_group_size   (optional) Make sure \f$\geq 1\f$
    @retval    antenna_pairs           See the sample output below.
 
* Parameters:
    * `antennas`: The `AntIdx` column is required, all other columns are optional.
      See the following examples.
        -#  A dataframe with
            [placeholder integers](#calibration.generate_antenna_phase_center_pairs_with_placeholders())
            as antenna phase center positions, or
        -#  Directly from [qrh.dat](#antenna_attributes.read_MI_antenna_geometry)
 
    * `phi_sector_group_size`: Actually this is (roughly) _half_ the group size.
      Check out the [plots](@ref ant_pair_img) in the file description
      to see what effect this parameter has on the number of pairs generated.
      @warning For non-default values, make sure @p phi_sector_group_size \f$\geq 1\f$
 
* Example input `antennas`:
    ```
    ┌────────┬──────┬───────────┬────────┬───────┬────────┬───────┬────────────┬──────────┬───────────┐
    │ AntNum ┆ Ring ┆ PhiSector ┆ AntIdx ┆ X[m]  ┆ Y[m]   ┆ Z[m]  ┆ Pitch[deg] ┆ Yaw[deg] ┆ Roll[deg] │
    │ ---    ┆ ---  ┆ ---       ┆ ---    ┆ ---   ┆ ---    ┆ ---   ┆ ---        ┆ ---      ┆ ---       │
    │ enum   ┆ u8   ┆ u8        ┆ u32    ┆ f64   ┆ f64    ┆ f64   ┆ f64        ┆ f64      ┆ f64       │
    ╞════════╪══════╪═══════════╪════════╪═══════╪════════╪═══════╪════════════╪══════════╪═══════════╡
    │ 101    ┆ 1    ┆ 1         ┆ 0      ┆ 1.488 ┆ 0.0    ┆ 5.114 ┆ -10.0      ┆ 0.0      ┆ 0.0       │
    │ 201    ┆ 2    ┆ 1         ┆ 1      ┆ 2.596 ┆ 0.0    ┆ 1.457 ┆ -10.0      ┆ 0.0      ┆ 0.0       │
    │ …      ┆ …    ┆ …         ┆ …      ┆ …     ┆ …      ┆ …     ┆ …          ┆ …        ┆ …         │
    │ 324    ┆ 3    ┆ 24        ┆ 94     ┆ 2.507 ┆ -0.672 ┆ 0.728 ┆ -10.0      ┆ -15.0    ┆ 0.0       │
    │ 424    ┆ 4    ┆ 24        ┆ 95     ┆ 2.507 ┆ -0.672 ┆ 0.0   ┆ -10.0      ┆ -15.0    ┆ 0.0       │
    └────────┴──────┴───────────┴────────┴───────┴────────┴───────┴────────────┴──────────┴───────────┘
    ```
* With default @p phi_sector_group_size and the example input above, the output schema looks like
    ```
    Rows: 912
    Columns: 20
    $ A1_AntNum     <enum>
    $ A1_Ring         <u8>
    $ A1_PhiSector    <u8>
    $ A1_AntIdx      <u32>
    $ A1_X[m]        <f64>
    $ A1_Y[m]        <f64>
    $ A1_Z[m]        <f64>
    $ A1_Pitch[deg]  <f64>
    $ A1_Yaw[deg]    <f64>
    $ A1_Roll[deg]   <f64>
    $ A2_AntNum     <enum>
    $ A2_Ring         <u8>
    $ A2_PhiSector    <u8>
    $ A2_AntIdx      <u32>
    $ A2_X[m]        <f64>
    $ A2_Y[m]        <f64>
    $ A2_Z[m]        <f64>
    $ A2_Pitch[deg]  <f64>
    $ A2_Yaw[deg]    <f64>
    $ A2_Roll[deg]   <f64>
    ```
    """
 
    from antenna_attributes import NUM_PHI_SECTORS
    from scipy.sparse import diags_array
 
    # will abort early if column not present
    antennas["AntIdx"]
 
    building_block = np.ones((4, 4), dtype=int)  # ie. four antennas in one phi sector
    if phi_sector_group_size < 1:
        raise ValueError("Group size has to be greater than 0!")
 
    dia_idx = np.unique(np.arange(-phi_sector_group_size + 1, phi_sector_group_size) % NUM_PHI_SECTORS)
 
    phi_neighbors = diags_array(np.ones(len(dia_idx)), offsets=dia_idx, dtype=int,
                                shape=(NUM_PHI_SECTORS, NUM_PHI_SECTORS)).toarray()
    # __make_plot_for_sanity_check(phi_neighbors, "Phi Sector Groups")
 
    antenna_pair_matrix = np.triu(np.kron(phi_neighbors, building_block), 1)
    # __make_plot_for_sanity_check(antenna_pair_matrix, "Antenna Pairs")
 
    antenna_pairs = (
        pl.DataFrame(np.where(antenna_pair_matrix > 0), schema=['1', '2'])
        .join(antennas, right_on="AntIdx", left_on='1', coalesce=False)
        .select(pl.all().exclude('1', '2').name.prefix("A1_"), '2')
        .join(antennas, right_on="AntIdx", left_on='2', coalesce=False)
        .select(
            pl.col(r"^A1_.*$"),
            pl.all().exclude(r"^A1_.*$", '2').name.prefix("A2_")
        )
    )
 
    return antenna_pairs
 
 
def __make_plot_for_sanity_check(square_matrix, titlename):
 
    import matplotlib.pyplot as plt
    plt.rcParams["figure.figsize"] = (15, 15)
    plt.rcParams["font.size"] = 9
 
    assert (np.shape(square_matrix)[0] == np.shape(square_matrix)[1]), "Not a square matrix."
 
    square_matrix_side_length = np.shape(square_matrix)[0]
    minor = np.arange(.5, square_matrix_side_length)
    plt.xticks(np.arange(square_matrix_side_length),
               np.arange(square_matrix_side_length) + 1, rotation=90)
    plt.xticks(minor, minor=True)
    plt.yticks(np.arange(square_matrix_side_length), np.arange(square_matrix_side_length) + 1)
    plt.yticks(minor, minor=True)
    plt.grid(which='minor', color='k', alpha=.3,)
    plt.tick_params(axis='both', which='both', top=True, right=True, bottom=True, left=True,
                    labeltop=True, labelright=True)
    plt.imshow(square_matrix, cmap='Greys', alpha=.6)
    plt.title(titlename)
    plt.savefig(f"{titlename}.svg")
 
 
# main function, serving as an example
if __name__ == "__main__":
    import os
    from pathlib import Path
    from antenna_attributes import read_MI_antenna_geometry
 
    uncalibrated_antenna_positions: Path = (
        os.environ.get("PUEO_UTIL_INSTALL_DIR") / Path("share/pueo/geometry/jun25/qrh.dat")
    )
    antennas: pl.DataFrame = read_MI_antenna_geometry(uncalibrated_antenna_positions)
 
    try:
        bad = antennas.select(pl.all().exclude("AntIdx"))
        generate_MI_antenna_pairs(antennas=bad)
 
    except pl.exceptions.ColumnNotFoundError:
        print("Column AntIdx is required, so this should fail!")
 
    print("On the other hand, keeping only AntIdx is okay:")
    good = antennas.select("AntIdx")
    print(generate_MI_antenna_pairs(antennas=good))