Pointing grids (pyobs.utils.grids)

Used by PointingSeries to define a sequence of sky positions (a grid) to point at in turn, optionally filtered/transformed by a chain of GridFilter steps.

class GridNode(log: bool = False, **kwargs: Any)[source]

Bases: Object

Abstract base class for grid nodes.

A GridNode implements the Python iterator protocol to produce a sequence of 2D points. Points can be either:

  • Tuples of floats representing (x, y) in degrees, or

  • astropy.coordinates.SkyCoord instances (e.g., in RA/Dec or Alt/Az frames).

Subclasses must implement:

  • _get_next(): fetches the next point

  • __len__(): number of remaining points (if known)

  • append_last(): append the last yielded point back to the sequence

  • log_last(): log the last yielded point

Logging is optional and can be enabled via the log flag. If enabled and the point is a SkyCoord, RA/Dec or Alt/Az are logged in human-readable form.

Initialize a GridNode.

Parameters:

  • log – If True, enable informational logging for picked points.

  • **kwargs – Additional keyword arguments forwarded to Object.__init__().

abstractmethod append_last() → None[source]

Append the last yielded point back to the underlying sequence.

This can be used to “re-queue” the most recently returned point.

Raises:

RuntimeError – If no last point exists to append.

log(point: tuple[float, float] | SkyCoord) → None[source]

Log a point if logging is enabled.

For SkyCoord instances, logs RA/Dec in hmsdms if available, otherwise Alt/Az in degrees if available. No logging occurs if log flag is False.

Parameters:

point – The point to log; either a (x, y) tuple in degrees or a SkyCoord.

abstractmethod log_last() → None[source]

Log the last yielded point, if any.

Implementations typically delegate to self.log(self._last).

Raises:

RuntimeError – If there is no last point to log.

class Grid(points: list[tuple[float, float] | SkyCoord], **kwargs: Any)[source]

Bases: GridNode

Abstract base class for grids backed by a mutable list of points.

This class consumes a list of points (tuples or SkyCoord) in FIFO order. It implements iteration, length, and appending the last element back to the list, enabling simple queue-like behavior.

Initialize a Grid with a list of points.

Parameters:

  • points – Initial list of points to yield. Each element must be either a (x, y) tuple in degrees or a SkyCoord.

  • **kwargs – Additional keyword arguments forwarded to GridNode.__init__().

append_last() → None[source]

Append the last yielded point to the end of the grid.

log_last() → None[source]

Log the last yielded point, if logging is enabled.

class RegularSphericalGrid(n_lon: int, n_lat: int, **kwargs: Any)[source]

Bases: Grid

Grid over a sphere using regular longitude/latitude sampling.

Produces points at intersections of equally spaced longitudes and latitudes:

  • Longitudes: [0, 360) with step 360 / n_lon (360 is excluded)

  • Latitudes: Inclusive linspace from -90 to +90 with n_lat values

The resulting points are (lon_deg, lat_deg) tuples in degrees.

Create a regular lon/lat grid.

Parameters:

  • n_lon – Number of longitudinal divisions. Must be > 0.

  • n_lat – Number of latitudinal points. Must be >= 2 to include both poles.

  • **kwargs – Additional keyword arguments forwarded to Grid.__init__().

Raises:

ValueError – If n_lon <= 0 or n_lat <= 0.

Notes

Longitudes are in [0, 360) and latitudes are in [-90, 90].

class GraticuleSphericalGrid(n: int, **kwargs: Any)[source]

Bases: Grid

Grid with approximately equidistributed points on a sphere.

Uses a graticule-like construction for near-uniform sampling over the sphere following the approach described by Deserno (2004):

Produces points as (lon_deg, lat_deg) tuples in degrees.

Create an approximately equidistributed spherical grid.

Parameters:

  • n – Target number of points on the sphere. Must be > 0.

  • **kwargs – Additional keyword arguments forwarded to Grid.__init__().

Raises:

ValueError – If n <= 0.

Notes

The actual number of generated points follows the construction and may be close to, but not necessarily exactly, n.

class GridPipeline(steps: list[GridNode | dict[str, Any]], **kwargs: Any)[source]

Bases: GridNode

A pipeline that composes a grid and a sequence of filters.

The pipeline expects a list of steps where the first element constructs a GridNode (e.g., a Grid) and subsequent elements construct GridNode filters that wrap the previous step. Each step can be either:

  • An instance of GridNode, or

  • A dict specification understood by Object.get_object() to construct a GridNode.

The pipeline itself is a GridNode that delegates iteration and other methods to the constructed chain.

Build a GridPipeline from a list of steps.

Parameters:

  • steps – A non-empty list where steps[0] yields a GridNode, and subsequent elements are GridNode filter specifications. Dict entries are passed to get_object() for construction.

  • **kwargs – Additional keyword arguments forwarded to Object.__init__().

Notes

If steps is empty, the pipeline is constructed in an “empty” state and iteration will immediately raise StopIteration.

append_last() → None[source]

Append the last yielded point back to the pipeline’s final stage.

log_last() → None[source]

Log the last yielded point via the pipeline’s final stage.

GridFilter

class GridFilter(grid: GridNode, **kwargs: Any)[source]

Bases: GridNode

Abstract base class for grid filters that wrap another GridNode.

A GridFilter delegates iteration and length to the wrapped grid, and can override _get_next() to transform or filter points. It also proxies append_last() and log_last() to the underlying grid, while optionally adding its own logging.

Initialize a filter with an underlying grid.

Parameters:

  • grid – The upstream GridNode to read points from.

  • **kwargs – Additional keyword arguments forwarded to GridNode.__init__().

append_last() → None[source]

Append the last yielded point back to the underlying grid.

log_last() → None[source]

Log the last point via the underlying grid, then log locally if enabled.

class GridFilterValue(grid: Grid | GridFilter, x_gt: int | None = None, x_gte: int | None = None, x_lt: int | None = None, x_lte: int | None = None, y_gt: int | None = None, y_gte: int | None = None, y_lt: int | None = None, y_lte: int | None = None, **kwargs: object)[source]

Bases: GridFilter

Filter points by numeric constraints on x and y.

Accepts points as:

  • (x, y) tuples (degrees), or

  • SkyCoord with RA/Dec (x=RA deg, y=Dec deg), or

  • SkyCoord with Alt/Az (x=Az deg, y=Alt deg).

Each constraint is optional; only provided constraints are applied. Points not satisfying all constraints are skipped.

Initialize the value-based filter.

Parameters:

  • grid – Upstream grid or filter.

  • x_gt – Keep points with x > x_gt (degrees).

  • x_gte – Keep points with x >= x_gte (degrees).

  • x_lt – Keep points with x < x_lt (degrees).

  • x_lte – Keep points with x <= x_lte (degrees).

  • y_gt – Keep points with y > y_gt (degrees).

  • y_gte – Keep points with y >= y_gte (degrees).

  • y_lt – Keep points with y < y_lt (degrees).

  • y_lte – Keep points with y <= y_lte (degrees).

  • **kwargs – Additional keyword arguments forwarded to GridFilter.__init__().

Notes

Constraints are combined conjunctively (logical AND). If multiple constraints on the same axis are provided, they are all applied.

class ConvertGridToSkyCoord(grid: Grid | GridFilter, frame: str = 'altaz', **kwargs: object)[source]

Bases: GridFilter

Convert (x, y) degree tuples to SkyCoord objects.

Wraps a tuple-producing grid and converts each point to a SkyCoord in the requested frame. The current time (Time.now()) is used as obstime. The location is taken from this object’s observer (inherited from Object).

Initialize the conversion filter.

Parameters:

  • grid – Upstream grid or filter that yields (x, y) tuples in degrees.

  • frame – Target frame for the SkyCoord (e.g., ‘altaz’, ‘icrs’, ‘galactic’).

  • **kwargs – Additional keyword arguments forwarded to GridFilter.__init__().

Notes

Requires that this object’s observer be defined (e.g., via the ‘observer’ or ‘location’ constructor argument) for frames that need a location (such as AltAz).

class ConvertGridFrame(grid: Grid | GridFilter, frame: str = 'altaz', **kwargs: object)[source]

Bases: GridFilter

Transform SkyCoord points to a different frame.

Initialize the frame conversion filter.

Parameters:

  • grid – Upstream grid or filter yielding SkyCoord objects.

  • frame – Target frame name passed to SkyCoord.transform_to().

  • **kwargs – Additional keyword arguments forwarded to GridFilter.__init__().

Notes

The ability to transform depends on the source and target frames; some transformations require obstime/location or other frame attributes.

class RandomizeGrid(grid: Grid | GridFilter, iterations: int = 50, **kwargs: object)[source]

Bases: GridFilter

Randomize iteration order by rotating the underlying sequence.

For each requested point, perform k random “rotations” by taking the next underlying point and appending it back (k chosen uniformly from [0, iterations)). Then yield the next point.

This preserves multiset contents while randomizing access order somewhat.

Initialize the randomizer.

Parameters:

  • grid – Upstream grid or filter.

  • iterations – Upper bound (exclusive) for the number of rotations per yield.

  • **kwargs – Additional keyword arguments forwarded to GridFilter.__init__().

Raises:

ValueError – If iterations < 0.

class AvoidMoon(grid: Grid | GridFilter, min_moon_distance: int = 20, **kwargs: object)[source]

Bases: GridFilter

Remove points too close to the moon.

If the next point in the underlying grid is too close to the moon, skip it.

Initialize the moon avoider.

Parameters:

  • grid – Upstream grid or filter.

  • min_moon_distance – Minimum distance to avoid the moon in degrees..

  • **kwargs – Additional keyword arguments forwarded to GridFilter.__init__().

Raises:

ValueError – If iterations < 0.

class FromList(grid: Grid | GridFilter, csv_file: str, max_distance: float = 1.0, field_ra: str = 'RA_ICRS', field_dec: str = 'DE_ICRS', **kwargs: object)[source]

Bases: GridFilter

Select closest point from a list.

Only select points if they are closer than a given distance to the grid point.

Initialize the moon avoider.

Parameters:

  • grid – Upstream grid or filter.

  • csv_file – Name of the csv file to read from.

  • max_distance – Maximum distance to grid point in degrees.

  • **kwargs – Additional keyword arguments forwarded to GridFilter.__init__().

Raises:

ValueError – If iterations < 0.