-
Notifications
You must be signed in to change notification settings - Fork 4
/
corktown_offline_indicators.py
188 lines (149 loc) · 7.45 KB
/
corktown_offline_indicators.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
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
#!/usr/bin/env python3
# -*- coding: utf-8 -*-
"""
Created on Fri Jun 12 11:21:39 2020
@author: doorleyr
"""
from proximity_indicator import ProxIndicator
from innovation_indicator import InnoIndicator
from economic_indicator import EconomicIndicator
from buildings_indicator import BuildingsIndicator
from diversity_indicator import DiversityIndicator
import json
import pandas as pd
import urllib
table_name='corktown'
host='https://cityio.media.mit.edu/'
cityIO_get_url=host+'api/table/'+table_name
# =============================================================================
# Initialise Indicators
# =============================================================================
I = InnoIndicator()
P = ProxIndicator(name='proximity', indicator_type_in='numeric', table_name='corktown')
B= BuildingsIndicator(name='buildings', table_name='corktown')
D= DiversityIndicator(name='diversity', table_name='corktown')
E = EconomicIndicator(name='Economic',
table_name='corktown')
# =============================================================================
# Load contextual data
# =============================================================================
with urllib.request.urlopen(cityIO_get_url+'/GEOGRID') as url:
geogrid=json.loads(url.read().decode())
cell_area=geogrid['properties']['header']['cellSize']**2
updatable=[((feat['properties']['interactive'])or (feat['properties']['static_new'])
) for feat in geogrid['features']]
types=geogrid['properties']['types']
static_types=geogrid['properties']['static_types']
types.update(static_types)
reporting_types=[t for t in types if t not in ['None','Residential Low Density', 'Industrial']]
for ind in [I, P,
# P_hm,
B, D, E]:
ind.types_def=types
ind.geogrid_header=geogrid['properties']['header']
# =============================================================================
# Load the saved land use scenarios from city_IO
# =============================================================================
# Scenarios are saved as scenrio0, scenario1 etc. on cityIO
# scenario name is contained in the 'info' field
# try 0 to N and save all the results
# Additional scenarios can be created and saved to cityIO using the CityScope interactive front-end
all_scenarios={}
for i in range(10):
try:
with urllib.request.urlopen(cityIO_get_url+'/scenarios'+str(i)) as url:
geogriddata=json.loads(url.read().decode())
all_scenarios[geogriddata['info']['name']]=geogriddata['GEOGRIDDATA']
except:
pass
print('Downloaded {} land use scenarios'.format(len(all_scenarios)))
# =============================================================================
# Define some functions to calculate indicators and land use stats for each scenario
# =============================================================================
def get_type_stats(geogrid_data, reporting_types, updatable,cell_area, types_def=types):
"""
calculates the square meters and capacity of each LU type added to the interactive area
"""
results={}
for type_name in reporting_types:
sqm_pp=types_def[type_name]['sqm_pperson']
floors=0
for gi, cell in enumerate(geogrid_data):
if ((updatable[gi]) and (cell['name']==type_name)):
height=cell['height']
if isinstance(height, list):
height=height[-1]
floors+=height
results[type_name]= {'sqm_pp': sqm_pp, 'sqm': floors*cell_area,
'capacity':floors*cell_area/sqm_pp }
return results
def get_all_indicators(geogrid_data):
"""
calculates values of all the individual indicators for a given scenario
"""
all_ind=[]
print('Innovation')
all_ind.extend(I.return_indicator(geogrid_data))
print('Economic')
all_ind.extend(E.return_indicator(geogrid_data))
print('Proximity')
all_ind.extend(P.return_indicator(geogrid_data))
print('Diversity')
all_ind.extend(D.return_indicator(geogrid_data))
print('Buildings')
all_ind.extend(B.return_indicator(geogrid_data))
return all_ind
def create_scenario_row(all_ind, stats, scenario_name):
"""
takes the indicators results, square meters and capacities for a given scenario
and creates a row of data for the output csv
"""
all_cols={'Scenario': scenario_name}
for ind in all_ind:
all_cols[ind['name']+' norm']=ind['value']
raw_name=ind['name']+' raw'
if not ind['units']==None:
raw_name+= ' ['+ind['units']+']'
all_cols[raw_name]=ind['raw_value']
return all_cols
# =============================================================================
# Calculate indicators and land use stats for each scenario
# =============================================================================
base_indicators=get_all_indicators(all_scenarios['Baseline'])
base_stats=get_type_stats(all_scenarios['Baseline'], reporting_types, updatable,cell_area)
campus_indicators=get_all_indicators(all_scenarios['Campus_Only'])
campus_stats=get_type_stats(all_scenarios['Campus_Only'], reporting_types, updatable,cell_area)
campus_mobility_indicators=get_all_indicators(all_scenarios['Campus_Only'])
campus_mobility_stats=campus_stats
housing_indicators=get_all_indicators(all_scenarios['Campus_Housing'])
housing_stats=get_type_stats(all_scenarios['Campus_Housing'], reporting_types, updatable,cell_area)
inno_com_indicators=get_all_indicators(all_scenarios['Innovation_Community'])
inno_com_stats=get_type_stats(all_scenarios['Innovation_Community'], reporting_types, updatable,cell_area)
all_scenarios=[]
all_scenarios.append(create_scenario_row(base_indicators, base_stats, scenario_name='BAU'))
all_scenarios.append(create_scenario_row(campus_indicators, campus_stats, scenario_name='Campus Only'))
all_scenarios.append(create_scenario_row(campus_mobility_indicators, campus_mobility_stats, scenario_name='Future Mobility'))
all_scenarios.append(create_scenario_row(housing_indicators, housing_stats, scenario_name='Housing'))
all_scenarios.append(create_scenario_row(inno_com_indicators, inno_com_stats, scenario_name='Innovation Community'))
output=pd.DataFrame(all_scenarios)
# =============================================================================
# Crea and calculate aggregated indicators
# =============================================================================
aggregation={
'Innovation Potential': ['Knowledge','Skills','R&D Funding'],
'Economic Performance': ['Average Salary','Productivity','Employment Density', 'Diversity Jobs'],
'Sustainable Buildings': ['Buildings Energy Performance'],
'Community Benefits': ['Access to housing', 'Access to education', 'Access to 3rd Places',
'Access to parks', 'Access to employment', 'Diversity Jobs',
'Diversity Third Places', 'Diversity Education']}
for comp_ind in aggregation:
cols=[ind_name +' norm' for ind_name in aggregation[comp_ind]]
output[comp_ind]=output[cols].mean(axis=1)
# =============================================================================
# Create and save output as csv file
# =============================================================================
col_order = ['Scenario']+[col for col in all_scenarios[0] if 'norm' in col]+[
col for col in all_scenarios[0] if 'raw' in col]+ [
comp_ind for comp_ind in aggregation]
output=output[col_order]
output.to_csv('scenario_outputs.csv')