Skip to content

semanticsegmentationresult

SemanticSegmentationResult

Bases: ProcessedResult

Source code in src/tcd_pipeline/result/semanticsegmentationresult.py 
 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
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317
318
319
320
321
322
323
324
325
326
327
328
329
330
331
332
333
334
335
336
337
338
339
340
341
342
343
344
345
346
347
348
349
350
351
352
353
354
355
356
357
358
359
360
361
362
363
364
365
366
367
368
369
370
371
372
373
374
375
376
377
378
379
380
381
382
383
384
385
386
387
388
389
390
391
392
393
394
395
class SemanticSegmentationResult(ProcessedResult):
    def __init__(
        self,
        image: rasterio.DatasetReader,
        tiled_masks: Optional[list] = [],
        bboxes: list[box] = [],
        confidence_threshold: float = 0.2,
        merge_pad: int = 32,
        config: dict = None,
    ) -> None:
        self.image = image
        self.masks = tiled_masks
        self.bboxes = bboxes
        self.merge_pad = merge_pad
        self.valid_region = None
        self.valid_mask = None
        self.prediction_time_s = -1
        self.config = config

        self.valid_window = rasterio.windows.from_bounds(
            *self.image.bounds, transform=self.image.transform
        )

        self.set_threshold(confidence_threshold)

    def serialise(
        self,
        output_folder: str,
        overwrite: bool = True,
        file_prefix: Optional[str] = "results",
    ) -> None:
        """Serialise raw prediction masks. Masks are stored as NPZ files with the
        keys "mask" and "bbox" as well as a timestamp which can be used as a sanity
        check when loading. A JSON file containing a list of masks will also be created.

        Args:
            output_folder (str): output folder
            overwrite (bool, optional): overwrite existing data, defaults True
            file_prefix (str, optional): file name, defaults to results
        """

        logger.info(f"Serialising results to {output_folder}/{file_prefix}")
        os.makedirs(output_folder, exist_ok=True)

        meta_path = os.path.join(output_folder, f"{file_prefix}.json")

        if os.path.exists(meta_path) and not overwrite:
            logger.error(
                f"Output metadata already exists {meta_path}, will not overwrite."
            )
            return

        timestamp = time.time()
        metadata = {}
        metadata["image"] = self.image.name
        metadata["timestamp"] = int(timestamp)
        metadata["masks"] = []
        metadata["confidence_threshold"] = self.confidence_threshold
        metadata["prediction_time_s"] = self.prediction_time_s
        # metadata["config"] = dict(self.config)
        metadata["hardware"] = self.get_hardware_information()

        for i, item in enumerate(zip(self.masks, self.bboxes)):
            mask, bbox = item
            file_name = f"{file_prefix}_{i}.npz"
            output_path = os.path.join(output_folder, file_name)

            if os.path.exists(output_path) and not overwrite:
                logger.error(
                    f"Output file already exists {output_path}, will not overwrite."
                )
                return
            np.savez_compressed(
                file=output_path,
                mask=mask[0][0],
                image_bbox=mask[0][1],
                bbox=np.array(bbox),
                timestamp=int(timestamp),
            )
            metadata["masks"].append(os.path.abspath(output_path))

        with open(meta_path, "w") as fp:
            json.dump(metadata, fp, indent=1)

    @classmethod
    def load_serialisation(cls, input_file: str, image_path: Optional[str] = None):
        """Loads a ProcessedResult based on a json serialization file.

        Args:
            input_file (str): serialised instance metadata JSON file
            image_path (Optional[str]): image path, optional
        Returns:
            SegmentationResult: SegmentationResult described by the file
        """
        tiled_masks = []
        bboxes = []

        with open(input_file, "r") as fp:
            metadata = json.load(fp)

        image_path = image_path if image_path else metadata["image"]
        image = rasterio.open(image_path)

        for mask_file in metadata["masks"]:
            data = np.load(mask_file, allow_pickle=True)

            tiled_masks.append([[data["mask"], data["image_bbox"]]])
            bboxes.append(box(*data["bbox"]))

            if data["timestamp"] != metadata["timestamp"]:
                logger.error(
                    "Timestamp in mask and metadat file don't match. Corrupted export?"
                )

        res = cls(
            image=image,
            tiled_masks=tiled_masks,
            bboxes=bboxes,
            confidence_threshold=metadata["confidence_threshold"],
            config=metadata["config"],
        )

        res.prediction_time_s = metadata["prediction_time_s"]

        return res

    def set_threshold(self, new_threshold: int) -> None:
        """Sets the threshold of the ProcessedResult, also regenerates
        prediction masks

        Args:
            new_threshold (double): new confidence threshold
        """
        self.confidence_threshold = new_threshold
        self._generate_masks()

    def save_masks(
        self,
        output_path: str,
        suffix: Optional[str] = "",
        prefix: Optional[str] = "",
        pad=0,
    ) -> None:
        """Save prediction masks for tree and canopy. If a source image is provided
        then it is used for georeferencing the output masks.

        Args:
            output_path (str): folder to store data
            suffix (str, optional): mask filename suffix
            prefix (str, optional): mask filename prefix

        """

        os.makedirs(output_path, exist_ok=True)

        canopy_mask = np.array(self.mask)

        if pad > 0:
            canopy_mask[:, :pad] = 0
            canopy_mask[:pad, :] = 0
            canopy_mask[:, -pad:] = 0
            canopy_mask[-pad:, :] = 0

        self._save_mask(
            mask=canopy_mask,
            output_path=os.path.join(output_path, f"{prefix}canopy_mask{suffix}.tif"),
        )

        if self.confidence_map.dtype != np.uint8:
            confidence_mask = np.array((255 * self.confidence_map)).astype(np.uint8)
        else:
            confidence_mask = self.confidence_map

        if pad > 0:
            confidence_mask[:, :pad] = 0
            confidence_mask[:pad, :] = 0
            confidence_mask[:, -pad:] = 0
            confidence_mask[-pad:, :] = 0

        self._save_mask(
            mask=confidence_mask,
            output_path=os.path.join(
                output_path, f"{prefix}canopy_confidence{suffix}.tif"
            ),
            binary=False,
        )

    def visualise(
        self,
        dpi=400,
        max_pixels: Optional[tuple[int, int]] = None,
        output_path=None,
        color_trees: Optional[tuple[int, int, int]] = (255, 105, 180),
        alpha: Optional[float] = 0.5,
        **kwargs: Any,
    ) -> None:
        """Visualise the results of the segmentation. If output path is not provided, the results
        will be displayed.

        Args:
            dpi (int, optional): dpi of the output image. Defaults to 200.
            max_pixels: maximum image size
            output_path (str, optional): path to save the output plots. Defaults to None.
            color_trees (tuple, optional): RGB tuple defining the colour for tree annotation
            alpha (float, optional): Alpha opacity for confidence mask when overlaid on original image
            **kwargs (Any): remaining arguments passed to figure creation

        """

        confidence_map = self.confidence_map

        reshape_factor = 1
        if max_pixels is not None:
            reshape_factor = min(
                max_pixels[0] / self.valid_window.height,
                max_pixels[1] / self.valid_window.width,
            )
            reshape_factor = min(reshape_factor, 1)

        shape = (
            math.ceil(self.valid_window.height * reshape_factor),
            math.ceil(self.valid_window.width * reshape_factor),
        )

        vis_image = self.image.read(
            out_shape=(self.image.count, shape[0], shape[1]),
            resampling=Resampling.bilinear,
            masked=True,
            window=self.valid_window,
        ).transpose(1, 2, 0)

        if self.valid_mask is not None:
            if reshape_factor != 1:
                vis_image = vis_image * np.expand_dims(
                    resize(self.valid_mask, shape), -1
                )
            else:
                vis_image = vis_image * np.expand_dims(self.valid_mask, -1)

        resized_confidence_map = confidence_map
        if reshape_factor < 1:
            resized_confidence_map = resize(confidence_map, shape)

        # Normal figure
        fig = plt.figure(dpi=dpi, **kwargs)
        ax = plt.axes()

        from rasterio import plot as rio_plot
        from rasterio import warp

        src, transform = warp.reproject(
            self.image.read(),
            src_crs=self.image.crs,
            src_transform=self.image.transform,
            dst_crs="EPSG:4326",
        )
        rio_plot.show(src, transform=transform, ax=ax, aspect="equal")

        ax.tick_params(axis="both", which="major", labelsize="x-small")
        ax.tick_params(axis="both", which="minor", labelsize="xx-small")
        ax.tick_params(axis="x", labelrotation=45)

        if output_path is not None:
            plt.savefig(os.path.join(output_path, "raw_image.jpg"), bbox_inches="tight")

        # Canopy Mask
        fig = plt.figure(dpi=dpi, **kwargs)
        ax = plt.axes()
        ax.tick_params(axis="both", which="major", labelsize="x-small")
        ax.tick_params(axis="both", which="minor", labelsize="xx-small")
        ax.imshow(vis_image)

        confidence_mask_image = np.zeros(
            (*resized_confidence_map.shape, 4), dtype=np.uint8
        )
        confidence_mask_image[
            resized_confidence_map > self.confidence_threshold
        ] = list(color_trees) + [255]
        ax.imshow(confidence_mask_image, alpha=alpha)

        if output_path is not None:
            plt.savefig(
                os.path.join(output_path, "canopy_overlay.jpg"), bbox_inches="tight"
            )

        # Confidence Map
        fig = plt.figure(dpi=dpi, **kwargs)
        ax = plt.axes()
        ax.tick_params(axis="both", which="major", labelsize="x-small")
        ax.tick_params(axis="both", which="minor", labelsize="xx-small")
        import matplotlib.colors

        palette = np.array(
            [
                (1, 1, 1, 0),
                (218 / 255, 215 / 255, 205 / 255, 1),
                (163 / 255, 177 / 255, 138 / 255, 1),
                (88 / 255, 129 / 255, 87 / 255, 1),
                (58 / 255, 9 / 255, 64 / 255, 1),
                (52 / 255, 78 / 255, 65 / 255, 1),
            ]
        )

        cmap = matplotlib.colors.ListedColormap(colors=palette)
        bounds = [0.2, 0.4, 0.6, 0.7, 0.8, 0.9]
        norm = matplotlib.colors.BoundaryNorm(bounds, cmap.N)

        im = ax.imshow(resized_confidence_map, cmap=cmap, norm=norm)
        cax = fig.add_axes(
            [
                ax.get_position().x1 + 0.01,
                ax.get_position().y0,
                0.02,
                ax.get_position().height,
            ]
        )

        cbar = plt.colorbar(
            im,
            cax=cax,
            extend="both",
            ticks=bounds,
            spacing="proportional",
            orientation="vertical",
        )
        cbar.set_label("Confidence", size="x-small")
        cbar.ax.tick_params(labelsize="xx-small")

        if output_path is not None:
            plt.savefig(
                os.path.join(output_path, "canopy_mask.jpg"), bbox_inches="tight"
            )

        if output_path is None:
            plt.show()

    def _filter_roi(self):
        if self.valid_region is not None:
            self.valid_window = rasterio.features.geometry_window(
                self.image, [self.valid_region]
            )

            self.valid_mask = rasterio.features.geometry_mask(
                [self.valid_region],
                out_shape=self.image.shape,
                transform=self.image.transform,
                invert=True,
            )[self.valid_window.toslices()]

        else:
            logger.warning("Unable to filter instances as no ROI has been set.")

    def __str__(self) -> str:
        """String representation, returns canopy cover for image."""
        return f" canopy_cover={self.canopy_cover:.4f})"

    def _repr_html_(self):
        # Save the plot to a SVG buffer
        from io import BytesIO

        buf = BytesIO()
        plt.imshow(self.confidence_map)
        plt.savefig(buf, format="svg")
        plt.tight_layout()
        plt.close()
        buf.seek(0)
        return buf.getvalue().decode("utf-8")

__str__()

String representation, returns canopy cover for image.

Source code in src/tcd_pipeline/result/semanticsegmentationresult.py 
381
382
383
def __str__(self) -> str:
    """String representation, returns canopy cover for image."""
    return f" canopy_cover={self.canopy_cover:.4f})"

load_serialisation(input_file, image_path=None) classmethod

Loads a ProcessedResult based on a json serialization file.

Parameters:

Name Type Description Default
input_file str

serialised instance metadata JSON file

 required
image_path Optional[str]

image path, optional

 None

Returns: SegmentationResult: SegmentationResult described by the file

Source code in src/tcd_pipeline/result/semanticsegmentationresult.py 
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
@classmethod
def load_serialisation(cls, input_file: str, image_path: Optional[str] = None):
    """Loads a ProcessedResult based on a json serialization file.

    Args:
        input_file (str): serialised instance metadata JSON file
        image_path (Optional[str]): image path, optional
    Returns:
        SegmentationResult: SegmentationResult described by the file
    """
    tiled_masks = []
    bboxes = []

    with open(input_file, "r") as fp:
        metadata = json.load(fp)

    image_path = image_path if image_path else metadata["image"]
    image = rasterio.open(image_path)

    for mask_file in metadata["masks"]:
        data = np.load(mask_file, allow_pickle=True)

        tiled_masks.append([[data["mask"], data["image_bbox"]]])
        bboxes.append(box(*data["bbox"]))

        if data["timestamp"] != metadata["timestamp"]:
            logger.error(
                "Timestamp in mask and metadat file don't match. Corrupted export?"
            )

    res = cls(
        image=image,
        tiled_masks=tiled_masks,
        bboxes=bboxes,
        confidence_threshold=metadata["confidence_threshold"],
        config=metadata["config"],
    )

    res.prediction_time_s = metadata["prediction_time_s"]

    return res

save_masks(output_path, suffix='', prefix='', pad=0)

Save prediction masks for tree and canopy. If a source image is provided then it is used for georeferencing the output masks.

Parameters:

Name Type Description Default
output_path str

folder to store data

 required
suffix str

mask filename suffix

 ''
prefix str

mask filename prefix

 ''
Source code in src/tcd_pipeline/result/semanticsegmentationresult.py 
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
def save_masks(
    self,
    output_path: str,
    suffix: Optional[str] = "",
    prefix: Optional[str] = "",
    pad=0,
) -> None:
    """Save prediction masks for tree and canopy. If a source image is provided
    then it is used for georeferencing the output masks.

    Args:
        output_path (str): folder to store data
        suffix (str, optional): mask filename suffix
        prefix (str, optional): mask filename prefix

    """

    os.makedirs(output_path, exist_ok=True)

    canopy_mask = np.array(self.mask)

    if pad > 0:
        canopy_mask[:, :pad] = 0
        canopy_mask[:pad, :] = 0
        canopy_mask[:, -pad:] = 0
        canopy_mask[-pad:, :] = 0

    self._save_mask(
        mask=canopy_mask,
        output_path=os.path.join(output_path, f"{prefix}canopy_mask{suffix}.tif"),
    )

    if self.confidence_map.dtype != np.uint8:
        confidence_mask = np.array((255 * self.confidence_map)).astype(np.uint8)
    else:
        confidence_mask = self.confidence_map

    if pad > 0:
        confidence_mask[:, :pad] = 0
        confidence_mask[:pad, :] = 0
        confidence_mask[:, -pad:] = 0
        confidence_mask[-pad:, :] = 0

    self._save_mask(
        mask=confidence_mask,
        output_path=os.path.join(
            output_path, f"{prefix}canopy_confidence{suffix}.tif"
        ),
        binary=False,
    )

serialise(output_folder, overwrite=True, file_prefix='results')

Serialise raw prediction masks. Masks are stored as NPZ files with the keys "mask" and "bbox" as well as a timestamp which can be used as a sanity check when loading. A JSON file containing a list of masks will also be created.

Parameters:

Name Type Description Default
output_folder str

output folder

 required
overwrite bool

overwrite existing data, defaults True

 True
file_prefix str

file name, defaults to results

 'results'
Source code in src/tcd_pipeline/result/semanticsegmentationresult.py 
 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
def serialise(
    self,
    output_folder: str,
    overwrite: bool = True,
    file_prefix: Optional[str] = "results",
) -> None:
    """Serialise raw prediction masks. Masks are stored as NPZ files with the
    keys "mask" and "bbox" as well as a timestamp which can be used as a sanity
    check when loading. A JSON file containing a list of masks will also be created.

    Args:
        output_folder (str): output folder
        overwrite (bool, optional): overwrite existing data, defaults True
        file_prefix (str, optional): file name, defaults to results
    """

    logger.info(f"Serialising results to {output_folder}/{file_prefix}")
    os.makedirs(output_folder, exist_ok=True)

    meta_path = os.path.join(output_folder, f"{file_prefix}.json")

    if os.path.exists(meta_path) and not overwrite:
        logger.error(
            f"Output metadata already exists {meta_path}, will not overwrite."
        )
        return

    timestamp = time.time()
    metadata = {}
    metadata["image"] = self.image.name
    metadata["timestamp"] = int(timestamp)
    metadata["masks"] = []
    metadata["confidence_threshold"] = self.confidence_threshold
    metadata["prediction_time_s"] = self.prediction_time_s
    # metadata["config"] = dict(self.config)
    metadata["hardware"] = self.get_hardware_information()

    for i, item in enumerate(zip(self.masks, self.bboxes)):
        mask, bbox = item
        file_name = f"{file_prefix}_{i}.npz"
        output_path = os.path.join(output_folder, file_name)

        if os.path.exists(output_path) and not overwrite:
            logger.error(
                f"Output file already exists {output_path}, will not overwrite."
            )
            return
        np.savez_compressed(
            file=output_path,
            mask=mask[0][0],
            image_bbox=mask[0][1],
            bbox=np.array(bbox),
            timestamp=int(timestamp),
        )
        metadata["masks"].append(os.path.abspath(output_path))

    with open(meta_path, "w") as fp:
        json.dump(metadata, fp, indent=1)

set_threshold(new_threshold)

Sets the threshold of the ProcessedResult, also regenerates prediction masks

Parameters:

Name Type Description Default
new_threshold double

new confidence threshold

 required
Source code in src/tcd_pipeline/result/semanticsegmentationresult.py 
155
156
157
158
159
160
161
162
163
def set_threshold(self, new_threshold: int) -> None:
    """Sets the threshold of the ProcessedResult, also regenerates
    prediction masks

    Args:
        new_threshold (double): new confidence threshold
    """
    self.confidence_threshold = new_threshold
    self._generate_masks()

visualise(dpi=400, max_pixels=None, output_path=None, color_trees=(255, 105, 180), alpha=0.5, **kwargs)

Visualise the results of the segmentation. If output path is not provided, the results will be displayed.

Parameters:

Name Type Description Default
dpi int

dpi of the output image. Defaults to 200.

 400
max_pixels Optional[tuple[int, int]]

maximum image size

 None
output_path str

path to save the output plots. Defaults to None.

 None
color_trees tuple

RGB tuple defining the colour for tree annotation

 (255, 105, 180)
alpha float

Alpha opacity for confidence mask when overlaid on original image

 0.5
**kwargs Any

remaining arguments passed to figure creation

 {}
Source code in src/tcd_pipeline/result/semanticsegmentationresult.py 
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317
318
319
320
321
322
323
324
325
326
327
328
329
330
331
332
333
334
335
336
337
338
339
340
341
342
343
344
345
346
347
348
349
350
351
352
353
354
355
356
357
358
359
360
361
362
363
def visualise(
    self,
    dpi=400,
    max_pixels: Optional[tuple[int, int]] = None,
    output_path=None,
    color_trees: Optional[tuple[int, int, int]] = (255, 105, 180),
    alpha: Optional[float] = 0.5,
    **kwargs: Any,
) -> None:
    """Visualise the results of the segmentation. If output path is not provided, the results
    will be displayed.

    Args:
        dpi (int, optional): dpi of the output image. Defaults to 200.
        max_pixels: maximum image size
        output_path (str, optional): path to save the output plots. Defaults to None.
        color_trees (tuple, optional): RGB tuple defining the colour for tree annotation
        alpha (float, optional): Alpha opacity for confidence mask when overlaid on original image
        **kwargs (Any): remaining arguments passed to figure creation

    """

    confidence_map = self.confidence_map

    reshape_factor = 1
    if max_pixels is not None:
        reshape_factor = min(
            max_pixels[0] / self.valid_window.height,
            max_pixels[1] / self.valid_window.width,
        )
        reshape_factor = min(reshape_factor, 1)

    shape = (
        math.ceil(self.valid_window.height * reshape_factor),
        math.ceil(self.valid_window.width * reshape_factor),
    )

    vis_image = self.image.read(
        out_shape=(self.image.count, shape[0], shape[1]),
        resampling=Resampling.bilinear,
        masked=True,
        window=self.valid_window,
    ).transpose(1, 2, 0)

    if self.valid_mask is not None:
        if reshape_factor != 1:
            vis_image = vis_image * np.expand_dims(
                resize(self.valid_mask, shape), -1
            )
        else:
            vis_image = vis_image * np.expand_dims(self.valid_mask, -1)

    resized_confidence_map = confidence_map
    if reshape_factor < 1:
        resized_confidence_map = resize(confidence_map, shape)

    # Normal figure
    fig = plt.figure(dpi=dpi, **kwargs)
    ax = plt.axes()

    from rasterio import plot as rio_plot
    from rasterio import warp

    src, transform = warp.reproject(
        self.image.read(),
        src_crs=self.image.crs,
        src_transform=self.image.transform,
        dst_crs="EPSG:4326",
    )
    rio_plot.show(src, transform=transform, ax=ax, aspect="equal")

    ax.tick_params(axis="both", which="major", labelsize="x-small")
    ax.tick_params(axis="both", which="minor", labelsize="xx-small")
    ax.tick_params(axis="x", labelrotation=45)

    if output_path is not None:
        plt.savefig(os.path.join(output_path, "raw_image.jpg"), bbox_inches="tight")

    # Canopy Mask
    fig = plt.figure(dpi=dpi, **kwargs)
    ax = plt.axes()
    ax.tick_params(axis="both", which="major", labelsize="x-small")
    ax.tick_params(axis="both", which="minor", labelsize="xx-small")
    ax.imshow(vis_image)

    confidence_mask_image = np.zeros(
        (*resized_confidence_map.shape, 4), dtype=np.uint8
    )
    confidence_mask_image[
        resized_confidence_map > self.confidence_threshold
    ] = list(color_trees) + [255]
    ax.imshow(confidence_mask_image, alpha=alpha)

    if output_path is not None:
        plt.savefig(
            os.path.join(output_path, "canopy_overlay.jpg"), bbox_inches="tight"
        )

    # Confidence Map
    fig = plt.figure(dpi=dpi, **kwargs)
    ax = plt.axes()
    ax.tick_params(axis="both", which="major", labelsize="x-small")
    ax.tick_params(axis="both", which="minor", labelsize="xx-small")
    import matplotlib.colors

    palette = np.array(
        [
            (1, 1, 1, 0),
            (218 / 255, 215 / 255, 205 / 255, 1),
            (163 / 255, 177 / 255, 138 / 255, 1),
            (88 / 255, 129 / 255, 87 / 255, 1),
            (58 / 255, 9 / 255, 64 / 255, 1),
            (52 / 255, 78 / 255, 65 / 255, 1),
        ]
    )

    cmap = matplotlib.colors.ListedColormap(colors=palette)
    bounds = [0.2, 0.4, 0.6, 0.7, 0.8, 0.9]
    norm = matplotlib.colors.BoundaryNorm(bounds, cmap.N)

    im = ax.imshow(resized_confidence_map, cmap=cmap, norm=norm)
    cax = fig.add_axes(
        [
            ax.get_position().x1 + 0.01,
            ax.get_position().y0,
            0.02,
            ax.get_position().height,
        ]
    )

    cbar = plt.colorbar(
        im,
        cax=cax,
        extend="both",
        ticks=bounds,
        spacing="proportional",
        orientation="vertical",
    )
    cbar.set_label("Confidence", size="x-small")
    cbar.ax.tick_params(labelsize="xx-small")

    if output_path is not None:
        plt.savefig(
            os.path.join(output_path, "canopy_mask.jpg"), bbox_inches="tight"
        )

    if output_path is None:
        plt.show()

SemanticSegmentationResultFromGeotiff

Bases: SemanticSegmentationResult

Source code in src/tcd_pipeline/result/semanticsegmentationresult.py 
398
399
400
401
402
403
404
405
406
407
408
409
410
411
412
413
414
415
416
417
418
419
420
421
422
423
424
425
426
427
428
429
430
431
432
433
434
435
436
437
438
439
440
441
442
443
444
445
446
447
448
449
450
451
452
453
454
455
456
457
458
459
460
461
462
463
464
465
466
467
468
469
470
471
472
473
474
475
class SemanticSegmentationResultFromGeotiff(SemanticSegmentationResult):
    def __init__(
        self,
        image: rasterio.DatasetReader,
        prediction: rasterio.DatasetReader,
        confidence_threshold: float = 0.2,
        config: dict = None,
    ) -> None:
        self.image = image
        self.confidence_map = prediction
        self.valid_region = None
        self.valid_mask = None
        self.config = config
        self.prediction_time_s = -1

        self.valid_window = rasterio.windows.from_bounds(
            *self.image.bounds, transform=self.image.transform
        )

        self.set_threshold(confidence_threshold)

    def save_masks(
        self,
        output_path: str,
        suffix: Optional[str] = "",
        prefix: Optional[str] = "",
    ) -> None:
        """Save prediction masks canopy cover.

        Args:
            output_path (str): folder to store data
            suffix (str, optional): mask filename suffix
            prefix (str, optional): mask filename prefix

        """

        os.makedirs(output_path, exist_ok=True)

        try:
            cmd = [
                "gdal_calc.py",
                "-A",
                self.confidence_map.name,
                "--outfile",
                os.path.join(output_path, f"{prefix}canopy_mask{suffix}.tif"),
                "--calc",
                f"A>{self.confidence_threshold}",
                "--NoDataValue=0",
                "--quiet",
                "--type=Byte",
            ]

            import subprocess

            _ = subprocess.run(
                cmd, check=True, stdout=subprocess.PIPE, stderr=subprocess.PIPE
            )

        except subprocess.CalledProcessError as e:
            logger.error(f"Error: {e.stderr.decode('utf-8')}")
        except Exception as e:
            logger.error(f"Unexpected error: {str(e)}")

    def _generate_masks(self):
        pass

    def set_threshold(self, thresh: float = 0.5):
        self.confidence_threshold = thresh

    def serialise(self, *args, **kwargs):
        raise NotImplementedError(
            "This is not required for a result based on a GeoTIFF cache"
        )

    def load(self):
        raise NotImplementedError(
            "This is not required for a result based on a GeoTIFF cache"
        )

save_masks(output_path, suffix='', prefix='')

Save prediction masks canopy cover.

Parameters:

Name Type Description Default
output_path str

folder to store data

 required
suffix str

mask filename suffix

 ''
prefix str

mask filename prefix

 ''
Source code in src/tcd_pipeline/result/semanticsegmentationresult.py 
419
420
421
422
423
424
425
426
427
428
429
430
431
432
433
434
435
436
437
438
439
440
441
442
443
444
445
446
447
448
449
450
451
452
453
454
455
456
457
458
459
def save_masks(
    self,
    output_path: str,
    suffix: Optional[str] = "",
    prefix: Optional[str] = "",
) -> None:
    """Save prediction masks canopy cover.

    Args:
        output_path (str): folder to store data
        suffix (str, optional): mask filename suffix
        prefix (str, optional): mask filename prefix

    """

    os.makedirs(output_path, exist_ok=True)

    try:
        cmd = [
            "gdal_calc.py",
            "-A",
            self.confidence_map.name,
            "--outfile",
            os.path.join(output_path, f"{prefix}canopy_mask{suffix}.tif"),
            "--calc",
            f"A>{self.confidence_threshold}",
            "--NoDataValue=0",
            "--quiet",
            "--type=Byte",
        ]

        import subprocess

        _ = subprocess.run(
            cmd, check=True, stdout=subprocess.PIPE, stderr=subprocess.PIPE
        )

    except subprocess.CalledProcessError as e:
        logger.error(f"Error: {e.stderr.decode('utf-8')}")
    except Exception as e:
        logger.error(f"Unexpected error: {str(e)}")