# -*- mode: python; coding: utf-8 -*- # Copyright (c) 2020 Radio Astronomy Software Group # Licensed under the 2-clause BSD License """Class for reading and writing Mir files.""" import numpy as np from .uvdata import UVData from . import mir_parser from .. import utils as uvutils from .. import get_telescope __all__ = ["Mir"] class Mir(UVData): """ A class for Mir file objects. This class defines an Mir-specific subclass of UVData for reading and writing Mir files. This class should not be interacted with directly, instead use the read_mir and write_mir methods on the UVData class. """ def read_mir(self, filepath, isource=None, irec=None, isb=None, corrchunk=None): """ Read in data from an SMA MIR file, and map to the UVData model. Note that with the exception of filename, the rest of the parameters are used to sub-select a range of data that matches the limitations of the current instantiation of pyuvdata -- namely 1 spectral window, 1 source. These could be dropped in the future, as pyuvdata capabilities grow. Parameters ---------- filepath : str The file path to the MIR folder to read from. isource : int Source code for MIR dataset irec : int Receiver code for MIR dataset isb : int Sideband code for MIR dataset corrchunk : int Correlator chunk code for MIR dataset """ # Use the mir_parser to read in metadata, which can be used to select data. mir_data = mir_parser.MirParser(filepath) # Select out data that we want to work with. if isource is None: isource = mir_data.in_read["isource"][0] if irec is None: irec = mir_data.bl_read["irec"][0] if isb is None: isb = mir_data.bl_read["isb"][0] if corrchunk is None: corrchunk = mir_data.sp_read["corrchunk"][0] mir_data.use_in = mir_data.in_read["isource"] == isource mir_data.use_bl = np.logical_and( np.logical_and( mir_data.bl_read["isb"] == isb, mir_data.bl_read["ipol"] == 0 ), mir_data.bl_read["irec"] == irec, ) mir_data.use_sp = mir_data.sp_read["corrchunk"] == corrchunk # Load up the visibilities into the MirParser object. This will also update the # filters, and will make sure we're looking at the right metadata. mir_data._update_filter() if len(mir_data.in_data) == 0: raise IndexError("No valid records matching those selections!") mir_data.load_data(load_vis=True, load_raw=True) # Create a simple array/list for broadcasting values stored on a # per-intergration basis in MIR into the (tasty) per-blt records in UVDATA. bl_in_maparr = [mir_data.inhid_dict[idx] for idx in mir_data.bl_data["inhid"]] # Derive Nants_data from baselines. self.Nants_data = len( np.unique( np.concatenate((mir_data.bl_data["iant1"], mir_data.bl_data["iant2"])) ) ) self.Nants_telescope = 8 self.Nbls = int(self.Nants_data * (self.Nants_data - 1) / 2) self.Nblts = len(mir_data.bl_data) self.Nfreqs = int(mir_data.sp_data["nch"][0]) self.Npols = 1 # todo: We will need to go back and expand this. self.Nspws = 1 # todo: We will need to go back and expand this. self.Ntimes = len(mir_data.in_data) self.ant_1_array = mir_data.bl_data["iant1"] - 1 self.ant_2_array = mir_data.bl_data["iant2"] - 1 self.antenna_names = [ "Ant 1", "Ant 2", "Ant 3", "Ant 4", "Ant 5", "Ant 6", "Ant 7", "Ant 8", ] self.antenna_numbers = np.arange(8) # Prepare the XYZ coordinates of the antenna positions. antXYZ = np.zeros([self.Nants_telescope, 3]) for idx in range(self.Nants_telescope): if (idx + 1) in mir_data.antpos_data["antenna"]: antXYZ[idx] = mir_data.antpos_data["xyz_pos"][ mir_data.antpos_data["antenna"] == (idx + 1) ] # Get the coordinates from the entry in telescope.py lat, lon, alt = get_telescope("SMA")._telescope_location.lat_lon_alt() self.telescope_location_lat_lon_alt = (lat, lon, alt) # Calculate antenna postions in EFEF frame. Note that since both # coordinate systems are in relative units, no subtraction from # telescope geocentric position is required , i.e we are going from # relRotECEF -> relECEF self.antenna_positions = uvutils.ECEF_from_rotECEF(antXYZ, lon) self.baseline_array = self.antnums_to_baseline( self.ant_1_array, self.ant_2_array, attempt256=False ) fsky = mir_data.sp_data["fsky"][0] * 1e9 # GHz -> Hz fres = mir_data.sp_data["fres"][0] * 1e6 # MHz -> Hz nch = mir_data.sp_data["nch"][0] self.channel_width = fres # Need the half-channel offset below because of the weird way # in which MIR identifies the "center" of the band self.freq_array = fsky + fres * (np.arange(nch) - (nch / 2 - 0.5)) # TODO: This will need to be fixed when spw > 1 self.freq_array = np.reshape(self.freq_array, (1, -1)) self.history = "Raw Data" self.instrument = "SWARM" # todo: This won't work when we have multiple spectral windows. self.integration_time = mir_data.sp_data["integ"] # todo: Using MIR V3 convention, will need to be V2 compatible eventually. self.lst_array = ( mir_data.in_data["lst"][bl_in_maparr].astype(float) + (0.0 / 3600.0) ) * (np.pi / 12.0) # todo: We change between xx yy and rr ll, so we will need to update this. self.polarization_array = np.asarray([-5]) self.spw_array = np.asarray([0]) self.telescope_name = "SMA" time_array_mjd = mir_data.in_read["mjd"][bl_in_maparr] self.time_array = time_array_mjd + 2400000.5 # Need to flip the sign convention here on uvw, since we use a1-a2 versus the # standard a2-a1 that uvdata expects self.uvw_array = (-1.0) * np.transpose( np.vstack( (mir_data.bl_data["u"], mir_data.bl_data["v"], mir_data.bl_data["w"]) ) ) # todo: Raw data is in correlation coefficients, we may want to convert to Jy. self.vis_units = "uncalib" self._set_phased() sou_list = mir_data.codes_data[mir_data.codes_data["v_name"] == b"source"] self.object_name = sou_list[sou_list["icode"] == isource]["code"][0].decode( "utf-8" ) self.phase_center_ra = mir_data.in_data["rar"][0] self.phase_center_dec = mir_data.in_data["decr"][0] self.phase_center_epoch = mir_data.in_data["epoch"][0] self.phase_center_epoch = float(self.phase_center_epoch) self.antenna_diameters = np.zeros(self.Nants_telescope) + 6 self.blt_order = ("time", "baseline") self.data_array = np.reshape( np.array(mir_data.vis_data), (self.Nblts, self.Nspws, self.Nfreqs, self.Npols), ) # Don't need the data anymore, so drop it mir_data.unload_data() self.flag_array = np.zeros(self.data_array.shape, dtype=bool) self.nsample_array = np.ones(self.data_array.shape, dtype=np.float32) def write_mir(self, filename): """ Write out the SMA MIR files. Parameters ---------- filename : str The path to the folder on disk to write data to. """ raise NotImplementedError