Add utility to create empty tables.

python/lsst/ap/association/diaPipe.py

@@ -47,7 +47,8 @@
  LoadDiaCatalogsTask,
  PackageAlertsTask)
 from lsst.ap.association.ssoAssociation import SolarSystemAssociationTask
-from lsst.ap.association.utils import convertTableToSdmSchema, readSchemaFromApdb, dropEmptyColumns
+from lsst.ap.association.utils import convertTableToSdmSchema, readSchemaFromApdb, dropEmptyColumns, \
+ make_empty_catalog
 from lsst.daf.base import DateTime
 from lsst.meas.base import DetectorVisitIdGeneratorConfig, \
  DiaObjectCalculationTask
@@ -658,7 +659,7 @@ def createNewDiaObjects(self, unAssocDiaSources):
  - ``nNewDiaObjects`` : Number of newly created diaObjects.(`int`)
  """
  if len(unAssocDiaSources) == 0:
- newDiaObjects = self.apdb._make_empty_catalog(daxApdb.ApdbTables.DiaObject)
+ newDiaObjects = make_empty_catalog(self.schema, tableName="DiaObject")
  else:
  unAssocDiaSources["diaObjectId"] = unAssocDiaSources["diaSourceId"]
  newDiaObjects = convertTableToSdmSchema(self.schema, unAssocDiaSources,

python/lsst/ap/association/utils.py

@@ -21,7 +21,8 @@
 
 """Utilities for working with the APDB.
 """
-__all__ = ("convertTableToSdmSchema", "readSdmSchemaFile", "readSchemaFromApdb", "dropEmptyColumns")
+__all__ = ("convertTableToSdmSchema", "readSdmSchemaFile", "readSchemaFromApdb",
+ "dropEmptyColumns", "make_empty_catalog")
 
 from collections.abc import Mapping
 import os
@@ -196,3 +197,27 @@ def dropEmptyColumns(apdbSchema, sourceTable, tableName):
  nullColNames = nullColumns[nullColumns].index.tolist()
  dropColumns = list(set(nullColNames) & set(nullableList))
  return sourceTable.drop(columns=dropColumns)
+
+
+def make_empty_catalog(apdbSchema, tableName):
+ """Make an empty catalog for a table with a given name.
+
+ Parameters
+ ----------
+ apdbSchema : `lsst.dax.apdb.apdbSchema.ApdbSchema`
+ Schema from ``sdm_schemas`` containing the table definition to use.
+ tableName : `str`
+ Name of the table in the schema to use.
+
+ Returns
+ -------
+ catalog : `pandas.DataFrame`
+ An empty catalog.
+ """
+ table = apdbSchema[tableName]
+
+ data = {
+ columnDef.name: pd.Series(column_dtype(columnDef.datatype))
+ for columnDef in table.columns
+ }
+ return pd.DataFrame(data)

tests/test_utils.py

@@ -0,0 +1,54 @@
+# This file is part of ap_association.
+#
+# Developed for the LSST Data Management System.
+# This product includes software developed by the LSST Project
+# (https://www.lsst.org).
+# See the COPYRIGHT file at the top-level directory of this distribution
+# for details of code ownership.
+#
+# This program is free software: you can redistribute it and/or modify
+# it under the terms of the GNU General Public License as published by
+# the Free Software Foundation, either version 3 of the License, or
+# (at your option) any later version.
+#
+# This program is distributed in the hope that it will be useful,
+# but WITHOUT ANY WARRANTY; without even the implied warranty of
+# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
+# GNU General Public License for more details.
+#
+# You should have received a copy of the GNU General Public License
+# along with this program. If not, see <https://www.gnu.org/licenses/>.
+
+import tempfile
+import unittest
+
+import pandas as pd
+
+from lsst.ap.association.utils import readSchemaFromApdb, make_empty_catalog, convertTableToSdmSchema
+import lsst.dax.apdb as daxApdb
+
+
+class TestUtils(unittest.TestCase):
+
+ def test_make_empty_catalog(self):
+ """Check that an empty catalog has the correct format.
+ """
+ apdb_config = daxApdb.ApdbSql.init_database(db_url="sqlite://")
+ config_file = tempfile.NamedTemporaryFile()
+ self.addCleanup(config_file.close)
+ apdb_config.save(config_file.name)
+ apdb = daxApdb.Apdb.from_uri(config_file.name)
+ schema = readSchemaFromApdb(apdb)
+
+ tableNames = ["DiaObject", "DiaSource", "DiaForcedSource"]
+ for tableName in tableNames:
+ emptyDiaObjects = make_empty_catalog(schema, tableName=tableName)
+ emptyColumns = set(emptyDiaObjects.columns)
+ self.assertIn("ra", emptyColumns)
+ self.assertIn("dec", emptyColumns)
+ self.assertIn("diaObjectId", emptyColumns)
+
+ emptyDf = pd.DataFrame(columns=["diaObjectId",])
+ emptyDf.set_index("diaObjectId")
+ convertedEmptyDiaObjects = convertTableToSdmSchema(schema, emptyDf, tableName=tableName)
+ self.assertEqual(set(convertedEmptyDiaObjects.columns), emptyColumns)