"""
This module defines the generic `Survey` and `GridSurvey` classes, combining healpix and polygon survey
capabilities with a camera footprint.
"""
import pandas
import numpy as np
from .healpix import HealpixSurvey
from .polygon import PolygonSurvey
# ================= #
# #
# Generic Survey #
# #
# ================= #
[docs]
class Survey( HealpixSurvey, _FootPrintHandler_ ):
# A healpixSurvey based on geometry, so contains a footprint
""" The `Survey` class.
Parameters
----------
footprint: `shapely.geometry`
footprint in the sky of the observing camera
nside : int
healpix nside parameter
data: `pandas.DataFrame`
observing data.
"""
def __init__(self, footprint=None, nside=200, data=None):
""" Initialize the Survey class."""
super().__init__(nside=nside, data=data)
self._footprint = footprint
# ============== #
# I/O #
# ============== #
[docs]
@classmethod
def from_random(cls, *args, **kwargs):
""" Not implemented """
raise NotImplementedError(" not implemented ")
[docs]
@classmethod
def from_data(cls, data, footprint=None, nside=200):
""" Load an instance given survey data and healpix size (nside) .
Parameters
----------
data: `pandas.DataFrame`
observing data.
footprint: `shapely.geometry`
footprint in the sky of the observing camera
nside : int
healpix nside parameter
Returns
-------
instance
See also
--------
from_random: generate random observing data and loads the instance.
"""
return cls(data=data, footprint=footprint, nside=nside)
[docs]
@classmethod
def from_pointings(cls, data, footprint=None,
rakey="ra", deckey="dec",
nside=200,
backend="polars",
use_pyarrow_extension_array=True,
**kwargs):
""" Loads an instance given observing poitings of a survey.
This loads an ``polygon.PolygonSurvey`` using ``from_pointing`` and
converts that into an healpix using the ``to_healpix()`` method.
Parameters
----------
data: `pandas.DataFrame` or dict
observing data, must contain the rakey and deckey columns.
footprint: `shapely.geometry`
footprint in the sky of the observing camera
rakey: str
name of the R.A. column (in deg)
deckey: str
name of the Declination column (in deg)
nside : int
healpix nside parameter
backend: str
which backend to use to merge the data (speed issue):
- `polars` (fastest): requires polars installed -> converted to pandas at the end
- `pandas` (classic): the normal way
- `dask` (lazy): as persisted dask.dataframe is returned
use_pyarrow_extension_array: bool
= ignored in backend != 'polars' or polars_to_pandas is not True =
should the pandas dataframe be based on numpy array (slow to load but faster then)
or based on pyarrow array (like in polars) ; faster but numpy.asarray will be
used by pandas when need (which will then slow things down).
**kwargs goes to ``polygon.PolygonSurvey.from_pointings``
Returns
-------
instance
"""
if footprint is None:
footprint = cls._FOOTPRINT
# super() calls HealpixSurvey.
this = super().from_pointings(nside=nside, data=data, footprint=footprint,
rakey=rakey, deckey=deckey,
backend=backend,
use_pyarrow_extension_array=use_pyarrow_extension_array,
**kwargs)
return cls.from_healpix(healpixsurvey=this, footprint=footprint)
[docs]
@classmethod
def from_healpix(cls, healpixsurvey, footprint):
""" Creates an instance given a heapixsurvey and a footprint.
Parameters
----------
healpixsurvey: `HealpixSurvey`
healpix survey instance
footprint: `shapely.geometry`
footprint in the sky of the observing camera
Returns
-------
Survey
"""
return cls(data=healpixsurvey.data,
footprint=footprint,
nside=healpixsurvey.nside)
# ================= #
# #
# Grid Survey #
# #
# ================= #
[docs]
class GridSurvey(PolygonSurvey, _FootPrintHandler_ ):
"""The `GridSurvey` class.
Parameters
----------
data: `pandas.DataFrame`
observing data.
fields: `geodataframe`
field definitions.
footprint: `shapely.geometry`
footprint in the sky of the observing camera.
"""
def __init__(self, data=None, fields=None, footprint=None, **kwargs):
""" Initialize the GridSurvey class."""
self._footprint = footprint
super().__init__(data=data, fields=fields)
[docs]
@classmethod
def from_pointings(cls, data, fields_or_coords=None, footprint=None, **kwargs):
""" Loads an instance given observing poitings of a survey.
Parameters
----------
data: `pandas.DataFrame` or `dict`
observing data, must contain the rakey and deckey columns.
fields_or_coords: `geodataframe` or dict
field definitions or coordinates.
footprint: `shapely.geometry`
footprint in the sky of the observing camera
**kwargs goes to ``super().__init__``
Returns
-------
GridSurvey
"""
if type(data) is dict:
data = pandas.DataFrame.from_dict(data)
fields = cls._parse_fields(fields_or_coords, footprint)
return cls(data=data, fields=fields, footprint=footprint, **kwargs)
[docs]
@classmethod
def from_logs(cls, **kwargs):
""" Not implemented """
raise NotImplementedError("from_logs is not Implemented for this survey")
# ============== #
# Internal #
# ============== #
[docs]
@classmethod
def _parse_fields(cls, fields_or_coords, footprint=None):
""" Parse the fields from coordinates.
Parameters
----------
fields_or_coords: `geodataframe` or dict
field definitions or coordinates.
footprint: `shapely.geometry`
footprint in the sky of the observing camera
Returns
-------
`geopandas.GeoDataFrame`
"""
if fields_or_coords is None:
if hasattr(cls, "_DEFAULT_FIELDS"):
return cls._DEFAULT_FIELDS
return None
# this is list of coords
if type(fields_or_coords) is dict and "ra" in list(fields_or_coords.values())[0]:
fields_or_coords = pandas.DataFrame(fields_or_coords).T
# this enters the new if.
if type(fields_or_coords) is pandas.DataFrame and "ra" in fields_or_coords:
if footprint is None:
raise ValueError("fields given as list of coordinates but no footprint given?")
from ztffields.projection import project_to_radec
import geopandas
if fields_or_coords.index.name is None:
fields_or_coords.index.name = "fieldid"
# Now expected geopandas
fields = geopandas.GeoDataFrame( geometry=project_to_radec(footprint,
fields_or_coords["ra"],
fields_or_coords["dec"]),
index=fields_or_coords.index)
fields = fields.join(fields_or_coords) # store input data
else:
fields = fields_or_coords
return super()._parse_fields(fields)
# ============== #
# Properties #
# ============== #
@property
def fields(self):
""" Geodataframe containing the fields coordinates. """
if not hasattr(self,"_fields") or self._fields is None:
if self._DEFAULT_FIELDS is None:
return None
self._fields = self._DEFAULT_FIELDS.copy()
return self._fields
