Secure your code as it's written. Use Snyk Code to scan source code in minutes - no build needed - and fix issues immediately.
if self.flux_unit is not None:
stokes_use *= units.Unit(self.flux_unit)
other = {}
if self.spectral_type in ['full', 'subband']:
other['freq_array'] = self.freq_array * units.Hz
if self.spectral_type == 'spectral_index':
other['reference_frequency'] = self.reference_frequency * units.Hz
other['spectral_index'] = self.spectral_index
if self.component_type == 'healpix':
other['nside'] = self.nside
other['hpx_inds'] = self.hpx_inds
else:
other['name'] = self.name
return pyradiosky.SkyModel(
ra=ra_use, dec=dec_use, stokes=stokes_use,
spectral_type=self.spectral_type, **other
)
if isinstance(array_location, MoonLocation):
localframe = 'lunartopo'
mock_keywords['world'] = 'moon'
else:
localframe = 'altaz'
mock_keywords['world'] = 'earth'
source_coord = SkyCoord(alt=Angle(alts, unit=units.deg), az=Angle(azs, unit=units.deg),
obstime=time, frame=localframe, location=array_location)
icrs_coord = source_coord.transform_to('icrs')
ra = icrs_coord.ra
dec = icrs_coord.dec
names = np.array(['src' + str(si) for si in range(Nsrcs)])
stokes = np.zeros((4, 1, Nsrcs))
stokes[0, :] = fluxes
catalog = pyradiosky.SkyModel(names, ra, dec, stokes, 'flat')
if return_data:
catalog = SkyModelData(catalog)
if get_rank() == 0 and save:
np.savez('mock_catalog_' + arrangement, ra=ra.rad, dec=dec.rad, alts=alts, azs=azs,
fluxes=fluxes)
return catalog, mock_keywords
warnings.warn(
"Warning: No julian date given for mock catalog. Defaulting to first time step."
)
else:
raise ValueError(
"input_uv must be supplied if using mock catalog without specified julian date"
)
time = mock_keywords.pop('time')
sky, mock_keywords = create_mock_catalog(time, **mock_keywords)
mock_keyvals = [str(key) + str(val) for key, val in mock_keywords.items()]
source_list_name = 'mock_' + "_".join(mock_keyvals)
elif isinstance(catalog, str):
source_list_name = os.path.basename(catalog)
sky = pyradiosky.SkyModel()
if not os.path.isfile(catalog):
catalog = os.path.join(param_dict['config_path'], catalog)
if catalog.endswith("txt"):
sky.read_text_catalog(catalog)
elif catalog.endswith('vot'):
if 'gleam' in catalog:
if "spectral_type" in source_params:
spectral_type = source_params["spectral_type"]
else:
warnings.warn("No spectral_type specified for GLEAM, using 'flat'.")
spectral_type = "flat"
sky.read_gleam_catalog(
catalog, spectral_type=spectral_type
)
else:
vo_params = {}
for a1 in range(Nants):
for a2 in range(a1, Nants):
if bi >= Nbls:
break
blsel.append('({},{})'.format(a1, a2))
bi += 1
# ----------------
# Sky Model setup
# Full frequency HEALPix shell.
# ----------------
fmin, fmax = 0.0, 1.0 # K (fluxes)
skydat = np.random.uniform(fmin, fmax, (Nfreqs, Nsrcs))
write_healpix_hdf5(os.path.join(confdir, hpx_fname), skydat, range(Nsrcs), freqs)
# ----------------
# Make config dictionaries
# ----------------
filedict = {
'outdir': outdir,
'outfile_name': 'benchmark',
'output_format': 'uvh5'
}
freqdict = freq_array_to_params(freqs)
srcdict = {
'catalog': hpx_fname
}
if bi >= Nbls:
break
blsel.append('({},{})'.format(a1, a2))
bi += 1
# ----------------
# Sky Model setup
# Full frequency HEALPix shell.
# ----------------
fmin, fmax = 0.0, 1.0 # K (fluxes)
skydat = np.random.uniform(fmin, fmax, (Nfreqs, Nsrcs))
hpx_fname = 'benchmark_skymodel.hdf5'
write_healpix_hdf5(os.path.join(confdir, hpx_fname), skydat, range(Nsrcs), freqs)
# ----------------
# Make config dictionaries
# ----------------
filedict = {
'outdir': outdir,
'outfile_name': 'benchmark',
'output_format': 'uvh5'
}
freqdict = freq_array_to_params(freqs)
srcdict = {
'catalog': hpx_fname