Skip to content

Create Seamlines

Source code in spectralmatch/seamline/seamline.py
 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
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
class Seamline:
    @staticmethod
    def weighted(
        input_polygons: str,
        output_mask: str,
        *,
        rank_function: str,
        image_field_name: str = "image",
        input_layer: str | None = None,
        output_layer: str = "seamlines",
        rank_descending: bool = True,
        debug_logs: Universal.DebugLogs = False,
        resume_from_outputs: Literal["no", "yes", "validate"] = "no",
    ) -> str:
        """Generate seamline polygons by ranking image footprints with a weighted expression.

Args:
    input_polygons (str): Input polygon layer path. Each feature should represent an image footprint or a piece of one.
    output_mask (str): Output GeoPackage path for the ranked seamline polygons.
    rank_function (str): Ranking expression using field placeholders like ``{cloud_cover}`` or formulas like ``1 / ({sun_elevation} + 1)``.
    image_field_name (str, optional): Field containing the image identifier. Features sharing the same value are merged before ranking. Defaults to ``"image"``.
    input_layer (str | None, optional): Optional input layer name when reading multi-layer vector sources. Defaults to None.
    output_layer (str, optional): Output GeoPackage layer name. Defaults to ``"seamlines"``.
    rank_descending (bool, optional): If True, larger scores rank higher and remain on top. Defaults to True.
    debug_logs (bool, optional): If True, prints ranking details. Defaults to False.

Returns:
    str: Written output GeoPackage path.
"""
        _print_step_start("weighted_seamline")
        SeamlineValidation._validate_weighted_seamline(
            input_polygons=input_polygons,
            output_mask=output_mask,
            rank_function=rank_function,
            image_field_name=image_field_name,
            input_layer=input_layer,
            output_layer=output_layer,
            rank_descending=rank_descending,
        )
        if debug_logs:
            print(f"Input polygons: {input_polygons}")
            print(f"Output mask: {output_mask}")
            print(f"Rank function: {rank_function}")
        if _existing_outputs_are_reusable(
            [output_mask],
            resume_mode=resume_from_outputs,
            debug_logs=debug_logs,
            step_name="weighted_seamline",
        ):
            return output_mask
        _print_image_start(input_polygons, output_mask)
        result = weighted_seamline(
            input_polygons=input_polygons,
            output_mask=output_mask,
            rank_function=rank_function,
            image_field_name=image_field_name,
            input_layer=input_layer,
            output_layer=output_layer,
            rank_descending=rank_descending,
            debug_logs=debug_logs,
        )
        _print_image_completed(input_polygons, 1, 1)
        return result

    @staticmethod
    def voronoi(
        input_images: Universal.CreateInFolderOrListFiles,
        output_mask: str,
        *,
        aoi_path: str | None = None,
        vector_mask: tuple[str, str] | None = None,
        image_field_name: str = "image",
        min_point_spacing: float = 10,
        min_cut_length: float = 0,
        debug_logs: Universal.DebugLogs = False,
        debug_vectors_path: str | None = None,
        resume_from_outputs: Literal["no", "yes", "validate"] = "no",
    ) -> None:
        """Generates a Voronoi-based seamline mask from edge-matching polygons (EMPs) and writes the result to a vector file.

Args:
    input_images (str | List[str], required): Defines input files from a glob path, folder, or list of paths. Specify like: "/input/files/*.tif", "/input/folder" (assumes *.tif), ["/input/one.tif", "/input/two.tif"].
    output_mask (str): Output path for the final seamline polygon vector file.
    aoi_path (str, optional): Path to an AOI vector file to clip overlapping image polygons; default is None.
    vector_mask (Tuple[str, str] | None, optional): Optional polygon source to use instead of extracting EMPs from rasters. The tuple is (vector_path, field_name). For each input image, polygons are selected when the field value is included anywhere in the image name. Matching polygons for the same image are unioned together.
    min_point_spacing (float, optional): Minimum spacing between Voronoi seed points; default is 10.
    min_cut_length (float, optional): Minimum cutline segment length to retain; default is 0.
    debug_logs (Universal.DebugLogs, optional): Enables debug print statements if True; default is False.
    image_field_name (str, optional): Name of the attribute field for image ID in output; default is 'image'.
    debug_vectors_path (str | None, optional): Optional path to save debug layers (cutlines, intersections).

Outputs:
    Saves a polygon seamline layer to `output_mask`, and optionally saves intermediate cutlines to `debug_vectors_path`."""
        _print_step_start("voronoi_center_seamline")
        output_dir = os.path.dirname(output_mask)
        if output_dir:
            os.makedirs(output_dir, exist_ok=True)
        if debug_vectors_path:
            debug_dir = os.path.dirname(debug_vectors_path)
            if debug_dir:
                os.makedirs(debug_dir, exist_ok=True)
        if _existing_outputs_are_reusable(
            [output_mask],
            resume_mode=resume_from_outputs,
            debug_logs=debug_logs,
            step_name="voronoi_center_seamline",
        ):
            return

        Universal._validate(
            input_images=input_images,
        )
        SeamlineValidation._validate_voronoi_center_seamline(
            output_mask=output_mask,
            aoi_path=aoi_path,
            vector_mask=vector_mask,
            image_field_name=image_field_name,
            min_point_spacing=min_point_spacing,
            min_cut_length=min_cut_length,
            debug_vectors_path=debug_vectors_path,
        )
        input_image_paths = _resolve_paths(
            "search", input_images, kwargs={"default_file_pattern": "*.tif"}
        )
        input_image_names = _resolve_paths("name", input_image_paths)

        if vector_mask is None:
            emps = []
            crs = None
            for path in input_image_paths:
                _print_image_start(path, [output_mask, debug_vectors_path] if debug_vectors_path else output_mask)
                emp = _emp_polygon_from_image(path)
                emps.append(emp)
                if crs is None:
                    ds = gdal.Open(path, gdal.GA_ReadOnly)
                    crs = ds.GetProjectionRef()
                    ds = None
        else:
            for path in input_image_paths:
                _print_image_start(path, [output_mask, debug_vectors_path] if debug_vectors_path else output_mask)
            emps, crs = _load_emp_polygons_from_vector(
                input_image_paths=input_image_paths,
                input_image_names=input_image_names,
                vector_mask=vector_mask,
                debug_logs=debug_logs,
            )

        image_details = [
            {"Footprint area": f"{emp.area:.2f}", "Bounds": str(emp.bounds)} if debug_logs else {}
            for emp in emps
        ]

        if debug_vectors_path:
            if os.path.exists(debug_vectors_path):
                os.remove(debug_vectors_path)
            _save_emp_outlines(
                emps,
                input_image_paths,
                debug_vectors_path,
                crs,
                image_field_name=image_field_name,
            )

        cuts: list[LineString] = []
        for i, (a, b) in enumerate(combinations(emps, 2)):
            ov = a.intersection(b)
            if debug_logs:
                print(f"Overlap {i} area: {ov.area:.2f}")
            if not ov.is_empty:
                if debug_vectors_path:
                    _save_intersection_points(a, b, debug_vectors_path, crs, f"{i}")
                cut = _compute_centerline(
                    a,
                    b,
                    min_point_spacing,
                    min_cut_length,
                    debug_logs,
                    crs,
                    debug_vectors_path,
                )
                cuts.append(cut)

        if debug_vectors_path:
            schema = {"geometry": "LineString", "properties": {"pair_id": "str"}}
            with fiona.open(
                debug_vectors_path,
                "w",
                driver="GPKG",
                crs_wkt=crs,
                schema=schema,
                layer="cutlines",
            ) as dst:
                for idx, line in enumerate(cuts):
                    dst.write(
                        {
                            "geometry": mapping(line),
                            "properties": {"pair_id": f"{idx}"},
                        }
                    )

        segmented: list[Polygon] = []
        for idx, emp in enumerate(emps):
            relevant = [cut for cut in cuts if emp.intersects(cut)]
            seg = _segment_emp(emp, relevant, debug_logs)
            if debug_logs:
                image_details[idx].update({"Cuts": len(relevant), "Segmented area": f"{seg.area:.2f}"})
            segmented.append(seg)

        if aoi_path is not None:
            segmented = _mask_by_aoi(segmented, aoi_path)

        schema = {"geometry": "Polygon", "properties": {image_field_name: "str"}}
        with fiona.open(
            output_mask,
            "w",
            driver="GPKG",
            crs_wkt=crs,
            schema=schema,
            layer="seamlines",
        ) as dst:
            for image_name, poly in zip(input_image_names, segmented):
                dst.write(
                    {
                        "geometry": mapping(poly),
                        "properties": {image_field_name: image_name},
                    }
                )


        for completed, path in enumerate(input_image_paths, 1):
            _print_image_completed(path, completed, len(input_image_paths), details=image_details[completed - 1])

voronoi(input_images, output_mask, *, aoi_path=None, vector_mask=None, image_field_name='image', min_point_spacing=10, min_cut_length=0, debug_logs=False, debug_vectors_path=None, resume_from_outputs='no') staticmethod

Generates a Voronoi-based seamline mask from edge-matching polygons (EMPs) and writes the result to a vector file.

Parameters:

Name Type Description Default
input_images (str | List[str], required)

Defines input files from a glob path, folder, or list of paths. Specify like: "/input/files/.tif", "/input/folder" (assumes .tif), ["/input/one.tif", "/input/two.tif"].

required
output_mask str

Output path for the final seamline polygon vector file.

required
aoi_path str

Path to an AOI vector file to clip overlapping image polygons; default is None.

None
vector_mask Tuple[str, str] | None

Optional polygon source to use instead of extracting EMPs from rasters. The tuple is (vector_path, field_name). For each input image, polygons are selected when the field value is included anywhere in the image name. Matching polygons for the same image are unioned together.

None
min_point_spacing float

Minimum spacing between Voronoi seed points; default is 10.

10
min_cut_length float

Minimum cutline segment length to retain; default is 0.

0
debug_logs DebugLogs

Enables debug print statements if True; default is False.

False
image_field_name str

Name of the attribute field for image ID in output; default is 'image'.

'image'
debug_vectors_path str | None

Optional path to save debug layers (cutlines, intersections).

None
Outputs

Saves a polygon seamline layer to output_mask, and optionally saves intermediate cutlines to debug_vectors_path.

Source code in spectralmatch/seamline/seamline.py
 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
    @staticmethod
    def voronoi(
        input_images: Universal.CreateInFolderOrListFiles,
        output_mask: str,
        *,
        aoi_path: str | None = None,
        vector_mask: tuple[str, str] | None = None,
        image_field_name: str = "image",
        min_point_spacing: float = 10,
        min_cut_length: float = 0,
        debug_logs: Universal.DebugLogs = False,
        debug_vectors_path: str | None = None,
        resume_from_outputs: Literal["no", "yes", "validate"] = "no",
    ) -> None:
        """Generates a Voronoi-based seamline mask from edge-matching polygons (EMPs) and writes the result to a vector file.

Args:
    input_images (str | List[str], required): Defines input files from a glob path, folder, or list of paths. Specify like: "/input/files/*.tif", "/input/folder" (assumes *.tif), ["/input/one.tif", "/input/two.tif"].
    output_mask (str): Output path for the final seamline polygon vector file.
    aoi_path (str, optional): Path to an AOI vector file to clip overlapping image polygons; default is None.
    vector_mask (Tuple[str, str] | None, optional): Optional polygon source to use instead of extracting EMPs from rasters. The tuple is (vector_path, field_name). For each input image, polygons are selected when the field value is included anywhere in the image name. Matching polygons for the same image are unioned together.
    min_point_spacing (float, optional): Minimum spacing between Voronoi seed points; default is 10.
    min_cut_length (float, optional): Minimum cutline segment length to retain; default is 0.
    debug_logs (Universal.DebugLogs, optional): Enables debug print statements if True; default is False.
    image_field_name (str, optional): Name of the attribute field for image ID in output; default is 'image'.
    debug_vectors_path (str | None, optional): Optional path to save debug layers (cutlines, intersections).

Outputs:
    Saves a polygon seamline layer to `output_mask`, and optionally saves intermediate cutlines to `debug_vectors_path`."""
        _print_step_start("voronoi_center_seamline")
        output_dir = os.path.dirname(output_mask)
        if output_dir:
            os.makedirs(output_dir, exist_ok=True)
        if debug_vectors_path:
            debug_dir = os.path.dirname(debug_vectors_path)
            if debug_dir:
                os.makedirs(debug_dir, exist_ok=True)
        if _existing_outputs_are_reusable(
            [output_mask],
            resume_mode=resume_from_outputs,
            debug_logs=debug_logs,
            step_name="voronoi_center_seamline",
        ):
            return

        Universal._validate(
            input_images=input_images,
        )
        SeamlineValidation._validate_voronoi_center_seamline(
            output_mask=output_mask,
            aoi_path=aoi_path,
            vector_mask=vector_mask,
            image_field_name=image_field_name,
            min_point_spacing=min_point_spacing,
            min_cut_length=min_cut_length,
            debug_vectors_path=debug_vectors_path,
        )
        input_image_paths = _resolve_paths(
            "search", input_images, kwargs={"default_file_pattern": "*.tif"}
        )
        input_image_names = _resolve_paths("name", input_image_paths)

        if vector_mask is None:
            emps = []
            crs = None
            for path in input_image_paths:
                _print_image_start(path, [output_mask, debug_vectors_path] if debug_vectors_path else output_mask)
                emp = _emp_polygon_from_image(path)
                emps.append(emp)
                if crs is None:
                    ds = gdal.Open(path, gdal.GA_ReadOnly)
                    crs = ds.GetProjectionRef()
                    ds = None
        else:
            for path in input_image_paths:
                _print_image_start(path, [output_mask, debug_vectors_path] if debug_vectors_path else output_mask)
            emps, crs = _load_emp_polygons_from_vector(
                input_image_paths=input_image_paths,
                input_image_names=input_image_names,
                vector_mask=vector_mask,
                debug_logs=debug_logs,
            )

        image_details = [
            {"Footprint area": f"{emp.area:.2f}", "Bounds": str(emp.bounds)} if debug_logs else {}
            for emp in emps
        ]

        if debug_vectors_path:
            if os.path.exists(debug_vectors_path):
                os.remove(debug_vectors_path)
            _save_emp_outlines(
                emps,
                input_image_paths,
                debug_vectors_path,
                crs,
                image_field_name=image_field_name,
            )

        cuts: list[LineString] = []
        for i, (a, b) in enumerate(combinations(emps, 2)):
            ov = a.intersection(b)
            if debug_logs:
                print(f"Overlap {i} area: {ov.area:.2f}")
            if not ov.is_empty:
                if debug_vectors_path:
                    _save_intersection_points(a, b, debug_vectors_path, crs, f"{i}")
                cut = _compute_centerline(
                    a,
                    b,
                    min_point_spacing,
                    min_cut_length,
                    debug_logs,
                    crs,
                    debug_vectors_path,
                )
                cuts.append(cut)

        if debug_vectors_path:
            schema = {"geometry": "LineString", "properties": {"pair_id": "str"}}
            with fiona.open(
                debug_vectors_path,
                "w",
                driver="GPKG",
                crs_wkt=crs,
                schema=schema,
                layer="cutlines",
            ) as dst:
                for idx, line in enumerate(cuts):
                    dst.write(
                        {
                            "geometry": mapping(line),
                            "properties": {"pair_id": f"{idx}"},
                        }
                    )

        segmented: list[Polygon] = []
        for idx, emp in enumerate(emps):
            relevant = [cut for cut in cuts if emp.intersects(cut)]
            seg = _segment_emp(emp, relevant, debug_logs)
            if debug_logs:
                image_details[idx].update({"Cuts": len(relevant), "Segmented area": f"{seg.area:.2f}"})
            segmented.append(seg)

        if aoi_path is not None:
            segmented = _mask_by_aoi(segmented, aoi_path)

        schema = {"geometry": "Polygon", "properties": {image_field_name: "str"}}
        with fiona.open(
            output_mask,
            "w",
            driver="GPKG",
            crs_wkt=crs,
            schema=schema,
            layer="seamlines",
        ) as dst:
            for image_name, poly in zip(input_image_names, segmented):
                dst.write(
                    {
                        "geometry": mapping(poly),
                        "properties": {image_field_name: image_name},
                    }
                )


        for completed, path in enumerate(input_image_paths, 1):
            _print_image_completed(path, completed, len(input_image_paths), details=image_details[completed - 1])

weighted(input_polygons, output_mask, *, rank_function, image_field_name='image', input_layer=None, output_layer='seamlines', rank_descending=True, debug_logs=False, resume_from_outputs='no') staticmethod

Generate seamline polygons by ranking image footprints with a weighted expression.

Parameters:

Name Type Description Default
input_polygons str

Input polygon layer path. Each feature should represent an image footprint or a piece of one.

required
output_mask str

Output GeoPackage path for the ranked seamline polygons.

required
rank_function str

Ranking expression using field placeholders like {cloud_cover} or formulas like 1 / ({sun_elevation} + 1).

required
image_field_name str

Field containing the image identifier. Features sharing the same value are merged before ranking. Defaults to "image".

'image'
input_layer str | None

Optional input layer name when reading multi-layer vector sources. Defaults to None.

None
output_layer str

Output GeoPackage layer name. Defaults to "seamlines".

'seamlines'
rank_descending bool

If True, larger scores rank higher and remain on top. Defaults to True.

True
debug_logs bool

If True, prints ranking details. Defaults to False.

False

Returns:

Name Type Description
str str

Written output GeoPackage path.

Source code in spectralmatch/seamline/seamline.py
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
    @staticmethod
    def weighted(
        input_polygons: str,
        output_mask: str,
        *,
        rank_function: str,
        image_field_name: str = "image",
        input_layer: str | None = None,
        output_layer: str = "seamlines",
        rank_descending: bool = True,
        debug_logs: Universal.DebugLogs = False,
        resume_from_outputs: Literal["no", "yes", "validate"] = "no",
    ) -> str:
        """Generate seamline polygons by ranking image footprints with a weighted expression.

Args:
    input_polygons (str): Input polygon layer path. Each feature should represent an image footprint or a piece of one.
    output_mask (str): Output GeoPackage path for the ranked seamline polygons.
    rank_function (str): Ranking expression using field placeholders like ``{cloud_cover}`` or formulas like ``1 / ({sun_elevation} + 1)``.
    image_field_name (str, optional): Field containing the image identifier. Features sharing the same value are merged before ranking. Defaults to ``"image"``.
    input_layer (str | None, optional): Optional input layer name when reading multi-layer vector sources. Defaults to None.
    output_layer (str, optional): Output GeoPackage layer name. Defaults to ``"seamlines"``.
    rank_descending (bool, optional): If True, larger scores rank higher and remain on top. Defaults to True.
    debug_logs (bool, optional): If True, prints ranking details. Defaults to False.

Returns:
    str: Written output GeoPackage path.
"""
        _print_step_start("weighted_seamline")
        SeamlineValidation._validate_weighted_seamline(
            input_polygons=input_polygons,
            output_mask=output_mask,
            rank_function=rank_function,
            image_field_name=image_field_name,
            input_layer=input_layer,
            output_layer=output_layer,
            rank_descending=rank_descending,
        )
        if debug_logs:
            print(f"Input polygons: {input_polygons}")
            print(f"Output mask: {output_mask}")
            print(f"Rank function: {rank_function}")
        if _existing_outputs_are_reusable(
            [output_mask],
            resume_mode=resume_from_outputs,
            debug_logs=debug_logs,
            step_name="weighted_seamline",
        ):
            return output_mask
        _print_image_start(input_polygons, output_mask)
        result = weighted_seamline(
            input_polygons=input_polygons,
            output_mask=output_mask,
            rank_function=rank_function,
            image_field_name=image_field_name,
            input_layer=input_layer,
            output_layer=output_layer,
            rank_descending=rank_descending,
            debug_logs=debug_logs,
        )
        _print_image_completed(input_polygons, 1, 1)
        return result