super().__init__(config=config)
if config and config.inputDimensions != self.inputData.dimensions:
self.dimensions.clear()
self.dimensions.update(config.inputDimensions)
self.inputData = cT.Input(
name=self.inputData.name,
doc=self.inputData.doc,
storageClass=self.inputData.storageClass,
dimensions=frozenset(config.inputDimensions),
)
self.outputData = cT.Output(
name=self.outputData.name,
doc=self.outputData.doc,
storageClass=self.outputData.storageClass,
dimensions=frozenset(config.inputDimensions),
)
if config and config.inputStorageClass != self.inputData.storageClass:
self.inputData = cT.Input(
name=self.inputData.name,
doc=self.inputData.doc,
storageClass=config.inputStorageClass,
dimensions=self.inputData.dimensions,
)
self.outputData = cT.Output(
name=self.outputData.name,
doc=self.outputData.doc,
storageClass=config.inputStorageClass,
dimensions=self.outputData.dimensions,
)
class BinImageDataConfig(
pipeBase.PipelineTaskConfig, pipelineConnections=BinImageDataConnections
):
inputDimensions = pexConfig.ListField(
# Sort to ensure default order is consistent between runs
default=sorted(BinImageDataConnections.dimensions),
dtype=str,
doc="Override for the dimensions of the input and output data.",
)
inputStorageClass = pexConfig.Field(
default="ExposureF",
dtype=str,
doc=(
"Override the storageClass of the input and output data. "
"Must be of type `Image`, `MaskedImage`, or `Exposure`, "
"or one of their subtypes."
),
)
binFactor = pexConfig.Field(
dtype=int,
doc="Binning factor applied to both spatial dimensions.",
default=8,
check=lambda x: x > 1,
)
class BinImageDataTask(pipeBase.PipelineTask):
ConfigClass = BinImageDataConfig
_DefaultName = "binImageData"
@timeMethod
def run(self, inputData, binFactor=None):
if not binFactor:
binFactor = self.config.binFactor
return pipeBase.Struct(outputData=binImageData(inputData, binFactor))
def binImageData(inputData, binFactor=8):
Definition at line 136 of file binImageDataTask.py.