feat(galt_auto_cont): Add container and task for saving raw Galt autos.

ch_pipeline/analysis/beam.py

@@ -328,6 +328,8 @@ def process(self, data):
         # Convert input times to hour angle
         lha = unwrap_lha(self.observer.unix_to_lsa(data.time), ra)
 
+        self.log.warning(f"The starting and ending hour angles are {lha.min():.2f} and {lha.max():.2f}.")
+
         # perform regridding
         success = 1
         try:
@@ -1302,6 +1304,7 @@ def unwrap_lha(lsa, src_ra):
         Hour angle.
     """
     # ensure monotonic
+    lsa[lsa > 180] -= 360.0
     start_lsa = lsa[0]
     lsa -= start_lsa
     lsa[lsa < 0] += 360.0

ch_pipeline/analysis/calibration.py

@@ -16,6 +16,7 @@
 from ch_util import fluxcat
 from ch_util import finder
 from ch_util import rfi
+from ch_util import layout
 
 from draco.core import task
 from draco.util import _fast_tools
@@ -2364,3 +2365,65 @@ def _calculate_uv(freq, prod, inputmap):
     uv = dist[:, np.newaxis, :] / lmbda[np.newaxis, :, np.newaxis]
 
     return uv
+
+class ExtractGaltAutoCorrelation(task.SingleTask):
+    """Extract the autocorrelations of the Galt telescope from a holography acquisition."""
+
+    def process(self, data):
+        """Extract the Galt autocorrelations and write them to disk.
+        Parameters
+        ----------
+        data: TimeStream
+            A TimeStream container holding a raw holography acquisition.
+        Returns
+        -------
+        autocorrelation: containers.GaltAutocorrelation
+            A GaltAutocorrelation container holding the extracted Galt autos
+            as a function of frequency, polarization product, and time.
+        """
+        # Redistribute over freq
+        data.redistribute("freq")
+
+        # Locate the holographic indices
+        layout_graph = layout.graph.from_db(data.time[0])
+
+        galt_inputs = get_holographic_index(
+            get_correlator_inputs(layout_graph, correlator="chime")
+        )
+
+        # Get the product map and inputs
+        prodmap = data.prod
+        ina, inb = prodmap["input_a"], prodmap["input_b"]
+
+        # Locate the Galt autocorrelations and cross-pol correlation
+        flag_YY = np.where((ina == galt_inputs[0]) & (inb == galt_inputs[0]), 1, 0)
+        flag_YX = np.where((ina == galt_inputs[0]) & (inb == galt_inputs[1]), 1, 0)
+        flag_XX = np.where((ina == galt_inputs[1]) & (inb == galt_inputs[1]), 1, 0)
+
+        auto_flag = (flag_YY + flag_YX + flag_XX).astype(bool)
+
+        # Dereference beam and weight datasets
+        beam = data.vis[:].local_array
+        weight = data.weight[:].local_array
+
+        # Load only the data corresponding to the Galt inputs
+        galt_auto = beam[:, auto_flag, :]
+        galt_weight = weight[:, auto_flag, :]
+
+        # Initialize the auto container
+        autocorrelation = containers.GaltAutocorrelation(
+            pol=np.array([b"YY", b"YX", b"XX"]),
+            attrs_from=data,
+            freq=data.freq,
+            time=data.index_map["time"],
+            comm=data.comm,
+            distributed=data.distributed,
+        )
+
+        # Redistribute output container over frequency
+        autocorrelation.redistribute("freq")
+
+        autocorrelation.auto[:].local_array[:] = galt_auto
+        autocorrelation.weight[:].local_array[:] = galt_weight
+
+        return autocorrelation

ch_pipeline/core/containers.py

@@ -802,6 +802,39 @@ def source(self):
         return self.index_map["source"]
 
 
+class GaltAutocorrelation(FreqContainer, TODContainer):
+    _axes = ("freq", "pol", "time")
+
+    _dataset_spec = {
+        "auto": {
+            "axes": ["freq", "pol", "time"],
+            "dtype": np.complex64,
+            "initialise": True,
+            "distributed": True,
+            "distributed_axis": "freq",
+        },
+        "weight": {
+           "axes": ["freq", "pol", "time"],
+           "dtype": np.float64,
+           "initialise": True,
+           "distributed": True,
+           "distributed_axis": "freq",
+        },
+    }
+
+    @property
+    def auto(self):
+        return self.datasets["auto"]
+
+    @property
+    def weight(self):
+        return self.datasets["weight"]
+
+    @property
+    def fpga_count(self):
+        return self.index_map["time"]["fpga_count"]
+
+
 def make_empty_corrdata(
     freq=None,
     input=None,

ch_pipeline/core/io.py

@@ -40,7 +40,6 @@
 
 from draco.core import task, io
 
-
 class LoadCorrDataFiles(task.SingleTask):
     """Load data from files passed into the setup routine.
 
@@ -473,3 +472,4 @@ def next(self, files):
         )
 
         return sorted(list(new_files))
+