dhs-st commited on
Commit
2cbfbdf
·
verified ·
1 Parent(s): e2592df

Create app.py

Browse files
Files changed (1) hide show
  1. app.py +139 -0
app.py ADDED
@@ -0,0 +1,139 @@
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
1
+ import os
2
+ import osmnx as ox
3
+ import geopandas as gpd
4
+ import gradio as gr
5
+ from datetime import datetime
6
+ import shutil
7
+ import re
8
+ import pandas as pd
9
+
10
+ def abbreviate_address(address):
11
+ abbreviations = {
12
+ "Street": "St", "Avenue": "Ave", "Boulevard": "Blvd", "Court": "Ct",
13
+ "Drive": "Dr", "Road": "Rd", "Lane": "Ln", "Circle": "Cir",
14
+ "Parkway": "Pkwy", "Highway": "Hwy", "Square": "Sq", "Terrace": "Ter",
15
+ "Place": "Pl", "North": "N", "South": "S", "East": "E", "West": "W",
16
+ "Northeast": "NE", "Northwest": "NW", "Southeast": "SE", "Southwest": "SW",
17
+ "First": "1st", "Second": "2nd", "Third": "3rd", "Fourth": "4th", "Fifth": "5th",
18
+ "Sixth":"6th", "Seventh":"7th", "Eighth":"8th", "Nineth":"9th",
19
+ "Expressway":"Expy", "Freeway":"Fwy"
20
+ }
21
+
22
+ for term, abbr in abbreviations.items():
23
+ if type(address) == list:
24
+ address = address[0]
25
+ address = re.sub(r'\b' + re.escape(term) + r'\b', abbr, address, flags=re.IGNORECASE)
26
+ else:
27
+ address = re.sub(r'\b' + re.escape(term) + r'\b', abbr, address, flags=re.IGNORECASE)
28
+
29
+ return address
30
+
31
+ def clean_street_name(name):
32
+ if isinstance(name, list):
33
+ name = name[0]
34
+ # Remove special characters and normalize spaces
35
+ cleaned = re.sub(r'[^\w\s-]', '', str(name))
36
+ return ' '.join(cleaned.split())
37
+
38
+ def download_road_shapefile(coordinates_upper_left, coordinates_bottom_right):
39
+ north, west = coordinates_upper_left
40
+ south, east = coordinates_bottom_right
41
+
42
+ current_time = datetime.now().strftime('%Y%m%d_%H%M%S')
43
+ output_folder = f'road_network_{current_time}'
44
+
45
+ G = ox.graph_from_bbox(north, south, east, west, network_type='all')
46
+ gdf_edges = ox.graph_to_gdfs(G, nodes=False, edges=True)
47
+
48
+ # Process road names and numbers
49
+ def process_road_name(row):
50
+ # Safely get values, converting NaN to empty string
51
+ #print(row['name'])
52
+ name = row.get('name', '')
53
+ ref = row.get('ref', '')
54
+ highway = row.get('highway', '')
55
+
56
+ # Handle lists
57
+ if isinstance(name, list): name = name[0] if name else ''
58
+ if isinstance(ref, list): ref = ref[0] if ref else ''
59
+ if isinstance(highway, list): highway = highway[0] if highway else ''
60
+
61
+ # Process interstate highways
62
+ if highway == 'motorway' and ref:
63
+ ref = ref.split(';')
64
+ try:
65
+ ref = ref[0]
66
+ except:
67
+ #print('here', ref[0])
68
+ ref = ref[0]
69
+
70
+ if ref.startswith('I'):
71
+ #print(ref)
72
+ #print(f"I-{ref.replace('I', '').strip()}")
73
+ return f"I-{ref.replace('I', '').strip()}"
74
+ return ref
75
+
76
+ # Combine ref and name for other roads
77
+ #print(name, ref, highway)
78
+ if pd.notnull(ref) and name:
79
+ if ref not in str(name):
80
+ print(f"{ref} {name}".strip())
81
+ return f"{ref}/{name}".strip()
82
+
83
+ return name if name else ref if ref else ''
84
+
85
+ gdf_edges = gdf_edges[gdf_edges['name'].notnull()]
86
+ # Apply the processing to create a combined name field
87
+ gdf_edges['display_name'] = gdf_edges.apply(process_road_name, axis = 1)
88
+
89
+ # Clean and abbreviate the display names
90
+ gdf_edges['display_name'] = gdf_edges['display_name'].apply(abbreviate_address)
91
+ gdf_edges['display_name'] = gdf_edges['display_name'].apply(clean_street_name)
92
+
93
+ # Remove unwanted entries
94
+ gdf_edges = gdf_edges[~gdf_edges['display_name'].str.contains('Cycletrack|Cycleway', case=False, na=False)]
95
+ gdf_edges = gdf_edges[gdf_edges['display_name'].str.len() > 0]
96
+
97
+ # Select final columns
98
+ gdf_edges = gdf_edges[['display_name', 'geometry']]
99
+ gdf_edges = gdf_edges.to_crs(epsg=4326)
100
+
101
+ # Save the file
102
+ os.makedirs(output_folder, exist_ok=True)
103
+ shapefile_path = os.path.join(output_folder, f'road_network_{current_time}.shp')
104
+ gdf_edges.to_file(shapefile_path, driver='ESRI Shapefile', encoding='utf-8')
105
+
106
+ zip_filename = f'{output_folder}.zip'
107
+ shutil.make_archive(output_folder, 'zip', output_folder)
108
+
109
+ return zip_filename
110
+
111
+ def process_coordinates(upper_left, bottom_right):
112
+ try:
113
+ # Parse coordinates
114
+ north, west = map(float, upper_left.split(','))
115
+ south, east = map(float, bottom_right.split(','))
116
+
117
+ # Validate coordinates logic (north must be greater than south and west must be less than east)
118
+ if north <= south or west >= east:
119
+ return "Invalid bounding box: Upper left corner must be north-west of the lower right corner."
120
+
121
+ zip_file = download_road_shapefile((north, west), (south, east))
122
+ return zip_file
123
+ except Exception as e:
124
+ return str(e)
125
+
126
+ # Gradio Interface
127
+ iface = gr.Interface(
128
+ fn=process_coordinates,
129
+ inputs=[
130
+ gr.Textbox(label="Upper Left Coordinates (lat, lon)", placeholder="e.g., 37.7749,-122.4194"),
131
+ gr.Textbox(label="Bottom Right Coordinates (lat, lon)", placeholder="e.g., 37.7740,-122.4180")
132
+ ],
133
+ outputs="file",
134
+ title="Download Road Network Shapefile",
135
+ description="Enter the coordinates to download a shapefile of the road network."
136
+ )
137
+
138
+ if __name__ == "__main__":
139
+ iface.launch()