forked from cyterat/deepstate-map-data
-
Notifications
You must be signed in to change notification settings - Fork 0
/
Copy pathscript.py
148 lines (113 loc) · 4.79 KB
/
script.py
1
2
3
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
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
#!/usr/bin/env python3
import os
import sys
import logging
from datetime import datetime
import time
import requests
import geopandas as gpd
from shapely.geometry import shape
from shapely.geometry import Polygon
from shapely.geometry import Point
from shapely import wkt
from shapely.geometry import JOIN_STYLE
import matplotlib.pyplot as plt
# Set up logging
logging.basicConfig(level=logging.INFO, format='%(asctime)s - %(levelname)s - %(message)s')
logger = logging.getLogger(__name__)
# Configuration
API_URL = "https://deepstatemap.live/api/history/last"
OUTPUT_DIR = "data"
OUTPUT_FILENAME = f"deepstatemap_data_{datetime.now().strftime('%Y%m%d')}.geojson"
MAX_RETRIES = 3
RETRY_DELAY = 5 # seconds
EXPORT_RAW = True
def make_api_request():
"""Make a request to the API and return the JSON response."""
headers = {
"User-Agent": "Mozilla/5.0 (Windows Phone 10.0; Android 6.0.1; Microsoft; RM-1152) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/52.0.2743.116 Mobile Safari/537.36 Edge/15.15254"
}
for attempt in range(MAX_RETRIES):
try:
response = requests.get(API_URL, headers=headers, timeout=10)
response.raise_for_status()
return response.json()
except requests.exceptions.RequestException as e:
logger.warning(f"API request failed (attempt {attempt + 1}/{MAX_RETRIES}): {e}")
if attempt < MAX_RETRIES - 1:
logger.info(f"Retrying in {RETRY_DELAY} seconds...")
time.sleep(RETRY_DELAY)
else:
logger.error("All API request attempts failed.")
sys.exit(1)
def process_data(data):
"""Process the API response data."""
geo_list = []
for f in data['map']['features']:
geom = f['geometry']
name = f['properties']['name']
new_feature = {
"name":name,
"geometry":wkt.loads(wkt.dumps(shape(geom), output_dimension=2))
}
geo_list.append(new_feature)
# Split the name by '///' and take the first part
def extract_first_part(name, part=0):
first_part = name.split('///')[part].strip()
return first_part
for item in geo_list:
item['name'] = extract_first_part(item['name'], part=1)
return geo_list
def create_geodataframe(geo_list):
"""Create a GeoDataFrame from the processed data."""
raw_gdf = gpd.GeoDataFrame(geo_list).set_crs(4326)
mask = raw_gdf.geometry.apply(lambda x: isinstance(x, Polygon))
polygon_gdf = raw_gdf[mask]
filtered_gdf = polygon_gdf[polygon_gdf['name'].isin(['CADR and CALR', 'Occupied', 'Occupied Crimea'])].reset_index()
merged_gdf = gpd.GeoSeries(filtered_gdf.union_all(), crs=4326)
# Applying buffer to remove union artifacts
eps = 0.000009
deartifacted_gdf = (
merged_gdf
.buffer(eps, 1, join_style=JOIN_STYLE.mitre)
.buffer(-eps, 1, join_style=JOIN_STYLE.mitre)
)
return deartifacted_gdf
def main():
# Ensure output directory exists
os.makedirs(OUTPUT_DIR, exist_ok=True)
# Make API request
logger.info("Making API request...")
raw_data = make_api_request()
# Process data
logger.info("Processing data...")
processed_data = process_data(raw_data)
if EXPORT_RAW:
# Create points GeoDataFrame
logger.info("Creating points GeoDataFrame")
raw_gdf = gpd.GeoDataFrame(processed_data).set_crs(4326)
mask = raw_gdf.geometry.apply(lambda x: isinstance(x, Point))
points_gdf = raw_gdf[mask]
# Create polygons GeoDataFrame
logger.info("Creating polygons GeoDataFrame")
mask = raw_gdf.geometry.apply(lambda x: isinstance(x, Polygon))
polygons_gdf = raw_gdf[mask]
# Export as GeoJSON
output_path = os.path.join(OUTPUT_DIR, f"{OUTPUT_FILENAME.split('.')[0]}_points.{OUTPUT_FILENAME.split('.')[1]}")
logger.info(f"Exporting data to {output_path}...")
points_gdf.to_file(output_path, driver="GeoJSON")
output_path = os.path.join(OUTPUT_DIR, f"{OUTPUT_FILENAME.split('.')[0]}_polygons.{OUTPUT_FILENAME.split('.')[1]}")
logger.info(f"Exporting data to {output_path}...")
polygons_gdf.to_file(output_path, driver="GeoJSON")
logger.info("Data update completed successfully.")
else:
# Create GeoDataFrame
logger.info("Creating GeoDataFrame...")
gdf = create_geodataframe(processed_data)
# Export as GeoJSON
output_path = os.path.join(OUTPUT_DIR, OUTPUT_FILENAME)
logger.info(f"Exporting data to {output_path}...")
gdf.to_file(output_path, driver="GeoJSON")
logger.info("Data update completed successfully.")
if __name__ == "__main__":
main()