Inheritance diagram for lsst.dax.apdb.apdbCassandra.ApdbCassandra:

Public Member Functions
def	__init__ (self, ApdbCassandraConfig config)

Optional[TableDef]	tableDef (self, ApdbTables table)

None	makeSchema (self, bool drop=False)

pandas.DataFrame	getDiaObjects (self, sphgeom.Region region)

Optional[pandas.DataFrame]	getDiaSources (self, sphgeom.Region region, Optional[Iterable[int]] object_ids, dafBase.DateTime visit_time)

Optional[pandas.DataFrame]	getDiaForcedSources (self, sphgeom.Region region, Optional[Iterable[int]] object_ids, dafBase.DateTime visit_time)

None	store (self, dafBase.DateTime visit_time, pandas.DataFrame objects, Optional[pandas.DataFrame] sources=None, Optional[pandas.DataFrame] forced_sources=None)

None	dailyJob (self)

int	countUnassociatedObjects (self)

pandas.DataFrame	getDiaObjects (self, Region region)

Optional[pandas.DataFrame]	getDiaSources (self, Region region, Optional[Iterable[int]] object_ids, dafBase.DateTime visit_time)

Optional[pandas.DataFrame]	getDiaForcedSources (self, Region region, Optional[Iterable[int]] object_ids, dafBase.DateTime visit_time)

ConfigurableField	makeField (cls, str doc)

Public Attributes
	config

Static Public Attributes
	partition_zero_epoch = dafBase.DateTime(1970, 1, 1, 0, 0, 0, dafBase.DateTime.TAI)

	ConfigClass = ApdbConfig

Detailed Description

Implementation of APDB database on to of Apache Cassandra.

The implementation is configured via standard ``pex_config`` mechanism
using `ApdbCassandraConfig` configuration class. For an example of
different configurations check config/ folder.

Parameters
----------
config : `ApdbCassandraConfig`
    Configuration object.

Definition at line 365 of file apdbCassandra.py.

Constructor & Destructor Documentation

◆ init()

def lsst.dax.apdb.apdbCassandra.ApdbCassandra.__init__	(		self,
		ApdbCassandraConfig	config
	)

Definition at line 381 of file apdbCassandra.py.

     def __init__(self, config: ApdbCassandraConfig):
  
         if not CASSANDRA_IMPORTED:
             raise CassandraMissingError()
  
         self.config = config
  
         _LOG.debug("ApdbCassandra Configuration:")
         _LOG.debug("    read_consistency: %s", self.config.read_consistency)
         _LOG.debug("    write_consistency: %s", self.config.write_consistency)
         _LOG.debug("    read_sources_months: %s", self.config.read_sources_months)
         _LOG.debug("    read_forced_sources_months: %s", self.config.read_forced_sources_months)
         _LOG.debug("    dia_object_columns: %s", self.config.dia_object_columns)
         _LOG.debug("    schema_file: %s", self.config.schema_file)
         _LOG.debug("    extra_schema_file: %s", self.config.extra_schema_file)
         _LOG.debug("    schema prefix: %s", self.config.prefix)
         _LOG.debug("    part_pixelization: %s", self.config.part_pixelization)
         _LOG.debug("    part_pix_level: %s", self.config.part_pix_level)
         _LOG.debug("    query_per_time_part: %s", self.config.query_per_time_part)
         _LOG.debug("    query_per_spatial_part: %s", self.config.query_per_spatial_part)
  
         self._partitioner = Partitioner(config)
  
         addressTranslator: Optional[AddressTranslator] = None
         if config.private_ips:
             loadBalancePolicy = WhiteListRoundRobinPolicy(hosts=config.contact_points)
             addressTranslator = _AddressTranslator(config.contact_points, config.private_ips)
         else:
             loadBalancePolicy = RoundRobinPolicy()
  
         self._read_consistency = getattr(cassandra.ConsistencyLevel, config.read_consistency)
         self._write_consistency = getattr(cassandra.ConsistencyLevel, config.write_consistency)
  
         self._cluster = Cluster(contact_points=self.config.contact_points,
                                 load_balancing_policy=loadBalancePolicy,
                                 address_translator=addressTranslator,
                                 protocol_version=self.config.protocol_version)
         self._session = self._cluster.connect(keyspace=config.keyspace)
         self._session.row_factory = cassandra.query.named_tuple_factory
  
         self._schema = ApdbCassandraSchema(session=self._session,
                                            schema_file=self.config.schema_file,
                                            extra_schema_file=self.config.extra_schema_file,
                                            prefix=self.config.prefix,
                                            packing=self.config.packing,
                                            time_partition_tables=self.config.time_partition_tables)
         self._partition_zero_epoch_mjd = self.partition_zero_epoch.get(system=dafBase.DateTime.MJD)
  

Member Function Documentation

◆ countUnassociatedObjects()

int lsst.dax.apdb.apdbCassandra.ApdbCassandra.countUnassociatedObjects ( self )

Return the number of DiaObjects that have only one DiaSource
associated with them.

Used as part of ap_verify metrics.

Returns
-------
count : `int`
    Number of DiaObjects with exactly one associated DiaSource.

Notes
-----
This method can be very inefficient or slow in some implementations.

Reimplemented from lsst.dax.apdb.apdb.Apdb.

Definition at line 705 of file apdbCassandra.py.

     def countUnassociatedObjects(self) -> int:
         # docstring is inherited from a base class
         raise NotImplementedError()
  

◆ dailyJob()

None lsst.dax.apdb.apdbCassandra.ApdbCassandra.dailyJob ( self )

Implement daily activities like cleanup/vacuum.

What should be done during daily activities is determined by
specific implementation.

Reimplemented from lsst.dax.apdb.apdb.Apdb.

Definition at line 701 of file apdbCassandra.py.

     def dailyJob(self) -> None:
         # docstring is inherited from a base class
         pass
  

◆ getDiaForcedSources() [1/2]

Optional[pandas.DataFrame] lsst.dax.apdb.apdb.Apdb.getDiaForcedSources	(		self,
		Region	region,
		Optional[Iterable[int]]	object_ids,
		dafBase.DateTime	visit_time
	)

inherited

Return catalog of DiaForcedSource instances from a given region.

Parameters
----------
region : `lsst.sphgeom.Region`
    Region to search for DIASources.
object_ids : iterable [ `int` ], optional
    List of DiaObject IDs to further constrain the set of returned
    sources. If list is empty then empty catalog is returned with a
    correct schema. If `None` then returned sources are not
    constrained. Some implementations may not support latter case.
visit_time : `lsst.daf.base.DateTime`
    Time of the current visit.

Returns
-------
catalog : `pandas.DataFrame`, or `None`
    Catalog containing DiaSource records. `None` is returned if
    ``read_forced_sources_months`` configuration parameter is set to 0.

Raises
------
NotImplementedError
    May be raised by some implementations if ``object_ids`` is `None`.

Notes
-----
This method returns DiaForcedSource catalog for a region with additional
filtering based on DiaObject IDs. Only a subset of DiaSource history
is returned limited by ``read_forced_sources_months`` config parameter,
w.r.t. ``visit_time``. If ``object_ids`` is empty then an empty catalog
is always returned with the correct schema (columns/types). If
``object_ids`` is `None` then no filtering is performed and some of the
returned records may be outside the specified region.

Reimplemented in lsst.dax.apdb.apdbSql.ApdbSql.

Definition at line 163 of file apdb.py.

                             visit_time: dafBase.DateTime) -> Optional[pandas.DataFrame]:
         """Return catalog of DiaForcedSource instances from a given region.
  
         Parameters
         ----------
         region : `lsst.sphgeom.Region`
             Region to search for DIASources.
         object_ids : iterable [ `int` ], optional
             List of DiaObject IDs to further constrain the set of returned
             sources. If list is empty then empty catalog is returned with a
             correct schema. If `None` then returned sources are not
             constrained. Some implementations may not support latter case.
         visit_time : `lsst.daf.base.DateTime`
             Time of the current visit.
  
         Returns
         -------
         catalog : `pandas.DataFrame`, or `None`
             Catalog containing DiaSource records. `None` is returned if
             ``read_forced_sources_months`` configuration parameter is set to 0.
  
         Raises
         ------
         NotImplementedError
             May be raised by some implementations if ``object_ids`` is `None`.
  
         Notes
         -----
         This method returns DiaForcedSource catalog for a region with additional
         filtering based on DiaObject IDs. Only a subset of DiaSource history
         is returned limited by ``read_forced_sources_months`` config parameter,
         w.r.t. ``visit_time``. If ``object_ids`` is empty then an empty catalog
         is always returned with the correct schema (columns/types). If
         ``object_ids`` is `None` then no filtering is performed and some of the
         returned records may be outside the specified region.
         """
         raise NotImplementedError()
  

◆ getDiaForcedSources() [2/2]

Optional[pandas.DataFrame] lsst.dax.apdb.apdbCassandra.ApdbCassandra.getDiaForcedSources	(		self,
		sphgeom.Region	region,
		Optional[Iterable[int]]	object_ids,
		dafBase.DateTime	visit_time
	)

Definition at line 486 of file apdbCassandra.py.

                             visit_time: dafBase.DateTime) -> Optional[pandas.DataFrame]:
         return self._getSources(region, object_ids, visit_time, ApdbTables.DiaForcedSource,
                                 self.config.read_forced_sources_months)
  

◆ getDiaObjects() [1/2]

pandas.DataFrame lsst.dax.apdb.apdb.Apdb.getDiaObjects	(		self,
		Region	region
	)

inherited

Returns catalog of DiaObject instances from a given region.

This method returns only the last version of each DiaObject. Some
records in a returned catalog may be outside the specified region, it
is up to a client to ignore those records or cleanup the catalog before
futher use.

Parameters
----------
region : `lsst.sphgeom.Region`
    Region to search for DIAObjects.

Returns
-------
catalog : `pandas.DataFrame`
    Catalog containing DiaObject records for a region that may be a
    superset of the specified region.

Reimplemented in lsst.dax.apdb.apdbSql.ApdbSql.

Definition at line 105 of file apdb.py.

     def getDiaObjects(self, region: Region) -> pandas.DataFrame:
         """Returns catalog of DiaObject instances from a given region.
  
         This method returns only the last version of each DiaObject. Some
         records in a returned catalog may be outside the specified region, it
         is up to a client to ignore those records or cleanup the catalog before
         futher use.
  
         Parameters
         ----------
         region : `lsst.sphgeom.Region`
             Region to search for DIAObjects.
  
         Returns
         -------
         catalog : `pandas.DataFrame`
             Catalog containing DiaObject records for a region that may be a
             superset of the specified region.
         """
         raise NotImplementedError()
  

◆ getDiaObjects() [2/2]

pandas.DataFrame lsst.dax.apdb.apdbCassandra.ApdbCassandra.getDiaObjects	(		self,
		sphgeom.Region	region
	)

Definition at line 447 of file apdbCassandra.py.

     def getDiaObjects(self, region: sphgeom.Region) -> pandas.DataFrame:
         # docstring is inherited from a base class
         packedColumns = self._schema.packedColumns(ApdbTables.DiaObjectLast)
         self._session.row_factory = _PandasRowFactory(packedColumns)
         self._session.default_fetch_size = None
  
         pixels = self._partitioner.pixels(region)
         _LOG.debug("getDiaObjects: #partitions: %s", len(pixels))
         pixels_str = ",".join([str(pix) for pix in pixels])
  
         queries: List[Tuple] = []
         query = f'SELECT * from "DiaObjectLast" WHERE "apdb_part" IN ({pixels_str})'
         queries += [(cassandra.query.SimpleStatement(query, consistency_level=self._read_consistency), {})]
         _LOG.debug("getDiaObjects: #queries: %s", len(queries))
         # _LOG.debug("getDiaObjects: queries: %s", queries)
  
         objects = None
         with Timer('DiaObject select', self.config.timer):
             # submit all queries
             futures = [self._session.execute_async(query, values, timeout=self.config.read_timeout)
                        for query, values in queries]
             # TODO: This orders result processing which is not very efficient
             dataframes = [future.result()._current_rows for future in futures]
             # concatenate all frames
             if len(dataframes) == 1:
                 objects = dataframes[0]
             else:
                 objects = pandas.concat(dataframes)
  
         _LOG.debug("found %s DiaObjects", objects.shape[0])
         return objects
  

◆ getDiaSources() [1/2]

Optional[pandas.DataFrame] lsst.dax.apdb.apdb.Apdb.getDiaSources	(		self,
		Region	region,
		Optional[Iterable[int]]	object_ids,
		dafBase.DateTime	visit_time
	)

inherited

Return catalog of DiaSource instances from a given region.

Parameters
----------
region : `lsst.sphgeom.Region`
    Region to search for DIASources.
object_ids : iterable [ `int` ], optional
    List of DiaObject IDs to further constrain the set of returned
    sources. If `None` then returned sources are not constrained. If
    list is empty then empty catalog is returned with a correct
    schema.
visit_time : `lsst.daf.base.DateTime`
    Time of the current visit.

Returns
-------
catalog : `pandas.DataFrame`, or `None`
    Catalog containing DiaSource records. `None` is returned if
    ``read_sources_months`` configuration parameter is set to 0.

Notes
-----
This method returns DiaSource catalog for a region with additional
filtering based on DiaObject IDs. Only a subset of DiaSource history
is returned limited by ``read_sources_months`` config parameter, w.r.t.
``visit_time``. If ``object_ids`` is empty then an empty catalog is
always returned with the correct schema (columns/types). If
``object_ids`` is `None` then no filtering is performed and some of the
returned records may be outside the specified region.

Reimplemented in lsst.dax.apdb.apdbSql.ApdbSql.

Definition at line 127 of file apdb.py.

                       visit_time: dafBase.DateTime) -> Optional[pandas.DataFrame]:
         """Return catalog of DiaSource instances from a given region.
  
         Parameters
         ----------
         region : `lsst.sphgeom.Region`
             Region to search for DIASources.
         object_ids : iterable [ `int` ], optional
             List of DiaObject IDs to further constrain the set of returned
             sources. If `None` then returned sources are not constrained. If
             list is empty then empty catalog is returned with a correct
             schema.
         visit_time : `lsst.daf.base.DateTime`
             Time of the current visit.
  
         Returns
         -------
         catalog : `pandas.DataFrame`, or `None`
             Catalog containing DiaSource records. `None` is returned if
             ``read_sources_months`` configuration parameter is set to 0.
  
         Notes
         -----
         This method returns DiaSource catalog for a region with additional
         filtering based on DiaObject IDs. Only a subset of DiaSource history
         is returned limited by ``read_sources_months`` config parameter, w.r.t.
         ``visit_time``. If ``object_ids`` is empty then an empty catalog is
         always returned with the correct schema (columns/types). If
         ``object_ids`` is `None` then no filtering is performed and some of the
         returned records may be outside the specified region.
         """
         raise NotImplementedError()
  

◆ getDiaSources() [2/2]

Optional[pandas.DataFrame] lsst.dax.apdb.apdbCassandra.ApdbCassandra.getDiaSources	(		self,
		sphgeom.Region	region,
		Optional[Iterable[int]]	object_ids,
		dafBase.DateTime	visit_time
	)

Definition at line 479 of file apdbCassandra.py.

                       visit_time: dafBase.DateTime) -> Optional[pandas.DataFrame]:
         # docstring is inherited from a base class
         return self._getSources(region, object_ids, visit_time, ApdbTables.DiaSource,
                                 self.config.read_sources_months)
  

◆ makeField()

ConfigurableField lsst.dax.apdb.apdb.Apdb.makeField	(		cls,
		str	doc
	)

inherited

Make a `~lsst.pex.config.ConfigurableField` for Apdb.

Parameters
----------
doc : `str`
    Help text for the field.

Returns
-------
configurableField : `lsst.pex.config.ConfigurableField`
    A `~lsst.pex.config.ConfigurableField` for Apdb.

Definition at line 268 of file apdb.py.

     def makeField(cls, doc: str) -> ConfigurableField:
         """Make a `~lsst.pex.config.ConfigurableField` for Apdb.
  
         Parameters
         ----------
         doc : `str`
             Help text for the field.
  
         Returns
         -------
         configurableField : `lsst.pex.config.ConfigurableField`
             A `~lsst.pex.config.ConfigurableField` for Apdb.
         """
         return ConfigurableField(doc=doc, target=cls)

◆ makeSchema()

None lsst.dax.apdb.apdbCassandra.ApdbCassandra.makeSchema	(		self,
		bool	drop = `False`
	)

Create or re-create whole database schema.

Parameters
----------
drop : `bool`
    If True then drop all tables before creating new ones.

Reimplemented from lsst.dax.apdb.apdb.Apdb.

Definition at line 433 of file apdbCassandra.py.

     def makeSchema(self, drop: bool = False) -> None:
         # docstring is inherited from a base class
  
         if self.config.time_partition_tables:
             time_partition_start = dafBase.DateTime(self.config.time_partition_start, dafBase.DateTime.TAI)
             time_partition_end = dafBase.DateTime(self.config.time_partition_end, dafBase.DateTime.TAI)
             part_range = (
                 self._time_partition(time_partition_start),
                 self._time_partition(time_partition_end) + 1
             )
             self._schema.makeSchema(drop=drop, part_range=part_range)
         else:
             self._schema.makeSchema(drop=drop)
  

◆ store()

None lsst.dax.apdb.apdbCassandra.ApdbCassandra.store	(		self,
		dafBase.DateTime	visit_time,
		pandas.DataFrame	objects,
		Optional[pandas.DataFrame]	sources = `None`,
		Optional[pandas.DataFrame]	forced_sources = `None`
	)

Store all three types of catalogs in the database.

Parameters
----------
visit_time : `lsst.daf.base.DateTime`
    Time of the visit.
objects : `pandas.DataFrame`
    Catalog with DiaObject records.
sources : `pandas.DataFrame`, optional
    Catalog with DiaSource records.
forced_sources : `pandas.DataFrame`, optional
    Catalog with DiaForcedSource records.

Notes
-----
This methods takes DataFrame catalogs, their schema must be
compatible with the schema of APDB table:

  - column names must correspond to database table columns
  - types and units of the columns must match database definitions,
    no unit conversion is performed presently
  - columns that have default values in database schema can be
    omitted from catalog
  - this method knows how to fill interval-related columns of DiaObject
    (validityStart, validityEnd) they do not need to appear in a
    catalog
  - source catalogs have ``diaObjectId`` column associating sources
    with objects

Reimplemented from lsst.dax.apdb.apdb.Apdb.

Definition at line 638 of file apdbCassandra.py.

               forced_sources: Optional[pandas.DataFrame] = None) -> None:
         # docstring is inherited from a base class
  
         # fill region partition column for DiaObjects
         objects = self._add_obj_part(objects)
         self._storeDiaObjects(objects, visit_time)
  
         if sources is not None:
             # copy apdb_part column from DiaObjects to DiaSources
             sources = self._add_src_part(sources, objects)
             self._storeDiaSources(ApdbTables.DiaSource, sources, visit_time)
  
         if forced_sources is not None:
             forced_sources = self._add_fsrc_part(forced_sources, objects)
             self._storeDiaSources(ApdbTables.DiaForcedSource, forced_sources, visit_time)
  

◆ tableDef()

Optional[TableDef] lsst.dax.apdb.apdbCassandra.ApdbCassandra.tableDef	(		self,
		ApdbTables	table
	)

Return table schema definition for a given table.

Parameters
----------
table : `ApdbTables`
    One of the known APDB tables.

Returns
-------
tableSchema : `TableDef` or `None`
    Table schema description, `None` is returned if table is not
    defined by this implementation.

Reimplemented from lsst.dax.apdb.apdb.Apdb.

Definition at line 429 of file apdbCassandra.py.

     def tableDef(self, table: ApdbTables) -> Optional[TableDef]:
         # docstring is inherited from a base class
         return self._schema.tableSchemas.get(table)
  

Member Data Documentation

◆ config

lsst.dax.apdb.apdbCassandra.ApdbCassandra.config

Definition at line 386 of file apdbCassandra.py.

◆ ConfigClass

lsst.dax.apdb.apdb.Apdb.ConfigClass = ApdbConfig

staticinherited

Definition at line 74 of file apdb.py.

◆ partition_zero_epoch

lsst.dax.apdb.apdbCassandra.ApdbCassandra.partition_zero_epoch = dafBase.DateTime(1970, 1, 1, 0, 0, 0, dafBase.DateTime.TAI)

static

Definition at line 378 of file apdbCassandra.py.

The documentation for this class was generated from the following file:

/j/snowflake/release/lsstsw/stack/lsst-scipipe-0.7.0/Linux64/dax_apdb/22.0.1-5-g75bb458+99c117b92f/python/lsst/dax/apdb/apdbCassandra.py

Public Member Functions

Public Attributes

Static Public Attributes

Detailed Description

Constructor & Destructor Documentation

◆ __init__()

Member Function Documentation

◆ countUnassociatedObjects()

◆ dailyJob()

◆ getDiaForcedSources() [1/2]

◆ getDiaForcedSources() [2/2]

◆ getDiaObjects() [1/2]

◆ getDiaObjects() [2/2]

◆ getDiaSources() [1/2]

◆ getDiaSources() [2/2]

◆ makeField()

◆ makeSchema()

◆ store()

◆ tableDef()

Member Data Documentation

◆ config

◆ ConfigClass

◆ partition_zero_epoch

◆ init()