orca.transform.hot_baselines
============================

.. py:module:: orca.transform.hot_baselines

.. autoapi-nested-parse::

   Hot-baseline and amplitude-vs-UV diagnostics.

   Ported from the standalone ``hot_baseline_worker.py`` script into the orca
   package so that the Celery pipeline does not depend on ``preliminary_pipeline/``.

   The main entry point is :func:`run_diagnostics`, which is called by the
   Phase-2 Celery task when ``hot_baselines=True``.



Attributes
----------

.. autoapisummary::

   orca.transform.hot_baselines.logger


Functions
---------

.. autoapisummary::

   orca.transform.hot_baselines.extract_ant_number
   orca.transform.hot_baselines.get_ms_id_map
   orca.transform.hot_baselines.build_complete_map
   orca.transform.hot_baselines.get_antenna_positions
   orca.transform.hot_baselines.get_mean_frequency
   orca.transform.hot_baselines.get_data_matrix
   orca.transform.hot_baselines.apply_uv_cut
   orca.transform.hot_baselines.analyze_amp_vs_uv
   orca.transform.hot_baselines.identify_bad_components
   orca.transform.hot_baselines.apply_flags_to_ms
   orca.transform.hot_baselines.plot_heatmap
   orca.transform.hot_baselines.plot_diagnosis_figure
   orca.transform.hot_baselines.run_diagnostics


Module Contents
---------------

.. py:data:: logger

.. py:function:: extract_ant_number(name)

   Extract the integer antenna number from a name like 'LWA-042'.


.. py:function:: get_ms_id_map(ms_name: str) -> Dict[int, int]

   Map LWA antenna numbers to MS row indices from the ANTENNA sub-table.


.. py:function:: build_complete_map(ms_id_map: Dict[int, int]) -> pandas.DataFrame

   Build a DataFrame mapping correlator numbers to hardware info.

   Uses ``corr_num`` directly from :data:`SYSTEM_CONFIG` (verified against
   the wiring spreadsheet) rather than inferring it from MS ANTENNA table
   row indices.  *ms_id_map* is still used to verify which antennas are
   present in the MS.


.. py:function:: get_antenna_positions(ms_name: str) -> numpy.ndarray

   Read ECEF antenna positions from the MS.


.. py:function:: get_mean_frequency(ms_name: str) -> float

   Read the mean frequency from the SPECTRAL_WINDOW sub-table.


.. py:function:: get_data_matrix(ms_name: str, data_col: str = 'CORRECTED_DATA') -> Tuple[Dict[str, numpy.ndarray], Dict[str, numpy.ndarray], int]

   Compute per-baseline mean and std amplitude matrices from an MS.

   Reads in chunks for memory efficiency.

   :returns: (mean_matrices, std_matrices, n_ant) — dicts keyed by pol name.


.. py:function:: apply_uv_cut(matrix: numpy.ndarray, pos_array: numpy.ndarray, uv_cut_m: float) -> numpy.ndarray

   Blank baselines shorter than *uv_cut_m* metres.


.. py:function:: analyze_amp_vs_uv(raw_mean: Dict[str, numpy.ndarray], pos_array: numpy.ndarray, n_ant: int, mean_freq: Optional[float] = None, sigma_cut: float = 5.0, plot_prefix: str = 'amp_vs_uv') -> List[Tuple[int, int]]

   Identify outlier baselines from amplitude-vs-UV-distance scatter.

   Uses a rolling-median envelope to detect baselines whose amplitude
   exceeds ``sigma_cut`` MADs above the local median.

   :returns: List of (ant1, ant2) tuples for bad baselines.


.. py:function:: identify_bad_components(combined_matrix: numpy.ndarray, df_map: pandas.DataFrame, n_ant: int, extra_bad_baselines: Optional[List[Tuple[int, int]]] = None, sigma_cut: float = 5.0, threshold_percent: float = 0.1) -> Tuple[List[Tuple[int, int]], Set[int], List[str]]

   Find hot baselines and bad antennas from the cross-pol heatmap.

   :returns: (heatmap_bad_baselines, bad_antennas, report_lines).


.. py:function:: apply_flags_to_ms(ms_name: str, bad_antennas: Set[int], bad_baselines: List[Tuple[int, int]]) -> None

   Flag bad antennas and baselines in the MS FLAG column.


.. py:function:: plot_heatmap(matrix: numpy.ndarray, title: str, filename: str, vmin: Optional[float] = None, vmax: Optional[float] = None, grid_lines: Optional[List[float]] = None, axis_ticks: Optional[numpy.ndarray] = None, axis_labels: Optional[numpy.ndarray] = None) -> None

   Save a single heatmap as a PNG.


.. py:function:: plot_diagnosis_figure(matrix: numpy.ndarray, bad_antennas: Set[int], bad_baselines: List[Tuple[int, int]], filename: str, uv_baselines: Optional[List[Tuple[int, int]]] = None) -> None

   Save a two-panel diagnosis heatmap with overlays.


.. py:function:: run_diagnostics(args_obj, run_logger=None) -> None

   Run the full hot-baseline analysis on a concatenated MS.

   :param args_obj: Object with attributes ``ms``, ``col``, ``uv_cut``,
                    ``uv_cut_lambda``, ``sigma``, ``uv_sigma``, ``threshold``,
                    ``apply_antenna_flags``, ``apply_baseline_flags``,
                    ``run_uv``, ``run_heatmap``.
   :param run_logger: Optional logger; defaults to module logger.