astrosaber.utils package#

Submodules#

astrosaber.utils.aslsq_fit module#

asymmetric least squares fit

astrosaber.utils.aslsq_fit.baseline_als(y: ndarray, lam: float, p: float, niter: int) → ndarray[source]#

Baseline smoothing using asymmetric least squares.

Parameters:

  • y (numpy.ndarray) – Spectrum to be smoothed.

  • lam (float) – Smoothing weight. Adjusts the amount of smoothing.

  • p (float) – Asymmetry weight. Adjusts how much weight positive or negative signals (wrt the smoothed baseline) will be given.

  • niter (int) – Number of iterations.

Returns:

z – Smoothed baseline.

Return type:

numpy.ndarray

astrosaber.utils.aslsq_fit.baseline_als_optimized(y: ndarray, lam: float, p: float, niter: int, mask: ndarray | None = None) → ndarray[source]#

Baseline smoothing using asymmetric least squares.

Parameters:

  • y (numpy.ndarray) – Spectrum to be smoothed.

  • lam (float) – Smoothing weight. Adjusts the amount of smoothing.

  • p (float) – Asymmetry weight. Adjusts how much weight positive or negative signals (wrt the smoothed baseline) will be given.

  • niter (int) – Number of iterations.

  • mask (numpy.ndarray) – Boolean mask indicating signal ranges, with True indicating signal, False indicating noise. Will be used to set asymmetry weights to 0.5 where there is noise.

Returns:

z – Smoothed baseline.

Return type:

numpy.ndarray

astrosaber.utils.aslsq_fit.one_step_extraction(lam1: float, p1: float, spectrum: ndarray | None = None, header: Header | None = None, check_signal_sigma: float = 6.0, noise: float | None = None, velo_range: float = 15.0, niters: int = 20, iterations_for_convergence: int = 3, add_residual: bool = False, thresh: float | None = None, p_limit: float = 0.02, cunit3: str = 'm/s') → Tuple[ndarray, ndarray, int, int][source]#

Baseline smoothing routine using one lambda smoothing value for all major iterations.

Parameters:

  • lam1 (float) – Smoothing weight. Adjusts the amount of smoothing.

  • p1 (float) – Asymmetry weight. Adjusts how much weight positive or negative signals (wrt the smoothed baseline) will be given.

  • spectrum (numpy.ndarray) – Spectrum to be smoothed.

  • header (~astropy.io.fits.Header) – Header of the file containing the spectrum. Will be passed to check_signal_ranges().

  • check_signal_sigma (float, optional) – Defines the significance of the signal that has to be present in the spectra for at least the range defined by velo_range. Default is 6.0.

  • noise (float) – Noise level of the data. Will be passed to check_signal_ranges()

  • velo_range (float, optional) – Velocity range [in km/s] of the spectra that has to contain significant signal for it to be considered in the baseline extraction. Default is 15.0. Will be passed to check_signal_ranges().

  • niters (int, optional) – Maximum number of iterations of the smoothing. Default is 20.

  • iterations_for_convergence (int, optional) – Number of iterations of the major cycle for the baseline to be considered converged. Default is 3.

  • add_residual (bool, optional) – Whether to add the residual (=difference between first and last major cycle iteration) to the baseline. Default is False.

  • thresh (float) – Convergence threshold. If residual falls below this threshold for iterations_for_convergence iterations, the algorithm terminates the smoothing.

  • p_limit (float, optional) – The p-limit of the Markov chain to estimate signal ranges in the spectra. Default is 0.02.

  • cunit3 (str, optional) – Type of velocity unit specified in the fits file header keyword ‘CUNIT3’. Default is ‘m/s’.

Returns:

  • bg (numpy.ndarray) – Background spectrum w/o self-absorption.

  • hisa (numpy.ndarray) – Inverted self-absorption spectrum (i.e. expressed as equivalent emission).

  • iterations (int) – Number of iterations until algorithm converged.

  • flag_map (int) – Flag whether background did/did not converge or whether spectrum does/does not contain signal. If flag is 1, the were no issues in the fit. If 0, fit did not converge or did not contain signal.

astrosaber.utils.aslsq_fit.two_step_extraction(lam1: float, p1: float, lam2: float, p2: float, spectrum: ndarray | None = None, header: Header | None = None, check_signal_sigma: float = 6.0, noise: float | None = None, velo_range: float = 15.0, niters: int = 20, iterations_for_convergence: int = 3, add_residual: bool = False, thresh: float | None = None, p_limit: float = 0.02, cunit3: str = 'm/s') → Tuple[ndarray, ndarray, int, int][source]#

Baseline smoothing routine using one lambda smoothing value for all major iterations.

Parameters:

  • lam1 (float) – Smoothing weight of the very first smoothing iteration. Adjusts the amount of smoothing.

  • p1 (float) – Asymmetry weight of the very first smoothing iteration. Adjusts how much weight positive or negative signals (wrt the smoothed baseline) will be given.

  • lam2 (float) – Smoothing weight of all remaining smoothing iterations. Adjusts the amount of smoothing.

  • p2 (float) – Asymmetry weight of all remaining smoothing iterations. Adjusts how much weight positive or negative signals (wrt the smoothed baseline) will be given.

  • spectrum (numpy.ndarray) – Spectrum to be smoothed.

  • header (~astropy.io.fits.Header) – Header of the file containing the spectrum. Will be passed to check_signal_ranges().

  • check_signal_sigma (float, optional) – Defines the significance of the signal that has to be present in the spectra for at least the range defined by velo_range. Default is 6.0.

  • noise (float) – Noise level of the data. Will be passed to check_signal_ranges()

  • velo_range (float, optional) – Velocity range [in km/s] of the spectra that has to contain significant signal for it to be considered in the baseline extraction. Default is 15.0. Will be passed to check_signal_ranges().

  • niters (int, optional) – Maximum number of iterations of the smoothing. Default is 20.

  • iterations_for_convergence (int, optional) – Number of iterations of the major cycle for the baseline to be considered converged. Default is 3.

  • add_residual (bool, optional) – Whether to add the residual (=difference between first and last major cycle iteration) to the baseline. Default is False.

  • thresh (float) – Convergence threshold. If residual falls below this threshold for iterations_for_convergence iterations, the algorithm terminates the smoothing.

  • p_limit (float, optional) – The p-limit of the Markov chain to estimate signal ranges in the spectra. Default is 0.02.

  • cunit3 (str, optional) – Type of velocity unit specified in the fits file header keyword ‘CUNIT3’. Default is ‘m/s’.

Returns:

  • bg (numpy.ndarray) – Background spectrum w/o self-absorption.

  • hisa (numpy.ndarray) – Inverted self-absorption spectrum (i.e. expressed as equivalent emission).

  • iterations (int) – Number of iterations until algorithm converged.

  • flag_map (int) – Flag whether background did/did not converge or whether spectrum does/does not contain signal. If flag is 1, the were no issues in the fit. If 0, fit did not converge or did not contain signal.

astrosaber.utils.aslsq_helper module#

helper functions

exception astrosaber.utils.aslsq_helper.IterationWarning[source]#

Bases: UserWarning

Passing on diagnostic messages.

astrosaber.utils.aslsq_helper.check_signal(spectrum: ndarray, sigma: float, noise: float) → bool[source]#

Check for signal in an array given a significance threshold.

Parameters:

  • spectrum (numpy.ndarray) – Data to check for signal.

  • sigma (float) – Significance of the signal to be checked for.

  • noise (float) – Noise level. Signal threshold will be sigma * noise.

Returns:

check – True if there any data point is significant, False if everything below threshold.

Return type:

bool

astrosaber.utils.aslsq_helper.check_signal_ranges(spectrum, header, sigma=None, noise=None, velo_range=None, cunit3='m/s')[source]#

Check for continuous signal range in an array given a significance threshold.

Parameters:

  • spectrum (numpy.ndarray) – Data to check for signal.

  • header – Header of the file containing the spectrum. This is required to read out the velocity resolution.

  • sigma (float) – Significance of the signal to be checked for.

  • noise (float) – Noise level. Signal threshold will be sigma * noise.

  • velo_range (float) – Velocity range [in km/s] of the spectrum that has to contain continuous significant signal.

  • cunit3 (str, optional) – Type of velocity unit specified in the fits file header keyword ‘CUNIT3’. Default is ‘m/s’.

Returns:

check – True if there is continuous signal, False if no signal.

Return type:

bool

astrosaber.utils.aslsq_helper.count_ones_in_row(data: ndarray) → ndarray[source]#

Counts number of continuous trailing ‘1’s.

Parameters:

data (numpy.ndarray) – Data containing 0’s and 1’s.

Returns:

output – Array containing the number of continuous 1’s to the point of each index.

Return type:

numpy.ndarray

astrosaber.utils.aslsq_helper.find_nearest(array: ndarray, value: float) → int[source]#

Find the index of an element in an array nearest to a given value.

Parameters:

  • array (numpy.ndarray) – Input array to index.

  • value (float) – Value of the element to find the closest index for.

Returns:

idx – Index of the element with value closest to value.

Return type:

int

astrosaber.utils.aslsq_helper.format_warning(message, category, filename, lineno, file=None, line=None)[source]#

Print warning message.

astrosaber.utils.aslsq_helper.md_header_2d(fitsfile: Path | str) → Header[source]#

Get 2D header from FITS file.

Parameters:

fitsfile (path-like object or file-like object) – Path to FITS file to get header from.

Returns:

header_2d – Header object without third axis.

Return type:

~astropy.io.fits.Header

astrosaber.utils.aslsq_helper.merge_ranges(ranges: ndarray) → ndarray[source]#

Merge intervals where they overlap.

Parameters:

ranges (numpy.ndarray) – Array of signal intervals indicating the start and end index.

Returns:

merged_ranges – Array of merged ranges.

Return type:

numpy.ndarray

astrosaber.utils.aslsq_helper.pixel_to_world(fitsfile: Path, x: float, y: float, ch: float | None = 0.0)[source]#

Convert pixel coordinates to world coordinates from a FITS file.

Parameters:

  • fitsfile (str) – Path to FITS file to get coordinates from.

  • x (float) – Pixel coordinate on the x-axis of the FITS file.

  • y (float) – Pixel coordinate on the y-axis of the FITS file.

  • ch (float, optional) – Velocity channel to convert (default is 0.).

Returns:

result –

Returns the world coordinates. If the input was a single array and origin,

a single array is returned, otherwise a tuple of arrays is returned.

Return type:

numpy.ndarray

astrosaber.utils.aslsq_helper.say(message, verbose=True, end=None)[source]#

Diagnostic messages.

astrosaber.utils.aslsq_helper.velocity_axes(name: Path, cunit3: str = 'm/s') → ndarray[source]#

Get velocity axis from FITS file in units of km/s.

Parameters:

  • name (Path) – Path to FITS file to get velocity axis from.

  • cunit3 (str, optional) – Type of velocity unit specified in the fits file header keyword ‘CUNIT3’. Default is ‘m/s’.

Returns:

velocity – Array of velocity axis.

Return type:

numpy.ndarray

astrosaber.utils.grogu module#

class astrosaber.utils.grogu.TqdmDefaultWriteLock[source]#

Bases: object

Methods

acquire

release
acquire()[source]#
release()[source]#
class astrosaber.utils.grogu.yoda(*_, **__)[source]#

Bases: tqdm

Attributes:
format_dict

Public API for read-only member access.

monitor

Methods

clear([nolock])

Clear current bar display.

close()

Cleanup and (if leave=False) close the progressbar.

display([msg, pos])

Use self.sp to display msg in the specified pos.

external_write_mode([file, nolock])

Disable tqdm within context and refresh tqdm when exits.

format_interval(t)

Formats a number of seconds as a clock time, [H:]MM:SS

format_meter(n, total, elapsed[, ncols, ...])

Return a string-based progress bar given some parameters

format_num(n)

Intelligent scientific notation (.3g).

format_sizeof(num[, suffix, divisor])

Formats a number (greater than unity) with SI Order of Magnitude prefixes.

get_lock()

Get the global lock.

pandas(**tqdm_kwargs)

Registers the current tqdm class with

refresh([nolock, lock_args])

Force refresh the display of this bar.

reset([total])

Resets to 0 iterations for repeated use.

set_description([desc, refresh])

Set/modify description of the progress bar.

set_description_str([desc, refresh])

Set/modify description without ': ' appended.

set_lock(lock)

Set the global lock.

set_postfix([ordered_dict, refresh])

Set/modify postfix (additional stats) with automatic formatting based on datatype.

set_postfix_str([s, refresh])

Postfix without dictionary expansion, similar to prefix handling.

status_printer(file)

Manage the printing and in-place updating of a line of characters.

unpause()

Restart tqdm timer from last print time.

update([n])

Manually update the progress bar, useful for streams such as reading files.

wrapattr(stream, method[, total, bytes])

stream : file-like object. method : str, "read" or "write". The result of read() and the first argument of write() should have a len().

write(s[, file, end, nolock])

Print a message via tqdm (without overlap with bars).

moveto
monitor = None#
monitor_interval = 10#

astrosaber.utils.quality_checks module#

astrosaber.utils.quality_checks.determine_peaks(spectrum: ndarray, peak: str = 'both', amp_threshold: float | None = None) → Tuple[ndarray, List][source]#

Find peaks in a spectrum.

Parameters:

  • spectrum (numpy.ndarray) – Array of the data values of the spectrum.

  • peak (str) – ‘both’ (default), ‘positive’, ‘negative’ Description of parameter peak.

  • amp_threshold (float) – Required minimum threshold that at least one data point in a peak feature has to exceed.

Returns:

  • consecutive_channels or amp_vals (numpy.ndarray) – If the ‘amp_threshold’ value is supplied an array with the maximum data values of the ranges is returned. Otherwise, the number of spectral channels of the ranges is returned.

  • ranges (list) – List of intervals [(low, upp), …] determined to contain peaks.

astrosaber.utils.quality_checks.get_max_consecutive_channels(n_channels: int, p_limit: float) → int[source]#

Determine the maximum number of random consecutive positive/negative channels. Calculate the number of consecutive positive or negative channels, whose probability of occurring due to random chance in a spectrum is less than p_limit.

Parameters:

  • n_channels (int) – Number of spectral channels.

  • p_limit (float) – Maximum probability for consecutive positive/negative channels being due to chance.

Returns:

consec_channels – Number of consecutive positive/negative channels that have a probability less than p_limit to be due to chance.

Return type:

int

astrosaber.utils.quality_checks.goodness_of_fit(data: ndarray, fit: ndarray, errors: ndarray | float, dof: int, mask: ndarray | None = None, get_aicc: bool = False) → float[source]#

Determine the goodness of fit (reduced chi-square, AICc).

Parameters:

  • data (numpy.ndarray) – Original data.

  • fit (numpy.ndarray) – Fit to the original data.

  • errors (numpy.ndarray or float) – Root-mean-square noise for each channel.

  • dof (int) – Degrees of freedom.

  • mask (numpy.ndarray) – Boolean array specifying which regions of the spectrum should be used.

  • get_aicc (bool) – If set to True, the AICc value will be returned in addition to the reduced chi2 value.

Returns:

  • rchi2 (float) – Reduced chi2 value.

  • aicc (float) – (optional:) The AICc value is returned if get_aicc is set to True. Default is False.

astrosaber.utils.quality_checks.mask_channels(n_channels: int, ranges: List, pad_channels: int | None = None, remove_intervals: List | None = None) → ndarray[source]#

Determine the 1D boolean mask for a given list of spectral ranges.

Parameters:

  • n_channels (int) – Number of spectral channels.

  • ranges (list) – List of intervals [(low, upp), …].

  • pad_channels (int) – Number of channels by which an interval (low, upp) gets extended on both sides, resulting in (low - pad_channels, upp + pad_channels).

  • remove_intervals (list) – Nested list containing info about ranges of the spectrum that should be masked out.

Returns:

mask – 1D boolean mask that has ‘True’ values at the position of the channels contained in ranges.

Return type:

numpy.ndarray

astrosaber.utils.spectrum_utils module#

spectrum utils

astrosaber.utils.spectrum_utils.calculate_spectrum(fitsfile: Path | str, pixel_array: List) → ndarray[source]#

Calculate an average spectrum given a p-p-v FITS cube and pixel coordinates. If NaN values are present at specific coordinates, these coordinates will be ignored.

Parameters:

  • fitsfile (Path | str) – Path to FITS file to get average spectrum from.

  • pixel_array (List) –

    List of tuples containing pixel coordinates [(y0,x0),(y1,x1),…]

    over which to average.

Returns:

spectrum_average – Averaged spectrum.

Return type:

numpy.ndarray

astrosaber.utils.spectrum_utils.pixel_circle_calculation(fitsfile: Path | str, glon: float, glat: float, r: float) → List[source]#

Extract a list of pixels [(y0,x0),(y1,x1),…] corresponding to the circle region with central coordinates glon, glat, and radius r.

Parameters:

  • fitsfile (Path | str) – Path to FITS file.

  • glon (float) – x-coordinate of central pixel in units given in the header.

  • glat (float) – y-coordinate of central pixel in units given in the header.

  • r (float) – Radius of region in units of arcseconds.

Returns:

pixel_array – List of pixel coordinates [(y0,x0),(y1,x1),…].

Return type:

List

astrosaber.utils.spectrum_utils.pixel_circle_calculation_px(fitsfile: Path | str, x: float, y: float, r: float) → List[source]#

Extract a list of pixels [(y0,x0),(y1,x1),…] corresponding to the circle region with central pixels x, y, and radius r.

Parameters:

  • fitsfile (Path | str) – Path to FITS file.

  • x (float) – Central x-pixel.

  • y (float) – Central y-pixel.

  • r (float) – Radius of region in units of arcseconds.

Returns:

pixel_array – List of pixel coordinates [(y0,x0),(y1,x1),…].

Return type:

List

Module contents#