Attach a sample plugin to the FlagHandler test.

tests/testFlagHandler.py

@@ -20,7 +20,7 @@
 # the GNU General Public License along with this program.  If not,
 # see <http://www.lsstcorp.org/LegalNotices/>.
 #
-
+import os
 import unittest
 
 import lsst.utils.tests
@@ -31,8 +31,95 @@
 from lsst.meas.base import FlagDefinition, FlagDefinitionVector, FlagHandler
 from lsst.meas.base.tests import (AlgorithmTestCase)
 
+import lsst.pex.exceptions
+from lsst.meas.base.pluginRegistry import register
+from lsst.meas.base.sfm import SingleFramePluginConfig, SingleFramePlugin
+from lsst.meas.base.baseLib import MeasurementError
+from lsst.meas.base import FlagDefinition, FlagDefinitionVector, FlagHandler
+
+
+class PythonPluginConfig(SingleFramePluginConfig):
+
+    failureType = lsst.pex.config.Field(dtype=int, default=None, optional=False,
+                                  doc="A failure mode to test")
+
+@register("test_PythonPlugin")
+class PythonPlugin(SingleFramePlugin):
+    '''
+    This is a sample Python plugin.  The flag handler for this plugin is created
+    during construction, and is called using the method fail().  All plugins are
+    required to implement this method, which is used to set the flags in the
+    output source record if an error occurs.
+    '''
+    ConfigClass = PythonPluginConfig
+
+    #   Constants to identify the failures.  Must match flagDefs below.
+    #   These are class statics which can be used during run() to identify
+    #   known error conditions
+    FAILURE = 0
+    ERROR1 = 1
+    ERROR2 = 2
+
+    FLAGDEFS = [ FlagDefinition("flag", "General failure error"),
+        FlagDefinition("flag_error1", "First type of failure occured."),
+        FlagDefinition("flag_error2", "Second type of failure occured."),
+    ]
+
+    @classmethod
+    def getExecutionOrder(cls):
+        return cls.SHAPE_ORDER
+
+    def __init__(self, config, name, schema, metadata):
+        SingleFramePlugin.__init__(self, config, name, schema, metadata)
+        self.flagHandler = FlagHandler.addFields(schema, name, FlagDefinitionVector(self.FLAGDEFS))
+
+    #   This is a measure routine which does nothing except to raise Exceptions
+    #   as requested by the caller. Errors normally don't occur unless there is
+    #   something wrong in the inputs, or if there is an error during the measurement
+    def measure(self, measRecord, exposure):
+        if not self.config.failureType is None:
+            if self.config.failureType == PythonPlugin.ERROR1:
+                raise MeasurementError(self.flagHandler.getDefinition(PythonPlugin.ERROR1).doc,
+                    PythonPlugin.ERROR1)
+            if self.config.failureType == PythonPlugin.ERROR2:
+                raise MeasurementError(self.flagHandler.getDefinition(PythonPlugin.ERROR2).doc,
+                    PythonPlugin.ERROR2)
+            raise RuntimeError("An unexpected error occurred")
+
+    #   This routine responds to the standard failure call in baseMeasurement
+    #   If the exception is a MeasurementError, the error will be passed to the
+    #   fail method by the MeasurementFramework.
+    def fail(self, measRecord, error=None):
+        if error is None:
+            self.flagHandler.handleFailure(measRecord)
+        else:
+            self.flagHandler.handleFailure(measRecord, error.cpp)
+
 class FlagHandlerTestCase(AlgorithmTestCase):
-
+
+    #   Setup a configuration and datasource to be used by the plugin tests
+    def setUp(self):
+        self.algName = "test_PythonPlugin"
+        bbox = lsst.afw.geom.Box2I(lsst.afw.geom.Point2I(0,0), lsst.afw.geom.Point2I(100, 100))
+        self.dataset = lsst.meas.base.tests.TestDataset(bbox)
+        self.dataset.addSource(flux=1E5, centroid=lsst.afw.geom.Point2D(25, 26))
+        config = lsst.meas.base.SingleFrameMeasurementConfig() 
+        config.plugins = [self.algName]
+        config.slots.centroid = None
+        config.slots.apFlux = None
+        config.slots.calibFlux = None
+        config.slots.instFlux = None
+        config.slots.modelFlux = None
+        config.slots.psfFlux = None
+        config.slots.shape = None
+        self.config = config
+
+    def tearDown(self):
+        del self.config
+        del self.dataset
+
+    #   Standalone test to create a flaghandler and call it
+    #   This is not a real world example, just a simple unit test
     def testFlagHandler(self):
         control = lsst.meas.base.GaussianCentroidControl()
         alg = lsst.meas.base.GaussianCentroidAlgorithm
@@ -87,6 +174,50 @@ def testFlagHandler(self):
         self.assertFalse(fh.getValue(record, FIRST))
         self.assertTrue(fh.getValue(record, SECOND))
 
+    def testNoError(self):
+        schema = self.dataset.makeMinimalSchema()
+        task = lsst.meas.base.SingleFrameMeasurementTask(schema=schema, config=self.config)
+        exposure, cat = self.dataset.realize(noise=100.0, schema=schema) 
+        task.run(cat, exposure)
+        source = cat[0]
+        self.assertEqual(source.get(self.algName + "_flag"), False)
+        self.assertEqual(source.get(self.algName + "_flag_error1"), False)
+        self.assertEqual(source.get(self.algName + "_flag_error2"), False)
+
+    def testUnexpectedError(self):
+        self.config.plugins[self.algName].failureType = -1     # any unknown error type will do
+        schema = self.dataset.makeMinimalSchema()
+        task = lsst.meas.base.SingleFrameMeasurementTask(schema=schema, config=self.config)
+        exposure, cat = self.dataset.realize(noise=100.0, schema=schema) 
+        task.log.setThreshold(task.log.FATAL)
+        task.run(cat, exposure)
+        source = cat[0]
+        self.assertEqual(source.get(self.algName + "_flag"), True)
+        self.assertEqual(source.get(self.algName + "_flag_error1"), False)
+        self.assertEqual(source.get(self.algName + "_flag_error2"), False)
+
+    def testError1(self):
+        self.config.plugins[self.algName].failureType = PythonPlugin.ERROR1 
+        schema = self.dataset.makeMinimalSchema()
+        task = lsst.meas.base.SingleFrameMeasurementTask(schema=schema, config=self.config)
+        exposure, cat = self.dataset.realize(noise=100.0, schema=schema) 
+        task.run(cat, exposure)
+        source = cat[0]
+        self.assertEqual(source.get(self.algName + "_flag"), True)
+        self.assertEqual(source.get(self.algName + "_flag_error1"), True)
+        self.assertEqual(source.get(self.algName + "_flag_error2"), False)
+
+    def testError2(self):
+        self.config.plugins[self.algName].failureType = PythonPlugin.ERROR2
+        schema = self.dataset.makeMinimalSchema()
+        task = lsst.meas.base.SingleFrameMeasurementTask(schema=schema, config=self.config)
+        exposure, cat = self.dataset.realize(noise=0.0, schema=schema)
+        task.run(cat, exposure)
+        source = cat[0]
+        self.assertEqual(source.get(self.algName + "_flag"), True)
+        self.assertEqual(source.get(self.algName + "_flag_error1"), False)
+        self.assertEqual(source.get(self.algName + "_flag_error2"), True)
+
 def suite():
     """Returns a suite containing all the test cases in this module."""
 
@@ -103,3 +234,5 @@ def run(shouldExit=False):
 
 if __name__ == "__main__":
     run(True)
+
+