Skip to content

API Docs

Main Functions

Parse a GeoJSON file or data struc into a normalized FeatureCollection.

Parameters:

Name Type Description Default
db str | Connection

Existing db connection, or connection string.

required
geojson_raw str | bytes | dict

GeoJSON file path, JSON string, dict, or file bytes.

required
merge bool

If any nested Polygons / MultiPolygon should be merged.

False

Returns:

Name Type Description
FeatureCollection FeatureCollection

a FeatureCollection.

Source code in geojson_aoi/parser.py
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
def parse_aoi(
    db: str | Connection, geojson_raw: str | bytes | dict, merge: bool = False
) -> FeatureCollection:
    """Parse a GeoJSON file or data struc into a normalized FeatureCollection.

    Args:
        db (str | Connection): Existing db connection, or connection string.
        geojson_raw (str | bytes | dict): GeoJSON file path, JSON string, dict,
            or file bytes.
        merge (bool): If any nested Polygons / MultiPolygon should be merged.

    Returns:
        FeatureCollection: a FeatureCollection.
    """
    # We want to maintain this list for input control.
    valid_geoms = ["Polygon", "MultiPolygon", "GeometryCollection"]

    # Parse different input types
    if isinstance(geojson_raw, bytes):
        geojson_parsed = json.loads(geojson_raw)

    if isinstance(geojson_raw, str):
        if Path(geojson_raw).exists():
            log.debug(f"Parsing geojson file: {geojson_raw}")
            with open(geojson_raw, "rb") as geojson_file:
                geojson_parsed = json.load(geojson_file)
        else:
            geojson_parsed = json.loads(geojson_raw)

    elif isinstance(geojson_raw, dict):
        geojson_parsed = geojson_raw
    else:
        raise ValueError("GeoJSON input must be a valid dict, str, or bytes")

    # Throw error if no data
    if geojson_parsed is None or geojson_parsed == {} or "type" not in geojson_parsed:
        raise ValueError("Provided GeoJSON is empty")

    # Throw error if wrong geometry type
    if geojson_parsed["type"] not in AllowedInputTypes:
        raise ValueError(f"The GeoJSON type must be one of: {AllowedInputTypes}")

    # Store properties in formats that contain them.
    properties = []
    if (
        geojson_parsed.get("type") == "Feature"
        and geojson_parsed.get("geometry")["type"] in valid_geoms
    ):
        properties.append(geojson_parsed.get("properties"))

    elif geojson_parsed.get("type") == "FeatureCollection":
        for feature in geojson_parsed.get("features"):
            if feature["geometry"]["type"] in valid_geoms:
                properties.append(feature["properties"])

    # Extract from FeatureCollection
    geoms = strip_featcol(geojson_parsed)

    # Strip away any geom type that isn't a Polygon
    geoms = [geom for geom in geoms if geom["type"] == "Polygon"]

    with PostGis(db, geoms, merge) as result:
        # Remove any properties that PostGIS might have assigned.
        for feature in result.featcol["features"]:
            feature.pop("properties", None)

        # Restore saved properties.
        if properties:
            feat_count = 0
            for feature in result.featcol["features"]:
                feature["properties"] = properties[feat_count]
                feat_count = feat_count + 1

        return result.featcol

options: show_source: false heading_level: 3

Helper Functions

Warn the user if an invalid CRS is detected.

Also does a rough check for one geometry, to determine if the coordinates are range 90/180 degree range.

Parameters:

Name Type Description Default
geojson GeoJSON

a GeoJSON.

required

Returns:

Type Description
None

None

Source code in geojson_aoi/parser.py
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
def check_crs(geojson: GeoJSON) -> None:
    """Warn the user if an invalid CRS is detected.

    Also does a rough check for one geometry, to determine if the
    coordinates are range 90/180 degree range.

    Args:
        geojson (GeoJSON): a GeoJSON.

    Returns:
        None
    """

    def is_valid_crs(crs_name: str) -> bool:
        valid_crs_list = [
            "urn:ogc:def:crs:OGC:1.3:CRS84",
            "urn:ogc:def:crs:EPSG::4326",
            "WGS 84",
        ]
        return crs_name in valid_crs_list

    def is_valid_coordinate(coord: list[float]) -> bool:
        return len(coord) == 2 and -180 <= coord[0] <= 180 and -90 <= coord[1] <= 90

    crs = geojson.get("crs", {}).get("properties", {}).get("name")
    if crs and not is_valid_crs(crs):
        warning_msg = (
            "Unsupported coordinate system. Use WGS84 (EPSG 4326) for best results."
        )
        log.warning(warning_msg)
        warnings.warn(UserWarning(warning_msg), stacklevel=2)

    geom = geojson.get("geometry") or geojson.get("features", [{}])[-1].get(
        "geometry", {}
    )
    coordinates = geom.get("coordinates", [])

    # Drill down into nested coordinates to find the first coordinate
    while isinstance(coordinates, list) and len(coordinates) > 0:
        coordinates = coordinates[0]

    if not is_valid_coordinate(coordinates):
        warning_msg = "Invalid coordinates in GeoJSON. Ensure the file is not empty."
        log.warning(warning_msg)
        warnings.warn(UserWarning(warning_msg), stacklevel=2)

options: show_source: false heading_level: 3

Remove FeatureCollection and Feature wrapping.

Parameters:

Name Type Description Default
geojson_obj dict

a parsed geojson.

required

Returns:

Type Description
list[GeoJSON]

list[GeoJSON]: a list of geometries.

Source code in geojson_aoi/parser.py
 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
def strip_featcol(geojson_obj: GeoJSON | Feature | FeatureCollection) -> list[GeoJSON]:
    """Remove FeatureCollection and Feature wrapping.

    Args:
        geojson_obj (dict): a parsed geojson.

    Returns:
        list[GeoJSON]: a list of geometries.
    """
    if geojson_obj.get("crs"):
        # Warn the user if invalid CRS detected
        check_crs(geojson_obj)

    geojson_type = geojson_obj.get("type")

    if geojson_type == "FeatureCollection":
        geoms = [feature["geometry"] for feature in geojson_obj.get("features", [])]

        # Drill in and check if each feature is a GeometryCollection.
        # If so, our work isn't done.
        temp_geoms = []
        for geom in geoms:
            if geom["type"] == "GeometryCollection":
                for item in geom["geometries"]:
                    temp_geoms.append(item)

                geoms = temp_geoms

    elif geojson_type == "Feature":
        geoms = [geojson_obj.get("geometry")]

    elif geojson_type == "GeometryCollection":
        geoms = geojson_obj.get("geometries")

    elif geojson_type == "MultiPolygon":
        # MultiPolygon should parse out into List of Polygons and maintain properties.
        temp_geoms = []
        for coordinate in geojson_obj.get("coordinates"):
            # Build a Polygon from scratch out of the coordinates.
            polygon = {"type": "Polygon", "coordinates": coordinate}
            temp_geoms.append(polygon)

        geoms = temp_geoms

    else:
        geoms = [geojson_obj]

    return geoms

options: show_source: false heading_level: 3