Skip to content

API Reference

osm_ai_helper.download_osm

download_osm(area, output_dir, selector, discard=None)

Download OSM elements for the given areas and selector.

Parameters:

Name Type Description Default
output_dir str

Output directory.

 required
selector str

OSM tag to select elements. Uses the Overpass API.

Example: "leisure=swimming_pool"

 required
area str

Name of area to download. Can be city, state, country, etc. Uses the Nominatim API.

 required
discard Optional[dict[str, str]]

Discard elements matching any of the given tags. Defaults to None. Example: {"location": "indoor", "building": "yes"}

 None
Source code in src/osm_ai_helper/download_osm.py 
11
12
13
14
15
16
17
18
19
20
21
22
23
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
@logger.catch(reraise=True)
def download_osm(
    area: str,
    output_dir: str,
    selector: str,
    discard: Optional[Dict[str, str]] = None,
):
    """Download OSM elements for the given areas and selector.

    Args:
        output_dir (str): Output directory.
        selector (str): OSM tag to select elements.
            Uses the [Overpass API](https://wiki.openstreetmap.org/wiki/Overpass_API/Language_Guide).

            Example: ["leisure=swimming_pool"](https://wiki.openstreetmap.org/wiki/Tag:leisure%3Dswimming_pool)

        area (str): Name of area to download.
            Can be city, state, country, etc.
            Uses the [Nominatim API](https://nominatim.org/release-docs/develop/api/Search/).

        discard (Optional[dict[str, str]], optional): Discard elements matching
            any of the given tags.
            Defaults to None.
            Example: {"location": "indoor", "building": "yes"}
    """
    output_path = Path(output_dir)
    output_path.mkdir(exist_ok=True, parents=True)

    discard = discard or {}

    logger.info(f"Downloading osm data for {area}")
    elements = [
        element
        for element in get_elements(selector, area=area)
        if all(element.get("tags", {}).get(k) != v for k, v in discard.items())
    ]

    output_file = output_path / f"{area}.json"
    logger.info(f"Writing {len(elements)} elements to {output_file}")
    output_file.write_text(json.dumps(elements))
    logger.success("Done!")

osm_ai_helper.group_elements_and_download_tiles

group_elements_and_download_tiles(elements_file, output_dir, mapbox_token, zoom=18)

Groups the elements by tile and downloads the satellite image corresponding to the tile.

Parameters:

Name Type Description Default
elements_file str

Path to the JSON file containing OSM elements. See download_osm.

 required
output_dir str

Output directory. The images and annotations will be saved in this directory. The images will be saved as JPEG files and the annotations as JSON files. The names of the files will be in the format {zoom}_{tile_col}_{tile_row}.

 required
mapbox_token str

Mapbox token.

 required
zoom int

Zoom level of the tiles to download. See https://docs.mapbox.com/help/glossary/zoom-level/. Defaults to 18.

 18
Source code in src/osm_ai_helper/group_elements_and_download_tiles.py 
12
13
14
15
16
17
18
19
20
21
22
23
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
@logger.catch(reraise=True)
def group_elements_and_download_tiles(
    elements_file: str, output_dir: str, mapbox_token: str, zoom: int = 18
):
    """
    Groups the elements by tile and downloads the satellite image corresponding to the tile.

    Args:
        elements_file (str): Path to the JSON file containing OSM elements.
            See [download_osm][osm_ai_helper.download_osm.download_osm].
        output_dir (str): Output directory.
            The images and annotations will be saved in this directory.
            The images will be saved as JPEG files and the annotations as JSON files.
            The names of the files will be in the format `{zoom}_{tile_col}_{tile_row}`.
        mapbox_token (str): [Mapbox](https://console.mapbox.com/) token.
        zoom (int, optional): Zoom level of the tiles to download.
            See https://docs.mapbox.com/help/glossary/zoom-level/.
            Defaults to 18.
    """
    annotation_path = Path(elements_file)
    output_path = Path(output_dir)
    output_path.mkdir(exist_ok=True, parents=True)

    elements = json.loads(annotation_path.read_text())

    logger.info("Grouping elements by tile")
    grouped = group_elements_by_tile(elements, zoom)

    download_tile_inputs = []
    annotation_names = []
    for (tile_col, tile_row), group in grouped.items():
        output_name = f"{zoom}_{tile_col}_{tile_row}"
        download_tile_inputs.append(
            (zoom, tile_col, tile_row, mapbox_token, f"{output_path / output_name}.jpg")
        )
        annotation_names.append(f"{output_path / output_name}.json")

    logger.info(f"Downloading tiles to {output_path}")
    with ThreadPoolExecutor() as executor:
        futures = [
            executor.submit(download_tile, *inputs) for inputs in download_tile_inputs
        ]
        for future in as_completed(futures):
            future.result()

    logger.info(f"Saving annotations to {output_path}")
    for annotation_name in annotation_names:
        Path(annotation_name).write_text(
            json.dumps(
                {
                    "elements": group,
                }
            )
        )

osm_ai_helper.convert_to_yolo_dataset

convert_to_yolo_dataset(input_dir)

Convert the output of group_elements_and_download_tiles.py to the YOLO format.

Parameters:

Name Type Description Default
input_dir str

Input directory containing the images and annotations. The images are expected to be in the format zoom_tile_col_tile_row.jpg. The annotations are expected to be in the format zoom_tile_col_tile_row.json.

 required
Source code in src/osm_ai_helper/convert_to_yolo_dataset.py 
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
@logger.catch(reraise=True)
def convert_to_yolo_dataset(
    input_dir: str,
):
    """Convert the output of `group_elements_and_download_tiles.py` to the [YOLO format](https://docs.ultralytics.com/datasets/detect/).

    Args:
        input_dir (str): Input directory containing the images and annotations.
            The images are expected to be in the format `zoom_tile_col_tile_row.jpg`.
            The annotations are expected to be in the format `zoom_tile_col_tile_row.json`.
    """
    input_path = Path(input_dir)

    for image_path in input_path.glob("**/*.jpg"):
        annotation_path = image_path.with_suffix(".json")
        annotation = json.loads(annotation_path.read_text())
        zoom, tile_col, tile_row = map(int, image_path.stem.split("_"))

        yolo_annotation = grouped_elements_to_annotation(
            annotation["elements"], zoom, tile_col, tile_row
        )
        image_path.with_suffix(".txt").write_text(yolo_annotation)

grouped_elements_to_annotation(group, zoom, tile_col, tile_row)

Output format: https://docs.ultralytics.com/datasets/detect/

Source code in src/osm_ai_helper/convert_to_yolo_dataset.py 
16
17
18
19
20
21
22
23
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
def grouped_elements_to_annotation(
    group: List[Dict], zoom: int, tile_col: int, tile_row: int
) -> str:
    """
    Output format: https://docs.ultralytics.com/datasets/detect/
    """
    annotation = ""
    left_pixel = tile_col * TILE_SIZE
    top_pixel = tile_row * TILE_SIZE
    bbox = box(left_pixel, top_pixel, left_pixel + TILE_SIZE, top_pixel + TILE_SIZE)
    for element in group:
        pixel_polygon = [
            meters_col_row_to_pixel_col_row(
                *lat_lon_to_meters_col_row(point["lat"], point["lon"]), zoom
            )
            for point in element["geometry"]
        ]
        try:
            bounded_polygon = Polygon(pixel_polygon).intersection(bbox)
        except AttributeError:
            continue
        min_col, min_row, max_col, max_row = bounded_polygon.bounds
        col_center = (min_col + max_col) / 2
        col_center = (col_center - left_pixel) / TILE_SIZE
        col_center = round(col_center, 2)
        row_center = (min_row + max_row) / 2
        row_center = (row_center - top_pixel) / TILE_SIZE
        row_center = round(row_center, 2)
        width = max_col - min_col
        width /= TILE_SIZE
        width = round(width, 2)
        height = max_row - min_row
        height /= TILE_SIZE
        height = round(height, 2)
        annotation += f"0 {col_center} {row_center} {width} {height}\n"

    return annotation

osm_ai_helper.run_inference

run_inference(yolo_model_file, output_dir, lat_lon, margin=1, sam_model='facebook/sam2.1-hiera-small', selector='leisure=swimming_pool', zoom=18, save_full_images=True, bbox_conf=0.5, batch_size=32)

Run inference on a given location.

Parameters:

Name Type Description Default
yolo_model_file str

Path to the YOLO model file.

 required
output_dir str

Output directory. The images and annotations will be saved in this directory. The images will be saved as PNG files and the annotations as JSON files. The names of the files will be in the format {zoom}_{tile_col}_{tile_row}.

 required
lat_lon Tuple[float, float]

Latitude and longitude of the location.

 required
margin int

Number of tiles around the location. Defaults to 1.

 1
sam_model str

SAM2 model to use. Defaults to "facebook/sam2.1-hiera-small".

 'facebook/sam2.1-hiera-small'
selector str

OpenStreetMap selector. Defaults to "leisure=swimming_pool".

 'leisure=swimming_pool'
zoom int

Zoom level. Defaults to 18. See https://docs.mapbox.com/help/glossary/zoom-level/.

 18
bbox_conf float

Sets the minimum confidence threshold for detections. Defaults to 0.4.

 0.5
batch_size int

Batch size for prediction. Defaults to 32.

 32
Source code in src/osm_ai_helper/run_inference.py 
 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
@logger.catch(reraise=True)
def run_inference(
    yolo_model_file: str,
    output_dir: str,
    lat_lon: Tuple[float, float],
    margin: int = 1,
    sam_model: str = "facebook/sam2.1-hiera-small",
    selector: str = "leisure=swimming_pool",
    zoom: int = 18,
    save_full_images: bool = True,
    bbox_conf: float = 0.5,
    batch_size: int = 32,
):
    """
    Run inference on a given location.

    Args:
        yolo_model_file (str): Path to the [YOLO](https://docs.ultralytics.com/tasks/detect/) model file.
        output_dir (str): Output directory.
            The images and annotations will be saved in this directory.
            The images will be saved as PNG files and the annotations as JSON files.
            The names of the files will be in the format `{zoom}_{tile_col}_{tile_row}`.
        lat_lon (Tuple[float, float]): Latitude and longitude of the location.
        margin (int, optional): Number of tiles around the location.
            Defaults to 1.
        sam_model (str, optional): [SAM2](https://github.com/facebookresearch/sam2) model to use.
            Defaults to "facebook/sam2.1-hiera-small".
        selector (str, optional): OpenStreetMap selector.
            Defaults to "leisure=swimming_pool".
        zoom (int, optional): Zoom level.
            Defaults to 18.
            See https://docs.mapbox.com/help/glossary/zoom-level/.
        bbox_conf (float): Sets the minimum confidence threshold for detections.
            Defaults to 0.4.
        batch_size (int): Batch size for prediction.
            Defaults to 32.
    """
    bbox_predictor = YOLO(yolo_model_file)
    sam_predictor = SAM2ImagePredictor.from_pretrained(
        sam_model, device="cuda" if torch.cuda.is_available() else "cpu"
    )

    bbox = lat_lon_to_bbox(*lat_lon, zoom, margin)

    output_path = Path(output_dir) / f"{zoom}_{'_'.join(map(str, bbox))}"
    output_path.mkdir(exist_ok=True, parents=True)

    logger.info(f"Downloading elements for {selector} in {bbox}")
    elements = get_elements(selector, bbox=bbox)
    grouped_elements = group_elements_by_tile(elements, zoom)
    logger.info(f"Found {len(elements)} elements")

    logger.info(f"Downloading all tiles within {bbox}")
    stacked_image, stacked_mask = download_stacked_image_and_mask(
        bbox, grouped_elements, zoom, os.environ["MAPBOX_TOKEN"]
    )
    if save_full_images:
        Image.fromarray(stacked_image).save(output_path / "full_image.png")
        Image.fromarray(stacked_mask).save(output_path / "full_mask.png")

    logger.info("Predicting on stacked image")
    # Change to BGR for inference
    stacked_output = tile_prediction(
        bbox_predictor,
        sam_predictor,
        stacked_image[:, :, ::-1],
        bbox_conf=bbox_conf,
        batch_size=batch_size,
    )

    logger.info("Finding existing, new and missed polygons")
    existing, new, missed = polygon_evaluation(stacked_mask, stacked_output)
    logger.info(f"{len(existing)} exiting, {len(new)} new and {len(missed)} missied.")
    logger.info("Painting evaluation")
    stacked_image_pil = Image.fromarray(stacked_image)
    painted_img = paint_polygon_evaluation(stacked_image_pil, existing, new, missed)

    if save_full_images:
        painted_img.save(output_path / "full_image_painted.png")

    _, west, north, _ = bbox
    left_col, top_row = lat_lon_to_tile_col_row(north, west, zoom)
    top_pixel = top_row * TILE_SIZE
    left_pixel = left_col * TILE_SIZE

    logger.info("Saving new polygons")
    for n, polygon in enumerate(new):
        lon_lat_polygon = pixel_polygon_to_lat_lon_polygon(
            polygon, top_pixel, left_pixel, zoom
        )

        with open(f"{output_path}/{n}.json", "w") as f:
            json.dump(lon_lat_polygon, f)

        crop_polygon(polygon, painted_img, margin=100).save(
            f"{output_path}/{n}_painted.png"
        )

        crop_polygon(polygon, stacked_image_pil, margin=100).save(
            f"{output_path}/{n}.png"
        )

    return output_path, existing, new, missed

osm_ai_helper.export_osm

export_osm(results_dir, output_dir, tags=None)

Export the polygons in results_dir to an .osc file.

Parameters:

Name Type Description Default
results_dir str

Directory containing the results. The results should be in the format of *.json files. See run_inference.

 required
Source code in src/osm_ai_helper/export_osm.py 
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
@logger.catch(reraise=True)
def export_osm(results_dir: str, output_dir: str, tags: dict = None) -> None:
    """
    Export the polygons in `results_dir` to an [`.osc`](https://wiki.openstreetmap.org/wiki/OsmChange) file.

    Args:
        results_dir (str): Directory containing the results.
            The results should be in the format of `*.json` files.
            See [`run_inference`][osm_ai_helper.run_inference.run_inference].
    """
    output_dir = Path(output_dir)
    output_dir.mkdir(exist_ok=True, parents=True)

    lon_lat_polygons = [
        json.loads(result.read_text()) for result in Path(results_dir).glob("*.json")
    ]
    logger.info(f"Converting {len(lon_lat_polygons)} polygons to OsmChange format.")
    logger.info(
        f"Each set of {MAX_ELEMENTS_PER_OSMCHANGE} polygons will be saved to a separate file."
    )
    for n_polygon in range(0, len(lon_lat_polygons), MAX_ELEMENTS_PER_OSMCHANGE):
        to_be_converted = lon_lat_polygons[
            n_polygon : n_polygon + MAX_ELEMENTS_PER_OSMCHANGE
        ]
        osmchange = convert_polygons(to_be_converted, tags)
        output_file = f"{n_polygon}-{n_polygon + MAX_ELEMENTS_PER_OSMCHANGE}.osc"
        logger.info(f"Writing OsmChange to {output_file}")
        (output_dir / output_file).write_bytes(ET.tostring(osmchange, "utf-8"))

osm_ai_helper.utils.inference

download_stacked_image_and_mask(bbox, grouped_elements, zoom, mapbox_token)

Download all tiles within a bounding box and stack them into a single image.

All the grouped_elements are painted on the mask.

Parameters:

Name Type Description Default
bbox tuple

Bounding box in the form of (south, west, north, east).

 required
grouped_elements dict

OpenStreetMap elements grouped with group_elements_by_tile.

 required
zoom int

Zoom level. See https://docs.mapbox.com/help/glossary/zoom-level/.

 required
mapbox_token str

Mapbox token. See https://docs.mapbox.com/help/getting-started/access-tokens/.

 required

Returns:

Name Type Description
tuple tuple[ndarray, ndarray]

Stacked image and mask.
Source code in src/osm_ai_helper/utils/inference.py 
 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
def download_stacked_image_and_mask(
    bbox: tuple[float, float, float, float],
    grouped_elements: dict,
    zoom: int,
    mapbox_token: str,
) -> tuple[np.ndarray, np.ndarray]:
    """Download all tiles within a bounding box and stack them into a single image.

    All the grouped_elements are painted on the mask.

    Args:
        bbox (tuple): Bounding box in the form of (south, west, north, east).
        grouped_elements (dict): OpenStreetMap elements grouped with
            [group_elements_by_tile][osm_ai_helper.utils.tiles.group_elements_by_tile].
        zoom (int): Zoom level.
            See https://docs.mapbox.com/help/glossary/zoom-level/.
        mapbox_token (str): Mapbox token.
            See https://docs.mapbox.com/help/getting-started/access-tokens/.

    Returns:
        tuple: Stacked image and mask.
    """
    south, west, north, east = bbox
    left, top = lat_lon_to_tile_col_row(north, west, zoom)
    right, bottom = lat_lon_to_tile_col_row(south, east, zoom)

    stacked_image = np.zeros(
        ((bottom - top) * TILE_SIZE, (right - left) * TILE_SIZE, 3), dtype=np.uint8
    )
    stacked_mask = np.zeros(
        ((bottom - top) * TILE_SIZE, (right - left) * TILE_SIZE), dtype=np.uint8
    )

    with ThreadPoolExecutor() as executor:
        futures = {}
        for n_col, tile_col in enumerate(range(left, right)):
            for n_row, tile_row in enumerate(range(top, bottom)):
                group = grouped_elements[(tile_col, tile_row)]
                future = executor.submit(
                    download_tile, zoom, tile_col, tile_row, mapbox_token
                )
                futures[(n_row, n_col)] = future

        for n_col, tile_col in enumerate(range(left, right)):
            for n_row, tile_row in enumerate(range(top, bottom)):
                group = grouped_elements[(tile_col, tile_row)]

                img = futures[(n_row, n_col)].result()

                mask = grouped_elements_to_mask(group, zoom, tile_col, tile_row)

                stacked_image[
                    n_row * TILE_SIZE : (n_row + 1) * TILE_SIZE,
                    n_col * TILE_SIZE : (n_col + 1) * TILE_SIZE,
                ] = np.array(img)

                stacked_mask[
                    n_row * TILE_SIZE : (n_row + 1) * TILE_SIZE,
                    n_col * TILE_SIZE : (n_col + 1) * TILE_SIZE,
                ] = mask

    return stacked_image, stacked_mask

split_area_into_lat_lon_centers(area_name, zoom, margin)

Split the bounding box of area_name into a list of lat lon centers.

If you iterate on the returned list using run_inference, you will cover the entire area.

Parameters:

Name Type Description Default
area_name str

Name of the area to split.

 required
zoom int

value to be used in run_inference.

 required
margin int

value to be used in run_inference.

 required

Returns:

Type Description
list[tuple[float, float]]

list[tuple[float, float]]: List of lat lon centers.
Source code in src/osm_ai_helper/utils/inference.py 
23
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
def split_area_into_lat_lon_centers(
    area_name: str, zoom: int, margin: int
) -> list[tuple[float, float]]:
    """
    Split the bounding box of `area_name` into a list of lat lon centers.

    If you iterate on the returned list using [run_inference][osm_ai_helper.run_inference.run_inference],
    you will cover the entire area.

    Args:
        area_name (str): Name of the area to split.
        zoom (int): value to be used in [run_inference][osm_ai_helper.run_inference.run_inference].
        margin (int): value to be used in [run_inference][osm_ai_helper.run_inference.run_inference].

    Returns:
        list[tuple[float, float]]: List of lat lon centers.
    """
    area = get_area(area_name)[0]
    south, north, west, east = area["boundingbox"]
    left, bottom = lat_lon_to_tile_col_row(float(south), float(west), zoom)
    right, top = lat_lon_to_tile_col_row(float(north), float(east), zoom)
    lat_lon_centers = []
    for col in range(left + margin, right + 1, (margin * 2) + 1):
        for row in range(top + margin, bottom + 1, (margin * 2) + 1):
            pixel_col_center = (col * 512) + 256
            pixel_row_center = (row * 512) + 256
            meters_col_center, meters_row_center = pixel_col_row_to_meters_col_row(
                pixel_col_center, pixel_row_center, zoom
            )
            lat_lon_center = meters_col_row_to_lat_lon(
                meters_col_center, meters_row_center
            )
            lat_lon_centers.append(lat_lon_center)
    return lat_lon_centers

tile_prediction(bbox_predictor, sam_predictor, image, overlap=0.125, bbox_conf=0.5, bbox_pad=0, batch_size=32)

Predict on a large image by splitting it into tiles.

Parameters:

Name Type Description Default
bbox_predictor YOLO

YOLO bounding box. See https://docs.ultralytics.com/tasks/detect/.

 required
sam_predictor SAM2ImagePredictor

Segment Anything Image Predictor. See https://github.com/facebookresearch/sam2?tab=readme-ov-file#image-prediction.

 required
image ndarray

Image to predict on.

 required
overlap float

Overlap between tiles. Defaults to 0.125.

 0.125
bbox_conf float

Sets the minimum confidence threshold for detections. Defaults to 0.4.

 0.5
bbox_pad int

Padding to be added to the predicted bbox. Defaults to 0.

 0
batch_size int

Batch size for prediction. Defaults to 32.

 32

Returns:

Type Description
ndarray

np.ndarray: Stacked output.
Source code in src/osm_ai_helper/utils/inference.py 
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
def tile_prediction(
    bbox_predictor: "YOLO",
    sam_predictor: "SAM2ImagePredictor",
    image: np.ndarray,
    overlap: float = 0.125,
    bbox_conf: float = 0.5,
    bbox_pad: int = 0,
    batch_size: int = 32,
) -> np.ndarray:
    """
    Predict on a large image by splitting it into tiles.

    Args:
        bbox_predictor (YOLO): YOLO bounding box.
            See https://docs.ultralytics.com/tasks/detect/.
        sam_predictor (SAM2ImagePredictor): Segment Anything Image Predictor.
            See https://github.com/facebookresearch/sam2?tab=readme-ov-file#image-prediction.
        image (np.ndarray): Image to predict on.
        overlap (float): Overlap between tiles.
            Defaults to 0.125.
        bbox_conf (float): Sets the minimum confidence threshold for detections.
            Defaults to 0.4.
        bbox_pad (int): Padding to be added to the predicted bbox.
            Defaults to 0.
        batch_size (int): Batch size for prediction.
            Defaults to 32.

    Returns:
        np.ndarray: Stacked output.
    """
    stacked_output = np.zeros((image.shape[0], image.shape[1]), dtype=np.uint8)
    tile_batch = []
    tile_coords_batch = []

    for top, left, bottom, right in yield_tile_corners(image, TILE_SIZE, overlap):
        logger.debug(f"Tile corners: {(top, left, bottom, right)}")
        tile_batch.append(image[left:right, top:bottom].copy())
        tile_coords_batch.append((top, left, bottom, right))

        if len(tile_batch) >= batch_size:
            stacked_output = predict_tile_batch(
                tile_batch,
                tile_coords_batch,
                bbox_predictor,
                sam_predictor,
                bbox_conf,
                bbox_pad,
                stacked_output,
            )
            tile_batch = []
            tile_coords_batch = []

    if tile_batch:
        stacked_output = predict_tile_batch(
            tile_batch,
            tile_coords_batch,
            bbox_predictor,
            sam_predictor,
            bbox_conf,
            bbox_pad,
            stacked_output,
        )

    return stacked_output

osm_ai_helper.utils.osm

get_area(area_name)

Get the area from Nominatim.

Uses the Nominatim API.

Parameters:

Name Type Description Default
area_name str

The name of the area.

 required

Returns:

Name Type Description
dict dict

The area found.
Source code in src/osm_ai_helper/utils/osm.py 
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
def get_area(area_name: str) -> dict:
    """Get the area from Nominatim.

    Uses the [Nominatim API](https://nominatim.org/release-docs/develop/api/Search/).

    Args:
        area_name (str): The name of the area.

    Returns:
        dict: The area found.
    """
    response = requests.get(
        f"https://nominatim.openstreetmap.org/search?q={area_name}&format=json",
        headers={"User-Agent": "Mozilla/5.0"},
    )
    response.raise_for_status()
    response_json = json.loads(response.content.decode())
    return response_json

get_area_id(area_name)

Get the Nominatim ID of an area.

Uses the Nominatim API.

Parameters:

Name Type Description Default
area_name str

The name of the area.

 required

Returns:

Name Type Description
int int

The Nominatim ID of the area.
Source code in src/osm_ai_helper/utils/osm.py 
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
def get_area_id(area_name: str) -> int:
    """
    Get the Nominatim ID of an area.

    Uses the [Nominatim API](https://nominatim.org/release-docs/develop/api/Search/).

    Args:
        area_name (str): The name of the area.

    Returns:
        int: The Nominatim ID of the area.
    """
    for area in get_area(area_name):
        osm_type = area.get("osm_type")
        osm_id = area.get("osm_id")
        if osm_type == "way":
            return osm_id + 2400000000
        if osm_type == "relation":
            return osm_id + 3600000000

get_elements(selector, area=None, bbox=None)

Get elements from OpenStreetMap using the Overpass API.

Uses the Overpass API.

Parameters:

Name Type Description Default
selector str

The selector to use. Example: "leisure=swimming_pool"

 required
area Optional[str]

The area to search in. Can be city, state, country, etc. Defaults to None.

 None
bbox Optional[Tuple[float, float, float, float]]

The bounding box to search in. Defaults to None. Format: https://wiki.openstreetmap.org/wiki/Overpass_API/Language_Guide#The_bounding_box

 None

Returns:

Type Description
list[dict]

The elements found.
Source code in src/osm_ai_helper/utils/osm.py 
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
def get_elements(
    selector: str,
    area: Optional[str] = None,
    bbox: Optional[Tuple[float, float, float, float]] = None,
) -> list[dict]:
    """
    Get elements from OpenStreetMap using the Overpass API.

    Uses the [Overpass API](https://wiki.openstreetmap.org/wiki/Overpass_API/Language_Guide).

    Args:
        selector (str): The selector to use.
            Example: "leisure=swimming_pool"
        area (Optional[str], optional): The area to search in.
            Can be city, state, country, etc.
            Defaults to None.

        bbox (Optional[Tuple[float, float, float, float]], optional): The bounding box to search in.
            Defaults to None.
            Format: https://wiki.openstreetmap.org/wiki/Overpass_API/Language_Guide#The_bounding_box

    Returns:
        The elements found.
    """
    query = "[out:json];"

    if area:
        area_id = get_area_id(area)
        query += f"area({area_id})->.searchArea;(way[{selector}](area.searchArea););"
    elif bbox:
        bbox_str = ",".join(map(str, bbox))
        query += f"(way[{selector}]({bbox_str}););"
    else:
        raise ValueError("area or bbox must be provided")

    query += " out body geom;"

    response = requests.get(
        "https://overpass-api.de/api/interpreter",
        params={"data": query},
        headers={"User-Agent": "Mozilla/5.0"},
    )
    response.raise_for_status()
    response_json = json.loads(response.content.decode())
    return response_json["elements"]

osm_ai_helper.utils.plots

show_vlm_entry(entry)

Extracts image and points from entry and draws the points.

Parameters:

Name Type Description Default
entry dict

Dataset entry generated by convert_to_vlm_dataset. Expected format:

entry = {
    "messages": [
        { "role": "user",
        "content" : [
            {"type" : "text",  "text"  : instruction},
            {"type" : "image", "image" : image} ]
        },
        { "role" : "assistant",
        "content" : [
            {"type" : "text",  "text"  : str(points)} ]
        },
    ]
}
required

Returns:

Name Type Description
Image Image

Image with points drawn.
Source code in src/osm_ai_helper/utils/plots.py 
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
def show_vlm_entry(entry) -> Image:
    """
    Extracts image and points from entry and draws the points.

    Args:
        entry (dict): Dataset entry generated by `convert_to_vlm_dataset`.
            Expected format:

            ```py
            entry = {
                "messages": [
                    { "role": "user",
                    "content" : [
                        {"type" : "text",  "text"  : instruction},
                        {"type" : "image", "image" : image} ]
                    },
                    { "role" : "assistant",
                    "content" : [
                        {"type" : "text",  "text"  : str(points)} ]
                    },
                ]
            }
            ```

    Returns:
        Image: Image with points drawn.
    """
    messages = entry["messages"]
    image = messages[0]["content"][1]["image"]
    width, height = image.size
    points = eval(messages[1]["content"][0]["text"])
    draw = ImageDraw.Draw(image)

    for point in points:
        draw.circle((point[0] * width, point[1] * height), 5, fill="red")

    return image

osm_ai_helper.utils.tiles

group_elements_by_tile(elements, zoom)

Broup elements by the tiles they belong to, based on the zoom level.

Each MAPBOX tile is a 512x512 pixel image.

Parameters:

Name Type Description Default
elements List[Dict]

List of elements from download_osm.

 required
zoom int

Zoom level. See https://docs.mapbox.com/help/glossary/zoom-level/.

 required

Returns:

Type Description
dict[tuple, list[dict]]

dict[tuple, list[dict]]: Grouped elements.
Source code in src/osm_ai_helper/utils/tiles.py 
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
def group_elements_by_tile(elements: List[Dict], zoom: int) -> dict[tuple, list[dict]]:
    """Broup elements by the tiles they belong to, based on the zoom level.

    Each MAPBOX tile is a 512x512 pixel image.

    Args:
        elements (List[Dict]): List of elements from
            [download_osm][osm_ai_helper.download_osm.download_osm].
        zoom (int): Zoom level. See https://docs.mapbox.com/help/glossary/zoom-level/.

    Returns:
        dict[tuple, list[dict]]: Grouped elements.
    """
    grouped: dict[tuple, list[dict]] = defaultdict(list)

    for element in elements:
        pixel_polygon = []
        for point in element["geometry"]:
            pixel_point = lat_lon_to_pixel_col_row(point["lat"], point["lon"], zoom)
            pixel_polygon.append(pixel_point)

        pixel_polygon = np.array(pixel_polygon, dtype=np.int32)

        tiles = map(tuple, np.unique(pixel_polygon // TILE_SIZE, axis=0))
        for group in tiles:
            grouped[group].append(element)

    return grouped