API Reference

pygsdata

An interface for 21cm Global Signal Data.

class pygsdata.GSData(telescope: Telescope, data: _Buffer | _SupportsArray[dtype[Any]] | _NestedSequence[_SupportsArray[dtype[Any]]] | complex | bytes | str | _NestedSequence[complex | bytes | str] | Quantity | None, freqs: _Buffer | _SupportsArray[dtype[Any]] | _NestedSequence[_SupportsArray[dtype[Any]]] | complex | bytes | str | _NestedSequence[complex | bytes | str] | Quantity | None, times: Time, pols: Sequence[str] = NOTHING, effective_integration_time: _Buffer | _SupportsArray[dtype[Any]] | _NestedSequence[_SupportsArray[dtype[Any]]] | complex | bytes | str | _NestedSequence[complex | bytes | str] | Quantity | None = NOTHING, nsamples: _Buffer | _SupportsArray[dtype[Any]] | _NestedSequence[_SupportsArray[dtype[Any]]] | complex | bytes | str | _NestedSequence[complex | bytes | str] | Quantity | None = NOTHING, loads=NOTHING, flags: dict[str, GSFlag] = NOTHING, history: History = NOTHING, residuals: _Buffer | _SupportsArray[dtype[Any]] | _NestedSequence[_SupportsArray[dtype[Any]]] | complex | bytes | str | _NestedSequence[complex | bytes | str] | Quantity | None = None, data_unit: Literal['power', 'temperature', 'uncalibrated', 'uncalibrated_temp'] = 'power', auxiliary_measurements=None, time_ranges: Time = NOTHING, lsts=NOTHING, lst_ranges=NOTHING, filename=None, file_appendable=True, name='')

A generic container for Global-Signal data.

Parameters:

• data (numpy.ndarray) – The data array (i.e. what the telescope measures). This must be a 4D array whose dimensions are (load, polarization, time, frequency). The data can be raw powers, calibrated temperatures, or even model residuals to such. Their type is specified by the data_unit attribute.

• freqs (astropy.units.quantity.Quantity) – The frequency array. This must be a 1D array of frequencies specified as an astropy Quantity.

• times (astropy.time.core.Time) – The time array. This must be a 2D array of shape (times, loads). It can be in one of two formats: either an astropy Time object, specifying the absolute time, or an astropy Longitude object, specying the LSTs. In “lst” mode, there are many methods that become unavailable.

• telescope_location – The telescope location. This must be an astropy EarthLocation object.

• loads (tuple[str]) – The names of the loads. Usually there is a single load (“ant”), but arbitrary loads may be specified.

• nsamples (numpy.ndarray) – An array with the same shape as the data array, specifying the number of samples that go into each data point. This is unitless, and can be used with the effective_integration_time attribute to compute the total effective integration time going into any measurement.

• effective_integration_time – An astropy Quantity that specifies the amount of time going into a single “sample” of the data. This can either be a scalar, or a 4D array with the same shape as the data array. If it is a scalar, it is assumed to be the same for all data points. The default value is the integration time of the telescope. Note that this value is only meant to be used to track the expected noise level in the data, in conjunction with nsamples. It is not checked for whether the time_ranges match the integration time (since the effective time can be smaller than the time range due to windowing, or because the time_range includes multiple observations).

• flags (dict[str, pygsdata.gsflag.GSFlag]) – A dictionary mapping filter names to boolean arrays. Each boolean array has the same shape as the data array, and is True where the data is flagged.

• history (pygsdata.history.History) – A tuple of dictionaries, each of which is a record of a previous processing step.

• telescope_name – The name of the telescope.

• residuals (numpy.ndarray | None) – An optional array of the same shape as data that holds the residuals of a model fit to the data.

• auxiliary_measurements (astropy.table.table.QTable | None) – A dictionary mapping measurement names to arrays. Each array must have its leading axis be the same length as the time array.

• filename (pathlib.Path | None) – The filename from which the data was read (if any). Used for writing additional data if more is added (eg. flags, data model).

add_flags(filt: str, flags: ndarray | GSFlag | Path, append_to_file: bool = False)

Append a set of flags to the object and optionally append them to file.

You can always write out a new file, but appending flags is non-destructive, and so we allow it to be appended, in order to save disk space and I/O.

property complete_flags: ndarray

Returns the complete flag array.

property effective_integration_time: Annotated[Quantity, Unit('s')]

The effective integration time.

property flagged_nsamples: ndarray

Weights accounting for all flags.

freq_iter() → Iterable[tuple[slice, slice, slice]]

Return an iterator over the frequency axis of data-shape arrays.

classmethod from_file(filename: str | Path, reader: str | None = None, selectors: dict[str, Any] | None = None, concat_axis: Literal['load', 'pol', 'time', 'freq'] | None = None, **kw) → Self

Create a GSData instance from a file.

This method attempts to auto-detect the file type and read it.

get_cumulative_flags(which_flags: tuple[str] | None = None, ignore_flags: tuple[str] = ()) → ndarray

Return accumulated flags.

get_flagged_nsamples(which_flags: tuple[str, ...] | None = None, ignore_flags: tuple[str, ...] = ()) → ndarray

Get the nsamples of the data after accounting for flags.

get_initial_yearday(hours: bool = False, minutes: bool = False) → str

Return the year-day representation of the first time-sample in the data.

get_moon_azel() → tuple[ndarray, ndarray]

Get the Moon’s azimuth and elevation for each time in deg.

get_sun_azel() → tuple[ndarray, ndarray]

Get the Sun’s azimuth and elevation for each time in deg.

property gha: ndarray

The GHA’s of the observations.

load_iter() → Iterable[tuple[int]]

Return an iterator over the load axis of data-shape arrays.

property model: ndarray | None

The model of the data.

property nflagging_ops: int

Returns the number of flagging operations.

property nfreqs: int

The number of frequency channels.

property nloads: int

The number of loads.

property npols: int

The number of polarizations.

property ntimes: int

The number of times.

remove_flags(filt: str) → Self

Remove flags for a given filter.

time_iter() → Iterable[tuple[slice, slice, slice]]

Return an iterator over the time axis of data-shape arrays.

update(**kwargs)

Return a new GSData object with updated attributes.

write_gsh5(filename: str | Path, group: str = '/') → Self

Write the data in the GSData object to a GSH5 file.

class pygsdata.GSFlag(flags: _Buffer | _SupportsArray[dtype[Any]] | _NestedSequence[_SupportsArray[dtype[Any]]] | complex | bytes | str | _NestedSequence[complex | bytes | str] | Quantity | None, axes=NOTHING, history: History = NOTHING, filename=None)

A generic container for Global-Signal flags data.

Parameters:

• flags (numpy.ndarray) – The flags as a boolean array. The array may have up to 4 dimensions – load, pol, time, and freq – but need not have all of the dimensions.

• axes (tuple[str]) – A tuple of strings specifying the axes of the data array. The possible axes are “load”, “pol”, “time”, and “freq”. They must be in that order, but not all must be present, only as many as flags has dimensions.

• history (pygsdata.history.History) – A tuple of dictionaries, each of which is a record of a previous processing step.

• filename (pathlib.Path | None) – The filename from which the data was read (if any). Used for writing additional data if more is added (eg. flags, data model).

all(axis: str | None = None) → bool | Self

Return True if any of the flags are True.

any(axis: str | None = None) → bool | Self

Return True if any of the flags are True.

concat(others: Self | Sequence[Self], axis: str) → Self

Get a new GSFlag by concatenating other flags to this one.

classmethod from_file(filename: str | Path, filetype: str | None = None, **kw) → Self

Create a GSFlag instance from a file.

This method attempts to auto-detect the file type and read it.

property full_rank_flags: ndarray

Return a full-rank flag array.

property nfreqs: int | None

The number of frequency channels.

property nloads: int | None

The number of loads.

property npols: int | None

The number of polarizations.

property ntimes: int | None

The number of times.

op_on_axis(op, axis: str) → Self

Apply an operation along the given axis.

classmethod read_gsflag(filename: str) → Self

Read a GSFlag file to create the object.

select(idx: ndarray | slice, axis: str, squeeze: bool = False) → Self

Select a subset of the data along the given axis.

update(**kwargs) → Self

Return a new GSFlag object with updated attributes.

write_gsflag(filename: str) → Self

Write the data in the GSData object to a GSH5 file.

class pygsdata.History(stamps=NOTHING)

A collection of Stamp objects defining the history.

add(stamp: Stamp | dict | tuple[Stamp] | tuple[dict] | Self)

Add a stamp to the history.

classmethod from_repr(repr_string: str)

Create a History object from a string representation.

pretty()

Return a rich-compatible string representation of the history.

class pygsdata.Stamp(message: str = '', function: str = '', parameters: dict = NOTHING, versions: dict = NOTHING, timestamp: datetime = NOTHING)

Class representing a historical record of a process applying to an object.

Parameters:

• message (str) – A message describing the process. Optional – either this or the function must be defined.

• function (str) – The name of the function that was applied. Optional – either this or the message must be defined.

• parameter(s) – The parameters passed to the function. Optional – if function is defined, this should be specified.

• versions (dict) – A dictionary of the versions of the software used to perform the process. Created by default when the History is created.

• timestamp (datetime.datetime) – A datetime object corresponding to the time the process was performed. By default, this is set to the time that the Stamp object is created.

classmethod from_repr(repr_string: str)

Create a Stamp object from a string representation.

classmethod from_yaml_dict(d: dict) → Self

Create a Stamp object from a dictionary representing a history record.

pretty()

Return a rich-compatible string representation of the history record.

class pygsdata.Telescope(*, name, location: ~astropy.coordinates.earth.EarthLocation, pols: ~collections.abc.Sequence[str], integration_time: ~astropy.units.quantity.Annotated[~astropy.units.quantity.Quantity, PhysicalType('time')] = <Quantity 1. s>, x_orientation=<Quantity 0. deg>)

Class representing a telescope.

Parameters:

• name (str) – The name of the telescope.

• location (astropy.coordinates.earth.EarthLocation) – The location of the telescope.

• pols (tuple[str]) – The polarizations that the telescope can measure. This should be all of the available polarizations that the telescope can measure, not just those that are in a particular observation. The polarizations should be given as a tuple of strings, where each string is one of “XX”, “XY”, “YX”, “YY”, “pI”, “pQ”, “pU”, or “pV”.

• integration_time (astropy.units.quantity.Quantity) – The integration time of the telescope. This should be a scalar Quantity. In principle each observation made by a telescope may have different integration time, but this is the default value.

• x_orientation (astropy.coordinates.angles.core.Angle) – The orientation of the X polarization. This should be an Angle, and represents the orientation with respect to East. The default is 0 degrees, which means that the X polarization is aligned with East (the angle rotates towards North).

classmethod from_hdf5(fname: str | Path | File | Group) → Self

Read a telescope from an HDF5 file.

write(fname: str | Path | File | Group)

Write the telescope to an HDF5 file.

pygsdata.gsdata_reader(formats: list[str], select_on_read: bool = False)

Register a function as a reader for GSData objects.

class pygsdata.gsregister(kind: Literal['gather', 'calibrate', 'filter', 'reduce', 'supplement'])

Decorator to register a function as a processor for GSData objects.

pygsdata.select_freqs(data: GSData, *, freq_range: tuple[Annotated[Quantity, PhysicalType('frequency')], Annotated[Quantity, PhysicalType('frequency')]] | None = None, indx: ndarray | None = None) → GSData

Select a subset of the frequency channels in a GSData object.

Parameters:

• data – The GSData object to downselect.

• freq_range – A two-tuple specifying the frequency range to select. The elements should each be astropy quantities with frequency-type units. For example (50*un.MHz, 100*un.MHz). If the values are simply floats (e.g. (50, 100)), a warning will be raised, and they will be assumed to be in MHz. This parameter is optional – you can instead specify indx if your selection is more involved.

• indx – An object that can slice a numpy array (i.e. a list of integers, a numpy array of integers, or a slice object). This will be used to directly select the frequency channels. This parameter is optional – you can instead specify freq_range. If both are specified, the returned frequencies will be the union of the selectors (i.e. in the freq range OR in the indx).

Notes

If both freq_range and indx are specified, they will be OR-ed, i.e. the returned frequencies will be either in the freq range or in the indx. To achieve and-like behavior, simply use the selector twice.

pygsdata.select_lsts(data: GSData, *, lst_range: tuple[Annotated[Quantity, Unit('hourangle')] | Longitude, Annotated[Quantity, Unit('hourangle')] | Longitude] | None = None, indx: ndarray | None = None, load: int | str = 'ant', gha: bool = False, **kwargs) → GSData

Select a subset of the times.

pygsdata.select_times(data: GSData, *, time_range: tuple[Time | float, Time | float] | None = None, fmt: str = 'jd', indx: ndarray | None = None, load: int | str = 'ant') → GSData

Select a subset of the times.