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
|
|
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
|
|
required
|
Returns:
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
|
|
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