Secure your code as it's written. Use Snyk Code to scan source code in minutes - no build needed - and fix issues immediately.
raise ValueError("Bad number of channels to be written (%d). " \
"Check lo/hi frequencies." % new_nchans)
print("Will extract")
print(" %d bins (%d to %d incl.)" % (new_nspec, startbin, endbin-1))
print(" (Original num bins: %d)" % fil.nspec)
print(" %d channels (%d to %d incl.)" % (new_nchans, lochan, hichan-1))
print(" (Original num chans: %d)" % fil.nchans)
# Create output file
outfn = options.outname % fil.header
print("Creating out file: %s" % outfn)
outhdr = copy.deepcopy(fil.header)
outhdr['nchans'] = new_nchans
outhdr['fch1'] = fil.frequencies[lochan]
filterbank.create_filterbank_file(outfn, outhdr, nbits=fil.nbits)
outfil = filterbank.FilterbankFile(outfn, mode='write')
# Write data
sys.stdout.write(" %3.0f %%\r" % 0)
sys.stdout.flush()
nblocks = int(new_nspec/options.block_size)
remainder = new_nspec % options.block_size
oldprogress = -1
for iblock in np.arange(nblocks):
lobin = iblock*options.block_size + startbin
hibin = lobin+options.block_size
spectra = fil.get_spectra(lobin, hibin)
spectra = spectra[:,lochan:hichan] # restrict channels
outfil.append_spectra(spectra)
progress = int(100.0*((hibin-startbin)/new_nspec))
if progress > oldprogress:
delays = psr_utils.delay_from_DM(dm, fil.frequencies)
delays -= delays[np.argmax(fil.frequencies)]
get_phases = lambda times: (times-delays)/period % 1
else:
get_phases = lambda times: times/period % 1
# Create the output filterbank file
if nbitsout is None:
nbitsout = fil.nbits
if inplace:
warnings.warn("Injecting pulsar signal *in-place*")
outfil = fil
else:
# Start an output file
print("Creating out file: %s" % outfn)
outfil = filterbank.create_filterbank_file(outfn, fil.header, \
nbits=nbitsout, mode='append')
if outfil.nbits == 8:
raise NotImplementedError("This code is out of date. 'delays' is not " \
"done in this way anymore..")
# Read the first second of data to get the global scaling to use
onesec = fil.get_timeslice(0, 1).copy()
onesec_nspec = onesec.shape[0]
times = np.atleast_2d(np.arange(onesec_nspec)*fil.tsamp).T+delays
phases = times/period % 1
onesec += prof(phases)
minimum = np.min(onesec)
median = np.median(onesec)
# Set median to 1/3 of dynamic range
global_scale = (256.0/3.0) / median
del onesec
def main(fits_fn, outfn, nbits, \
apply_weights, apply_scales, apply_offsets):
start = time.time()
psrfits_file = psrfits.PsrfitsFile(fits_fn)
fil_header = translate_header(psrfits_file)
fil_header['nbits'] = nbits
outfil = filterbank.create_filterbank_file(outfn, fil_header, \
nbits=nbits)
# if frequency channels are in ascending order
# band will need to be flipped
if psrfits_file.fits['SUBINT'].header['CHAN_BW'] > 0:
flip_band=True
print("\nFits file frequencies in ascending order.")
print("\tFlipping frequency band.\n")
else:
flip_band=False
# check nbits for input
if psrfits_file.nbits < 4:
raise ValueError('Does not support %d-bit data' % \
psrfits_file.nbits)
output_subints, skip):
start = time.time()
firstfits = pyfits.open(fits_fn[0],memmap=True)
nchan = firstfits['SUBINT'].header['NCHAN']
nsamps = firstfits['SUBINT'].header['NSBLK']
nsubints = firstfits['SUBINT'].header['NAXIS2']
fil_header = translate_header(firstfits,skip,output_subints)
fil_header['nbits'] = nbits
outfil = filterbank.create_filterbank_file(outfn, fil_header, \
nbits=nbits)
# if frequency channels are in ascending order
# band will need to be flipped
if firstfits['SUBINT'].header['CHAN_BW'] > 0:
flip_band=True
print("\nFits file frequencies in ascending order.")
print("\tFlipping frequency band.\n")
else:
flip_band=False
# check nbits for input
input_nbits = firstfits['SUBINT'].header['NBITS']
if input_nbits < 4:
raise ValueError('Does not support %d-bit data' % input_nbits)