Skip to content

Classifier

Bases: BaseComponent

Classifies objects in polygon-tiled imagery.

Requirements
  • tiles_path: Directory containing polygon tiles
  • infer_coco_path: COCO annotations with instance masks
Produces
  • Updated infer_gdf with classification results
  • Columns: classifier_score, classifier_class, classifier_scores
Source code in canopyrs/engine/components/classifier.py 
 24
 25
 26
 27
 28
 29
 30
 31
 32
 33
 34
 35
 36
 37
 38
 39
 40
 41
 42
 43
 44
 45
 46
 47
 48
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
class ClassifierComponent(BaseComponent):
    """
    Classifies objects in polygon-tiled imagery.

    Requirements:
        - tiles_path: Directory containing polygon tiles
        - infer_coco_path: COCO annotations with instance masks

    Produces:
        - Updated infer_gdf with classification results
        - Columns: classifier_score, classifier_class, classifier_scores
    """

    name = 'classifier'

    BASE_REQUIRES_STATE = {StateKey.TILES_PATH, StateKey.INFER_COCO_PATH}
    BASE_REQUIRES_COLUMNS: Set[str] = set()

    BASE_PRODUCES_STATE = {StateKey.INFER_GDF, StateKey.INFER_COCO_PATH}
    BASE_PRODUCES_COLUMNS = {Col.CLASSIFIER_SCORE, Col.CLASSIFIER_CLASS, Col.CLASSIFIER_SCORES}

    BASE_STATE_HINTS = {
        StateKey.TILES_PATH: "Classifier needs polygon tiles. Add a tilerizer with tile_type='polygon'.",
        StateKey.INFER_COCO_PATH: "Classifier needs COCO annotations from a polygon tilerizer.",
    }

    BASE_COLUMN_HINTS = {
        Col.OBJECT_ID: "Classifier needs object IDs to merge results back to infer_gdf.",
    }

    def __init__(
        self,
        config: ClassifierConfig,
        parent_output_path: str = None,
        component_id: int = None
    ):
        super().__init__(config, parent_output_path, component_id)

        # Store model class (instantiate in __call__ to avoid loading during validation)
        if config.model not in CLASSIFIER_REGISTRY:
            raise ValueError(f'Invalid classifier model: {config.model}')
        self._model_class = CLASSIFIER_REGISTRY.get(config.model)

        # Set requirements
        self.requires_state = set(self.BASE_REQUIRES_STATE)
        self.requires_columns = set(self.BASE_REQUIRES_COLUMNS)
        self.produces_state = set(self.BASE_PRODUCES_STATE)
        self.produces_columns = set(self.BASE_PRODUCES_COLUMNS)

        # Set hints
        self.state_hints = dict(self.BASE_STATE_HINTS)
        self.column_hints = dict(self.BASE_COLUMN_HINTS)

    @classmethod
    def run_standalone(
        cls,
        config: ClassifierConfig,
        tiles_path: str,
        infer_coco_path: str,
        output_path: str,
    ) -> 'DataState':
        """
        Run classifier standalone on polygon-tiled imagery.

        Args:
            config: Classifier configuration
            tiles_path: Path to directory containing polygon tiles
            infer_coco_path: Path to COCO annotations with instance masks
            output_path: Where to save outputs

        Returns:
            DataState with classification results (access .infer_gdf for the GeoDataFrame)

        Example:
            result = ClassifierComponent.run_standalone(
                config=ClassifierConfig(model='resnet50', ...),
                tiles_path='./polygon_tiles',
                infer_coco_path='./coco.json',
                output_path='./output',
            )
            print(result.infer_gdf)
        """
        from canopyrs.engine.pipeline import run_component
        return run_component(
            component=cls(config),
            output_path=output_path,
            tiles_path=tiles_path,
            infer_coco_path=infer_coco_path,
        )

    @validate_requirements
    def __call__(self, data_state: DataState) -> ComponentResult:
        """
        Run classification on polygon tiles.

        Returns flattened DataFrame (no geometry) with classification results.
        Pipeline handles merging into existing GDF.
        """

        classifier = self._model_class(self.config)

        # Create dataset
        infer_ds = InstanceSegmentationLabeledRasterCocoDataset(
            root_path=[data_state.tiles_path, Path(data_state.infer_coco_path).parent],
            transform=None,
            fold=INFER_AOI_NAME,
            other_attributes_names_to_pass=[Col.OBJECT_ID]
        )

        # Run inference
        tiles_paths, class_scores, class_predictions, object_ids = classifier.infer(
            infer_ds, collate_fn_infer_image_masks
        )

        # Flatten outputs into DataFrame (no geometry - will merge into existing)
        df = pd.DataFrame({
            Col.OBJECT_ID: object_ids,
            Col.TILE_PATH: tiles_paths,  # Include for fallback merge key
            Col.CLASSIFIER_CLASS: class_predictions,
            Col.CLASSIFIER_SCORE: [
                scores[pred_idx] for scores, pred_idx in zip(class_scores, class_predictions)
            ],
            Col.CLASSIFIER_SCORES: class_scores,
        })

        # Component-specific validation: warn about unclassified items
        unclassified = df[Col.CLASSIFIER_CLASS].isnull().sum()
        if unclassified > 0:
            warnings.warn(f"{unclassified} items could not be classified.")

        print(f"ClassifierComponent: Classified {len(df) - unclassified}/{len(df)} items.")

        return ComponentResult(
            gdf=df,  # DataFrame, not GeoDataFrame - no geometry
            produced_columns={Col.CLASSIFIER_SCORE, Col.CLASSIFIER_CLASS, Col.CLASSIFIER_SCORES},
            objects_are_new=False,
            save_gpkg=True,
            gpkg_name_suffix="notaggregated",  # classifier saves final results
            save_coco=True,
            coco_scores_column=Col.CLASSIFIER_SCORE,
            coco_categories_column=Col.CLASSIFIER_CLASS,
        )

__call__(data_state)

Run classification on polygon tiles.

Returns flattened DataFrame (no geometry) with classification results. Pipeline handles merging into existing GDF.

Source code in canopyrs/engine/components/classifier.py 
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
@validate_requirements
def __call__(self, data_state: DataState) -> ComponentResult:
    """
    Run classification on polygon tiles.

    Returns flattened DataFrame (no geometry) with classification results.
    Pipeline handles merging into existing GDF.
    """

    classifier = self._model_class(self.config)

    # Create dataset
    infer_ds = InstanceSegmentationLabeledRasterCocoDataset(
        root_path=[data_state.tiles_path, Path(data_state.infer_coco_path).parent],
        transform=None,
        fold=INFER_AOI_NAME,
        other_attributes_names_to_pass=[Col.OBJECT_ID]
    )

    # Run inference
    tiles_paths, class_scores, class_predictions, object_ids = classifier.infer(
        infer_ds, collate_fn_infer_image_masks
    )

    # Flatten outputs into DataFrame (no geometry - will merge into existing)
    df = pd.DataFrame({
        Col.OBJECT_ID: object_ids,
        Col.TILE_PATH: tiles_paths,  # Include for fallback merge key
        Col.CLASSIFIER_CLASS: class_predictions,
        Col.CLASSIFIER_SCORE: [
            scores[pred_idx] for scores, pred_idx in zip(class_scores, class_predictions)
        ],
        Col.CLASSIFIER_SCORES: class_scores,
    })

    # Component-specific validation: warn about unclassified items
    unclassified = df[Col.CLASSIFIER_CLASS].isnull().sum()
    if unclassified > 0:
        warnings.warn(f"{unclassified} items could not be classified.")

    print(f"ClassifierComponent: Classified {len(df) - unclassified}/{len(df)} items.")

    return ComponentResult(
        gdf=df,  # DataFrame, not GeoDataFrame - no geometry
        produced_columns={Col.CLASSIFIER_SCORE, Col.CLASSIFIER_CLASS, Col.CLASSIFIER_SCORES},
        objects_are_new=False,
        save_gpkg=True,
        gpkg_name_suffix="notaggregated",  # classifier saves final results
        save_coco=True,
        coco_scores_column=Col.CLASSIFIER_SCORE,
        coco_categories_column=Col.CLASSIFIER_CLASS,
    )

run_standalone(config, tiles_path, infer_coco_path, output_path) classmethod

Run classifier standalone on polygon-tiled imagery.

Parameters:

Name Type Description Default
config ClassifierConfig

Classifier configuration

 required
tiles_path str

Path to directory containing polygon tiles

 required
infer_coco_path str

Path to COCO annotations with instance masks

 required
output_path str

Where to save outputs

 required

Returns:

Type Description
DataState

DataState with classification results (access .infer_gdf for the GeoDataFrame)
Example

result = ClassifierComponent.run_standalone( config=ClassifierConfig(model='resnet50', ...), tiles_path='./polygon_tiles', infer_coco_path='./coco.json', output_path='./output', ) print(result.infer_gdf)

Source code in canopyrs/engine/components/classifier.py 
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
@classmethod
def run_standalone(
    cls,
    config: ClassifierConfig,
    tiles_path: str,
    infer_coco_path: str,
    output_path: str,
) -> 'DataState':
    """
    Run classifier standalone on polygon-tiled imagery.

    Args:
        config: Classifier configuration
        tiles_path: Path to directory containing polygon tiles
        infer_coco_path: Path to COCO annotations with instance masks
        output_path: Where to save outputs

    Returns:
        DataState with classification results (access .infer_gdf for the GeoDataFrame)

    Example:
        result = ClassifierComponent.run_standalone(
            config=ClassifierConfig(model='resnet50', ...),
            tiles_path='./polygon_tiles',
            infer_coco_path='./coco.json',
            output_path='./output',
        )
        print(result.infer_gdf)
    """
    from canopyrs.engine.pipeline import run_component
    return run_component(
        component=cls(config),
        output_path=output_path,
        tiles_path=tiles_path,
        infer_coco_path=infer_coco_path,
    )