"""
Implements creation of geojsons for visualizations.
"""
import os
import json
import threading
import logging
import time
from typing import Dict, List, Union
from pyproj import Transformer
from flask import Blueprint, send_file, Response
from flask import current_app as app
from app.api import kpis as kpis_module
from app import utils
geojsons = Blueprint("geojsons", __name__)
logger = logging.getLogger(__name__)
ts_url = app.dotenv["TS_URL"]
storage_url = app.dotenv["STORAGE_URL"]
city = app.dotenv["URBANITE_CITY"]
data_dir = f"{app.dotenv['DATA_DIR']}/{app.dotenv['URBANITE_CITY']}"
# DEBUG
@app.route("/dss/geojson/test/<sim_id>")
def geojsontest(sim_id):
app.logger.debug("starting test")
kpi_name = "pedestrianTravelTime"
app.logger.debug("starting load network and events")
network, events = utils.get_network_and_events(sim_id)
app.logger.debug("done")
app.logger.debug("starting calculate pedestrian travel time")
start_t = time.time()
pedestrianTravelTime = events.pedestrian_travel_time()["detailed_long_trips"]
delta_t = time.time() - start_t
app.logger.debug("done in %s s", delta_t)
app.logger.debug("starting write geojson pedestrian trips")
start_t = time.time()
pedestrianTravelTime = write_geojson_pedestrian_trips(pedestrianTravelTime, network, sim_id)
delta_t = time.time() - start_t
app.logger.info("written pedestrianTravelTime in %s s", delta_t)
return pedestrianTravelTime
@app.route("/dss/geojson/<int:sim_id>/<kpi_name>")
def get_kpi_geojson(sim_id, kpi_name):
sim_dir = utils.get_sim_dir_names([sim_id])[0]
path = f"{data_dir}/simulations/{sim_id}/results/{sim_dir}/kpi_vis_{kpi_name}.json"
app.logger.warning("path to get kpi vis: %s", path)
# check if file exists
if os.path.exists(path) and os.stat(path).st_size != 0:
app.logger.info("returning file at %s", path)
return send_file(path)
else:
# does not exist, create it
app.logger.info("Starting to calculate geojson.")
calculate_geojson(sim_id)
return {"message": "Calculating..."}
@app.route("/dss/map-info", methods=["GET"])
def get_map_info():
lat, lon = app.dotenv["CITY_COORDINATES"].split(",")
zoom_level = app.dotenv["CITY_ZOOM_LEVEL"]
return {"coordinates": {"lat": lat, "lon": lon}, "zoom": zoom_level}
@app.route("/dss/kpi-vis-info", methods=["GET"])
def get_kpi_vis_info():
kpi_info = []
if city == "bilbao":
kpi_info = [
{"buttonValue": "pollution", "buttonLabel": "Pollution"},
{
"buttonValue": "pedestrianTravelTime",
"buttonLabel": "Pedestrian Travel Time",
},
{"buttonValue": "dailyInternalTravels", "buttonLabel": "Internal Trips"},
]
elif city == "amsterdam":
kpi_info = [
{"buttonValue": "pollution", "buttonLabel": "Pollution"},
{"buttonValue": "bikeability", "buttonLabel": "Bikeability"},
{"buttonValue": "bikeSafety", "buttonLabel": "Bike Traffic Safety"},
{"buttonValue": "bikeIntensity", "buttonLabel": "Bike Intensity"},
{"buttonValue": "bikeCongestion", "buttonLabel": "Bike Congestions"},
]
elif city == "helsinki":
kpi_info = [
{"buttonValue": "pollution", "buttonLabel": "Pollution"},
{
"buttonValue": "accousticPollution",
"buttonLabel": "Acoustic Pollution - WIP",
},
{
"buttonValue": "congestionsAndBottlenecks",
"buttonLabel": "Congestions and Bottlenecks",
},
{
"buttonValue": "harbourAreaTrafficFlow",
"buttonLabel": "Harbour Area Traffic Flow",
},
]
elif city == "messina":
kpi_info = [
{"buttonValue": "pollution", "buttonLabel": "Pollution"},
{
"buttonValue": "publicTransportUse",
"buttonLabel": "Public Transport Use",
},
{
"buttonValue": "averageSpeedOfPublicTransport",
"buttonLabel": "Avg Public Transport Speed",
},
# {"buttonValue": "numberOfBikeTrips", "buttonLabel": "Number of Bike Trips"},
{
"buttonValue": "shareOfVehicles",
"buttonLabel": "Share of Trips per Mode",
},
{
"buttonValue": "dailyInternalTravels",
"buttonLabel": "Daily Internal Trips",
}
]
else:
raise ValueError("Invalid deploy parameters - city set to %s", city)
return Response(json.dumps(kpi_info), mimetype="application/json")
def get_kpi_data_and_write(network, events, sim_id):
"""
get_kpi_data_final_geojson is a list of links with kpis calculated
"""
app.logger.info(f"getting kpi data to write, city {city}")
results = {}
emisson_object = kpis_module.get_emissions(sim_id, network)
rush_hour = utils.get_max_traffic_time(sim_id)
if network.city == "bilbao":
if emisson_object is not None:
pollution = emisson_object.total_emissions_by_link()
app.logger.info("Emission by link")
write_geojson_links(pollution, "pollution", network, sim_id)
app.logger.info("written emission")
pedestrianTravelTime = events.pedestrian_travel_time()["detailed_long_trips"]
app.logger.info("Pedestrian travel time")
write_geojson_pedestrian_trips(pedestrianTravelTime, network, sim_id)
app.logger.info("written pedestrianTravelTime")
dailyInternalTravels = events.vehicles_count_per_link()
app.logger.info("Vehicles count per link")
write_geojson_links(dailyInternalTravels, "dailyInternalTravels", network, sim_id)
app.logger.info("written dailyInternalTravels")
elif network.city == "helsinki":
if emisson_object is not None:
pollution = emisson_object.total_emissions_by_link()
app.logger.info("total emissions by link")
write_geojson_links(pollution, "pollution", network, sim_id)
app.logger.info("written emission")
# acousticPollution = None # TODO integrate with Inaki
# app.logger.info("accoustic pollution - TODO")
# write_geojson(pollution, "pollution", network, sim_id)
# app.logger.info("written emission")
congestionsAndBottlenecks = events.get_congested_links(rush_hour)["congested_links"]
app.logger.info("congestions and bottlenecks")
write_geojson_links(
congestionsAndBottlenecks, "congestionsAndBottlenecks", network, sim_id
)
app.logger.info("written congestionsAndBottlenecks")
results["harbourAreaTrafficFlow"] = None # TODO implement!
app.logger.info("harbour area traffic flow - TODO")
elif network.city == "messina":
if emisson_object is not None:
pollution = emisson_object.total_emissions_by_link()
app.logger.info("Emission by link")
write_geojson_links(pollution, "pollution", network, sim_id)
app.logger.info("written emission")
publicTransportUse = events.public_transport_use_geojson()
app.logger.info("public transport use")
write_geojson_links(publicTransportUse, "publicTransportUse", network, sim_id)
app.logger.info("written public transport use")
averageSpeedOfPublicTransport = events.average_bus_speed_geojson()
app.logger.info("average speed of public transport")
write_geojson_links(
averageSpeedOfPublicTransport,
"averageSpeedOfPublicTransport",
network,
sim_id,
)
app.logger.info("written averageSpeedOfPublicTransport")
numberOfBikeTrips = events.vehicles_count_per_link(only="bike")
app.logger.info("number of bike trips")
write_geojson_links(numberOfBikeTrips, "numberOfBikeTrips", network, sim_id)
app.logger.info("written numberOfBikeTrips")
shareOfVehicles = vehicleCountToShareGeojson(events.vehicles_count_per_link())
app.logger.info("share of vehicles")
write_geojson_links(shareOfVehicles, "shareOfVehicles", network, sim_id)
app.logger.info("written shareOfVehicles")
dailyInternalTravels = events.vehicles_count_per_link()
app.logger.info("daily internal travels")
write_geojson_links(dailyInternalTravels, "dailyInternalTravels", network, sim_id)
app.logger.info("written dailyInternalTravels")
elif network.city == "amsterdam":
if emisson_object is not None:
pollution = emisson_object.total_emissions_by_link()
app.logger.info("got emissions by link")
write_geojson_links(pollution, "pollution", network, sim_id)
app.logger.info("written pollution")
bikeability = events.bikeability_index()
app.logger.info("bikeability")
write_geojson_links(bikeability, "bikeability", network, sim_id)
app.logger.info("written bikeability")
bike_safety = events.bike_safety_index()
app.logger.info("bike safety")
write_geojson_links(bike_safety, "bikeSafety", network, sim_id)
app.logger.info("written bikeSafety")
bike_intensity = events.vehicles_count_per_link()
app.logger.info("bikeIntensity")
write_geojson_links(bike_intensity, "bikeIntensity", network, sim_id)
app.logger.info("written bikeIntensity")
bike_congestion = events.get_congested_links(rush_hour, vehicle_mode="bicycle")["congested_links"]
app.logger.info("bikeCongestion")
write_geojson_links(bike_congestion, "bikeCongestion", network, sim_id)
app.logger.info("written bikeCongestion")
app.logger.info("done")
def vehicleCountToShareGeojson(data_per_link):
results = {}
sum_vehicles_per_link = {}
for link, data in data_per_link.items():
n_veh = 0
for key in data:
n_veh += data[key]
sum_vehicles_per_link[link] = sum_vehicles_per_link.get(link, 0) + n_veh
for link, data in data_per_link.items():
if link.startswith("pt"):
continue
n_veh = sum_vehicles_per_link[link]
if n_veh < 1:
continue
results[link] = {
"bike": data["bike_count"] / n_veh,
"car": data["car_count"] / n_veh,
"bus": data["bus_count"] / n_veh,
}
return results
def write_geojson_pedestrian_trips(kpi_data, network, sim_id):
"""
Write a pedestrian kpi_data object to a file.
kpi_data should be a list [{"start_link": _, "end_link": _, "average_time": _ } ]
"""
data = {"type": "FeatureCollection", "features": []}
crs_transformer = Transformer.from_crs(network.crs_epsg, "epsg:4326")
# crs_transformer = Transformer.from_crs("epsg:2062", "epsg:4326")
for element in kpi_data:
if element["average_time"] < 3600.: continue
s_link = network.get_link(element["start_link"])
s_nodes = [network.get_node(s_link["from"]), network.get_node(s_link["to"]) ]
s_point = (
(float(s_nodes[0]["x"]) + float(s_nodes[1]["x"])) / 2.,
(float(s_nodes[0]["y"]) + float(s_nodes[1]["y"])) / 2.
)
s_point = tuple(reversed(crs_transformer.transform(*s_point)))
e_link = network.get_link(element["end_link"])
e_nodes = [network.get_node(e_link["from"]), network.get_node(e_link["to"])]
e_point = (
(float(e_nodes[0]["x"]) + float(e_nodes[1]["x"])) / 2.,
(float(e_nodes[0]["y"]) + float(e_nodes[1]["y"])) / 2.
)
e_point = tuple(reversed(crs_transformer.transform(*e_point)))
data["features"].append(
{
"type": "Feature",
"properties": {
"long pedestrian trip": element["average_time"] / (60)
},
"geometry": {
"type": "LineString",
"coordinates": [s_point, e_point]
}
}
)
filename = get_filename(sim_id, "pedestrianTravelTime")
with open(filename, "w") as fp:
json.dump(data, fp, indent=2)
return data
def write_geojson_links(kpi_data, kpi_name, network, sim_id):
"""
Write a per link kpi_data object to a file.
kpi_data should be a dict { "link_id": {kpi1, kpi2, ...}, ...}
"""
# logger.debug("KPI DATA to be written: \n%s", kpi_data)
data = {"type": "FeatureCollection", "features": []}
crs_transformer = Transformer.from_crs(network.crs_epsg, "epsg:4326")
logger.warning("network city: %s, network epsg: %s", network.city, network.crs_epsg)
for link_id, link_data in kpi_data.items():
if link_id.startswith("pt"):
continue
# if isinstance(link_data, float):
# link_data = {kpi_name: link_data}
elif len(link_data) < 1:
continue
link = network.get_link(link_id)
from_coords_orig = (
float(network.get_node(link["from"])["x"]),
float(network.get_node(link["from"])["y"]),
)
from_coords = tuple(reversed(crs_transformer.transform(*from_coords_orig)))
to_coords_orig = (
float(network.get_node(link["to"])["x"]),
float(network.get_node(link["to"])["y"]),
)
to_coords = tuple(reversed(crs_transformer.transform(*to_coords_orig)))
data["features"].append(
{
"type": "Feature",
"properties": {},
"geometry": {
"type": "LineString",
"coordinates": [from_coords, to_coords],
},
}
)
data["features"][-1]["properties"] = link_data
filename = get_filename(sim_id, kpi_name)
with open(filename, "w") as fp:
json.dump(data, fp, indent=2)
def get_filename(sim_id, kpi_name):
"""
Returns the filename for the simulation and kpi provided.
"""
date = utils.get_sim_dir_names([sim_id])[0]
return f"{data_dir}/simulations/{sim_id}/results/{date}/kpi_vis_{kpi_name}.json"
def get_kpi_names(
kpis_obj: Union[Dict, float], kpi_names=[], last_kpi=None
) -> List[str]:
"""
Returns a list of kpi names from the kpis_obj.
"""
for key, value in kpis_obj.items():
if isinstance(value, Dict):
kpi_names += get_kpi_names(value, kpi_names, key)
else:
kpi_names.append(key)
return kpi_names
def write_final_geojson(simulation_id, app_con=None):
"""
Calculates the kpi_vis geojson for visualization of KPIs.
:param sim_id: simulation id
"""
if app_con is not None:
app = app_con.app
else:
from flask import current_app as app
app.logger.info("write final geojson call")
# Check if thread is running
thread_running = False
for thread in threading.enumerate():
if "get_kpi_data_and_write" in thread.name:
thread_running = True
logger.info("calculate kpi_vis thread already running")
if not thread_running:
with app.app_context():
app.logger.info("network, events loading")
network, events = utils.get_network_and_events(simulation_id)
app.logger.info("network, events loaded")
with app.app_context():
get_kpi_data_and_write_thread = threading.Thread(
target=get_kpi_data_and_write(network, events, simulation_id)
)
if thread_running:
return {"message": "still preparing..."}
else:
return {"message": "preparing visualizations."}
def calculate_geojson(sim_id):
"""
Calculates the geojson for visualization of KPI values on the streets.
:param sim_id: simulation id
:returns: status
"""
app.logger.info("starting a calculate kpi_vis thread...")
# Check if thread is running
thread_running = False
for thread in threading.enumerate():
if "write_final_geojson" in thread.name:
thread_running = True
logger.info("calculate kpi_vis thread already running")
break
# If thread is not running than start it
if not thread_running:
app_con = app.app_context()
create_kpi_vis_thread = threading.Thread(
target=write_final_geojson, args=[sim_id, app_con]
)
create_kpi_vis_thread.start()
logger.info("started a calculate kpi_vis thread.")
else:
return {"message": "Still preparing the KPI visualization."}
return {"message": "Preparing the KPI visualization."}