Secure your code as it's written. Use Snyk Code to scan source code in minutes - no build needed - and fix issues immediately.
input_uv.Nblts = args.Nbls * args.Ntimes
# Beam selection:
# Default is uniform
if args.beam == 'hera':
beam = UVBeam()
beamfile = '/users/alanman/data/alanman/NickFagnoniBeams/HERA_NicCST_fullfreq.uvbeam'
beam_list = [beamfile]
mock_keywords = {'mock_arrangement': 'random', 'Nsrcs': args.Nsrcs,
'min_alt': min_alt, 'time': input_uv.time_array[0]}
print("Beam: {}".format(beam_list[0]))
params['sources'].update(**mock_keywords)
# Catalog setup
catalog, _ = simsetup.initialize_catalog_from_params(params)
comm = mpi.world_comm
input_uv = comm.bcast(input_uv, root=0)
beam_list = comm.bcast(beam_list, root=0)
beam_dict = comm.bcast(beam_dict, root=0)
catalog = mpi.shared_mem_bcast(catalog, root=0)
if rank == 0:
print("Starting simulation.")
sys.stdout.flush()
uv_out = uvsim.run_uvdata_uvsim(input_uv, beam_list, beam_dict=beam_dict, catalog=catalog)
memory_usage_GB = resource.getrusage(resource.RUSAGE_SELF).ru_maxrss / 1e6
comm.Barrier()
memory_usage_GB = comm.gather(memory_usage_GB, root=0)
if not isinstance(input_uv, UVData):
raise TypeError("input_uv must be UVData object")
if not ((input_uv.Npols == 4) and (input_uv.polarization_array.tolist() == [-5, -6, -7, -8])):
raise ValueError("input_uv must have XX,YY,XY,YX polarization")
# The root node will initialize our simulation
# Read input file and make uvtask list
if rank == 0 and not quiet:
print('Nbls:', input_uv.Nbls, flush=True)
print('Ntimes:', input_uv.Ntimes, flush=True)
print('Nfreqs:', input_uv.Nfreqs, flush=True)
print('Nsrcs:', catalog.Ncomponents, flush=True)
if rank == 0:
uv_container = simsetup._complete_uvdata(input_uv, inplace=False)
if 'world' in input_uv.extra_keywords:
uv_container.extra_keywords['world'] = input_uv.extra_keywords['world']
Nbls = input_uv.Nbls
Ntimes = input_uv.Ntimes
Nfreqs = input_uv.Nfreqs
Nsrcs = catalog.Ncomponents
task_inds, src_inds, Ntasks_local, Nsrcs_local = _make_task_inds(
Nbls, Ntimes, Nfreqs, Nsrcs, rank, Npus
)
# Construct beam objects from strings
beam_list.set_obj_mode(use_shared_mem=True)
# Estimating required memory to decide how to split source array.
# Will assume the same beam_id for all antennas for now.
parser = argparse.ArgumentParser(description=("Extracts antenna position info from uvfits."))
parser.add_argument('file_in', metavar='', type=str, nargs='+')
parser.add_argument('-l', '--layout_csv_name', default=None)
parser.add_argument('-t', '--telescope_config_name', default=None)
parser.add_argument('-b', '--beam_filepath', type=str,
default=os.path.join(SIM_DATA_PATH, 'HERA_NicCST.uvbeam'))
args = parser.parse_args()
uvd = UVData()
uvd.read(args.file_in[0])
pyuvsim.simsetup.uvdata_to_telescope_config(
uvd, args.beam_filepath, layout_csv_name=args.layout_csv_name,
telescope_config_name=args.telescope_config_name,
return_names=False, path_out='.'
)
# Take a uvfits file, and save sim parameters as a yaml file.
parser = argparse.ArgumentParser(description=("Generate basic simulation parameters from uvfits."))
parser.add_argument('file_in', metavar='', type=str, nargs='+')
parser.add_argument('-p', '--param_filename', default=None)
parser.add_argument('-t', '--telescope_config_path', default='')
parser.add_argument('-l', '--layout_csv_path', default='')
args = parser.parse_args()
uvd = UVData()
uvd.read(args.file_in[0])
pyuvsim.simsetup.uvdata_to_config_file(uvd, param_filename=args.param_filename,
telescope_config_name=args.telescope_config_path,
layout_csv_name=args.layout_csv_path,
catalog='mock', path_out='.')