Newer
Older
# imports
# standard
from ast import literal_eval
import random
# local, external
from geopandas import GeoDataFrame
from pandas import concat, MultiIndex, Series
import networkx as nx
import osmnx as ox
from shapely.geometry import Point, LineString
# local, internal
import src.topupopt.data.gis.identify as gis_iden
import src.topupopt.data.gis.utils as gis_utils
import src.topupopt.data.gis.osm as _osm
ox.settings.use_cache = True
Pedro L. Magalhães
committed
# *****************************************************************************
# *****************************************************************************
class TestGisUtils:
# *************************************************************************
# *************************************************************************
self.example_io_geodataframe(preserve_original_gdf=True, identify_columns=False)
self.example_io_geodataframe(
preserve_original_gdf=False, identify_columns=False
# example_io_geodataframe(
# preserve_original_gdf=True, identify_columns=True
# )
# example_io_geodataframe(
# preserve_original_gdf=False, identify_columns=True
# )
# example_io_geodataframe(preserve_original_gdf=True,
# file_extension='.json')
# example_io_geodataframe(preserve_original_gdf=False,
# file_extension='.json')
# *************************************************************************
# *************************************************************************
def test_identifying_entrances_simple_no_driveway(self):
# no driveway, all nodes
self.example_identify_entrances_simple_no_driveway(
AB_right_BC_wrong=True,
Pedro L. Magalhães
committed
create_reversed_edges=False,
self.example_identify_entrances_simple_no_driveway(
AB_right_BC_wrong=False,
Pedro L. Magalhães
committed
create_reversed_edges=False,
self.example_identify_entrances_simple_no_driveway(
AB_right_BC_wrong=True,
Pedro L. Magalhães
committed
create_reversed_edges=True,
self.example_identify_entrances_simple_no_driveway(
AB_right_BC_wrong=False,
Pedro L. Magalhães
committed
create_reversed_edges=True,
Pedro L. Magalhães
committed
# no driveway, all nodes, multiple addresses per edge
self.example_identify_entrances_simple_no_driveway(
AB_right_BC_wrong=True,
Pedro L. Magalhães
committed
create_reversed_edges=False,
self.example_identify_entrances_simple_no_driveway(
AB_right_BC_wrong=False,
Pedro L. Magalhães
committed
create_reversed_edges=False,
self.example_identify_entrances_simple_no_driveway(
AB_right_BC_wrong=True,
Pedro L. Magalhães
committed
create_reversed_edges=True,
self.example_identify_entrances_simple_no_driveway(
AB_right_BC_wrong=False,
Pedro L. Magalhães
committed
create_reversed_edges=True,
self.example_identify_entrances_simple_no_driveway(
AB_right_BC_wrong=False,
Pedro L. Magalhães
committed
create_reversed_edges=True,
self.example_identify_entrances_simple_no_driveway(
AB_right_BC_wrong=True,
Pedro L. Magalhães
committed
create_reversed_edges=False,
self.example_identify_entrances_simple_no_driveway(
AB_right_BC_wrong=False,
Pedro L. Magalhães
committed
create_reversed_edges=False,
self.example_identify_entrances_simple_no_driveway(
AB_right_BC_wrong=True,
Pedro L. Magalhães
committed
create_reversed_edges=True,
self.example_identify_entrances_simple_no_driveway(
AB_right_BC_wrong=False,
Pedro L. Magalhães
committed
create_reversed_edges=True,
AB_right_BC_wrong=True, create_reversed_edges=False
)
AB_right_BC_wrong=False, create_reversed_edges=False
)
AB_right_BC_wrong=True, create_reversed_edges=True
)
AB_right_BC_wrong=False, create_reversed_edges=True
)
Pedro L. Magalhães
committed
# driveway, all nodes, multiple addresses per edge
self.example_identify_entrances_simple_driveway(
AB_right_BC_wrong=True,
Pedro L. Magalhães
committed
create_reversed_edges=False,
self.example_identify_entrances_simple_driveway(
AB_right_BC_wrong=False,
Pedro L. Magalhães
committed
create_reversed_edges=False,
self.example_identify_entrances_simple_driveway(
AB_right_BC_wrong=True,
Pedro L. Magalhães
committed
create_reversed_edges=True,
self.example_identify_entrances_simple_driveway(
AB_right_BC_wrong=False,
Pedro L. Magalhães
committed
create_reversed_edges=True,
self.example_identify_entrances_simple_driveway(
AB_right_BC_wrong=True,
Pedro L. Magalhães
committed
create_reversed_edges=False,
self.example_identify_entrances_simple_driveway(
AB_right_BC_wrong=False,
Pedro L. Magalhães
committed
create_reversed_edges=False,
self.example_identify_entrances_simple_driveway(
AB_right_BC_wrong=True,
Pedro L. Magalhães
committed
create_reversed_edges=True,
self.example_identify_entrances_simple_driveway(
AB_right_BC_wrong=False,
Pedro L. Magalhães
committed
create_reversed_edges=True,
self.example_identify_entrances_simple_driveway(
AB_right_BC_wrong=True,
Pedro L. Magalhães
committed
create_reversed_edges=False,
self.example_identify_entrances_simple_driveway(
AB_right_BC_wrong=True,
Pedro L. Magalhães
committed
create_reversed_edges=False,
self.example_identify_entrances_simple_driveway(
AB_right_BC_wrong=False,
Pedro L. Magalhães
committed
create_reversed_edges=False,
self.example_identify_entrances_simple_driveway(
AB_right_BC_wrong=True,
Pedro L. Magalhães
committed
create_reversed_edges=False,
Pedro L. Magalhães
committed
create_reversed_edges=False,
Pedro L. Magalhães
committed
create_reversed_edges=True,
Pedro L. Magalhães
committed
create_reversed_edges=False,
Pedro L. Magalhães
committed
create_reversed_edges=True,
Pedro L. Magalhães
committed
create_reversed_edges=False,
revert_to_original_crs=False,
focus_on_node_P_only=False,
Pedro L. Magalhães
committed
# TODO: test a case with multiple parallel edges
# *************************************************************************
# *************************************************************************
def test_generating_node_containers(self):
# test generating containers
self.example_generate_node_container(False, False, True)
# *************************************************************************
# *************************************************************************
def get_node_gdf_A(
self,
right_address: str = "right",
wrong_address: str = "wrong",
country_code: str = _osm.KEY_COUNTRY_DK,
):
# **************************************************************************
# geodataframe: should have 'addr:street', 'osak:identifier' and index
data={
_osm.KEY_OSM_STREET: [
address_A, # A
address_B, # B
address_C, # C
address_P,
], # P
_osm.KEY_OSM_BUILDING_ENTRANCE_ID[country_code]: [
node_uid_A, # A
node_uid_B, # B
node_uid_C, # C
node_uid_P,
], # P
_osm.KEY_OSMNX_ELEMENT_TYPE: ["node", "node", "node", "node"],
_osm.KEY_OSMNX_OSMID: [osmid_A, osmid_B, osmid_C, osmid_P],
},
geometry=[geo_A, geo_B, geo_C, geo_P],
)
gdf.set_index(
[_osm.KEY_OSMNX_ELEMENT_TYPE, _osm.KEY_OSMNX_OSMID], drop=True, inplace=True
)
# **************************************************************************
Pedro L. Magalhães
committed
# *****************************************************************************
# *****************************************************************************
def get_node_gdf_B(
self,
right_address: str = "right",
wrong_address: str = "wrong",
country_code: str = _osm.KEY_COUNTRY_DK,
):
# **************************************************************************
gdf = self.get_node_gdf_A(
right_address=right_address,
wrong_address=wrong_address,
country_code=country_code,
)
gdf_node_D = GeoDataFrame(
{
_osm.KEY_OSMNX_GEOMETRY: [geo_D],
_osm.KEY_OSM_BUILDING_ENTRANCE_ID[country_code]: [node_uid_D],
# _osm.KEY_OSM_STREET: [address_D],# P
# _osm.KEY_OSMNX_ELEMENT_TYPE: ['node'],
# _osm.KEY_OSMNX_OSMID: [osmid_D]
},
# index=[('node', osmid_D)],
index=MultiIndex.from_tuples(
[("node", osmid_D)],
names=[_osm.KEY_OSMNX_ELEMENT_TYPE, _osm.KEY_OSMNX_OSMID],
),
)
# **************************************************************************
# **************************************************************************
Pedro L. Magalhães
committed
# *****************************************************************************
# *****************************************************************************
def get_node_gdf_C(
self,
right_address: str = "right",
wrong_address: str = "wrong",
country_code: str = _osm.KEY_COUNTRY_DK,
):
# **************************************************************************
gdf = self.get_node_gdf_B(
right_address=right_address,
wrong_address=wrong_address,
country_code=country_code,
)
# address_E = right_address
gdf_node_E = GeoDataFrame(
{
_osm.KEY_OSMNX_GEOMETRY: [geo_E],
_osm.KEY_OSM_BUILDING_ENTRANCE_ID[country_code]: [node_uid_E],
# _osm.KEY_OSM_STREET: [address_E],
},
# index=[('node', osmid_E)]
index=MultiIndex.from_tuples(
[("node", osmid_E)],
names=[_osm.KEY_OSMNX_ELEMENT_TYPE, _osm.KEY_OSMNX_OSMID],
),
)
# **************************************************************************
# **************************************************************************
Pedro L. Magalhães
committed
# *****************************************************************************
# *****************************************************************************
def get_network_A(
self,
gdf: GeoDataFrame,
right_address: str = "right",
wrong_address: str = "wrong",
AB_right_BC_wrong: bool = True,
country_code: str = _osm.KEY_COUNTRY_DK,
use_multiple_addresses: bool = False,
):
# **************************************************************************
(
node_keys,
node_data_container,
node_key_to_gdf_index_dict,
) = gis_utils.prepare_node_data_from_geodataframe(
include_geometry=True, gdf=gdf
)
# **************************************************************************
Pedro L. Magalhães
committed
# two edges: AB and BC
node_key_A = "A"
node_key_B = "B"
node_key_C = "C"
Pedro L. Magalhães
committed
# edge AB
[
(
network.nodes[node_key_A][_osm.KEY_OSMNX_X],
network.nodes[node_key_A][_osm.KEY_OSMNX_Y],
),
(
network.nodes[node_key_B][_osm.KEY_OSMNX_X],
network.nodes[node_key_B][_osm.KEY_OSMNX_Y],
),
]
)
length_AB = network.nodes[node_key_A][_osm.KEY_OSMNX_GEOMETRY].distance(
network.nodes[node_key_B][_osm.KEY_OSMNX_GEOMETRY]
)
network.add_edge(
node_key_A,
node_key_B,
**{
_osm.KEY_OSMNX_GEOMETRY: geo_AB,
_osm.KEY_OSMNX_LENGTH: length_AB,
_osm.KEY_OSMNX_NAME: (
["HZ", (right_address if AB_right_BC_wrong else wrong_address)]
if use_multiple_addresses
else (right_address if AB_right_BC_wrong else wrong_address)
),
}
)
Pedro L. Magalhães
committed
# edge BC
[
(
network.nodes[node_key_B][_osm.KEY_OSMNX_X],
network.nodes[node_key_B][_osm.KEY_OSMNX_Y],
),
(
network.nodes[node_key_C][_osm.KEY_OSMNX_X],
network.nodes[node_key_C][_osm.KEY_OSMNX_Y],
),
]
)
length_BC = network.nodes[node_key_B][_osm.KEY_OSMNX_GEOMETRY].distance(
network.nodes[node_key_C][_osm.KEY_OSMNX_GEOMETRY]
)
network.add_edge(
node_key_B,
node_key_C,
**{
_osm.KEY_OSMNX_GEOMETRY: geo_BC,
_osm.KEY_OSMNX_LENGTH: length_BC,
_osm.KEY_OSMNX_NAME: (
[(wrong_address if AB_right_BC_wrong else right_address), "UQ"]
if use_multiple_addresses
else (wrong_address if AB_right_BC_wrong else right_address)
),
}
)
# **************************************************************************
return network, node_keys, node_key_to_gdf_index_dict
# **************************************************************************
Pedro L. Magalhães
committed
# *****************************************************************************
# *****************************************************************************
def get_network_B(
self,
gdf: GeoDataFrame,
right_address: str = "right",
wrong_address: str = "wrong",
AB_right_BC_wrong: bool = True,
BD_with_name: bool = False,
BD_right_address: bool = False,
country_code: str = _osm.KEY_COUNTRY_DK,
use_multiple_addresses: bool = False,
):
# **************************************************************************
network, node_keys, node_key_to_gdf_index_dict = self.get_network_A(
gdf=gdf,
right_address=right_address,
wrong_address=wrong_address,
country_code=country_code,
AB_right_BC_wrong=AB_right_BC_wrong,
use_multiple_addresses=use_multiple_addresses,
)
Pedro L. Magalhães
committed
# add nameless BD edge
Pedro L. Magalhães
committed
# edge AB
[
(
network.nodes[node_key_B][_osm.KEY_OSMNX_X],
network.nodes[node_key_B][_osm.KEY_OSMNX_Y],
),
(
network.nodes[node_key_D][_osm.KEY_OSMNX_X],
network.nodes[node_key_D][_osm.KEY_OSMNX_Y],
),
]
)
length_BD = network.nodes[node_key_B][_osm.KEY_OSMNX_GEOMETRY].distance(
network.nodes[node_key_D][_osm.KEY_OSMNX_GEOMETRY]
_osm.KEY_OSMNX_GEOMETRY: geo_BD,
_osm.KEY_OSMNX_LENGTH: length_BD,
# _osm.KEY_OSMNX_NAME: ( # no name for BD
# right_address if AB_right_BC_wrong else
# wrong_address)
}
)
network.add_edge(node_key_B, node_key_D, **BD_dict)
# **************************************************************************
return network, node_keys, node_key_to_gdf_index_dict
# **************************************************************************
Pedro L. Magalhães
committed
# *****************************************************************************
# *****************************************************************************
def get_network_C(
self,
gdf: GeoDataFrame,
right_address: str = "right",
wrong_address: str = "wrong",
country_code: str = _osm.KEY_COUNTRY_DK,
CE_wrong: bool = False,
use_multiple_addresses: bool = False,
):
# **************************************************************************
network, node_keys, node_key_to_gdf_index_dict = self.get_network_B(
gdf=gdf,
right_address=wrong_address,
wrong_address=wrong_address,
country_code=country_code,
Pedro L. Magalhães
committed
# add a CE edge with the right name
Pedro L. Magalhães
committed
# edge AB
[
(
network.nodes[node_key_C][_osm.KEY_OSMNX_X],
network.nodes[node_key_C][_osm.KEY_OSMNX_Y],
),
(
network.nodes[node_key_E][_osm.KEY_OSMNX_X],
network.nodes[node_key_E][_osm.KEY_OSMNX_Y],
),
]
)
length_CE = network.nodes[node_key_C][_osm.KEY_OSMNX_GEOMETRY].distance(
network.nodes[node_key_E][_osm.KEY_OSMNX_GEOMETRY]
)
network.add_edge(
node_key_C,
node_key_E,
**{
_osm.KEY_OSMNX_GEOMETRY: geo_CE,
_osm.KEY_OSMNX_LENGTH: length_CE,
_osm.KEY_OSMNX_NAME: (wrong_address if CE_wrong else right_address),
}
)
# **************************************************************************
return network, node_keys, node_key_to_gdf_index_dict
# **************************************************************************
Pedro L. Magalhães
committed
# *****************************************************************************
# *****************************************************************************
self,
create_reversed_edges: bool = False,
revert_to_original_crs: bool = False,
focus_on_node_P_only: bool = False,
CE_wrong: bool = False,
):
network, node_keys, node_key_to_gdf_index_dict = self.get_network_C(
gdf=gdf, country_code=country_code, CE_wrong=CE_wrong
)
Pedro L. Magalhães
committed
# create reverse edges
Pedro L. Magalhães
committed
if create_reversed_edges:
previous_edge_keys = list(edge_key for edge_key in network.edges(keys=True))
Pedro L. Magalhães
committed
for edge_key in previous_edge_keys:
edge_dict = network.get_edge_data(
u=edge_key[0], v=edge_key[1], key=edge_key[2]
)
network.add_edge(
u_for_edge=edge_key[1], v_for_edge=edge_key[0], **edge_dict
)
Pedro L. Magalhães
committed
# find out which is the closest edge
Pedro L. Magalhães
committed
nearest_edge_keys, _, _ = gis_utils.identify_building_entrance_edges(
gdf_street_column=_osm.KEY_OSM_STREET,
network=network,
node_key_to_gdf_index_dict={"P": node_key_to_gdf_index_dict["P"]},
revert_to_original_crs=revert_to_original_crs,
)
Pedro L. Magalhães
committed
nearest_edge_keys, _, _ = gis_utils.identify_building_entrance_edges(
gdf_street_column=_osm.KEY_OSM_STREET,
network=network,
node_key_to_gdf_index_dict=node_key_to_gdf_index_dict,
crs=None,
revert_to_original_crs=revert_to_original_crs,
)
Pedro L. Magalhães
committed
# no edges with the right address, the closest edge should be selected
assert ("B", "D", 0) == nearest_edge_keys["P"] or (
"D",
"B",
0,
) == nearest_edge_keys["P"]
else:
# CE has the right address, it should be selected
assert ("C", "E", 0) == nearest_edge_keys["P"] or (
"E",
"C",
0,
) == nearest_edge_keys["P"]
# **************************************************************************
Pedro L. Magalhães
committed
# *****************************************************************************
# *****************************************************************************
self,
AB_right_BC_wrong: bool = True,
create_reversed_edges: bool = False,
revert_to_original_crs: bool = False,
focus_on_node_P_only: bool = False,
BD_with_name: bool = False,
BD_right_address: bool = False,
use_multiple_addresses: bool = False,
):
network, node_keys, node_key_to_gdf_index_dict = self.get_network_B(
gdf=gdf,
country_code=country_code,
BD_with_name=BD_with_name,
BD_right_address=BD_right_address,
AB_right_BC_wrong=AB_right_BC_wrong,
use_multiple_addresses=use_multiple_addresses,
)
Pedro L. Magalhães
committed
# create reverse edges
Pedro L. Magalhães
committed
if create_reversed_edges:
previous_edge_keys = list(edge_key for edge_key in network.edges(keys=True))
Pedro L. Magalhães
committed
for edge_key in previous_edge_keys:
edge_dict = network.get_edge_data(
u=edge_key[0], v=edge_key[1], key=edge_key[2]
)
network.add_edge(
u_for_edge=edge_key[1], v_for_edge=edge_key[0], **edge_dict
)
Pedro L. Magalhães
committed
# find out which is the closest edge
Pedro L. Magalhães
committed
nearest_edge_keys, _, _ = gis_utils.identify_building_entrance_edges(
gdf_street_column=_osm.KEY_OSM_STREET,
network=network,
node_key_to_gdf_index_dict={"P": node_key_to_gdf_index_dict["P"]},
revert_to_original_crs=revert_to_original_crs,
)
Pedro L. Magalhães
committed
nearest_edge_keys, _, _ = gis_utils.identify_building_entrance_edges(
gdf_street_column=_osm.KEY_OSM_STREET,
network=network,
node_key_to_gdf_index_dict=node_key_to_gdf_index_dict,
crs=None,
revert_to_original_crs=revert_to_original_crs,
)
if BD_with_name and not BD_right_address:
if AB_right_BC_wrong:
assert ("A", "B", 0) == nearest_edge_keys["P"] or (
"B",
"A",
0,
) == nearest_edge_keys["P"]
assert ("B", "C", 0) == nearest_edge_keys["P"] or (
"C",
"B",
0,
) == nearest_edge_keys["P"]
Pedro L. Magalhães
committed
# assert (('B','D', 0) == nearest_edge_keys['P'] or
# ('D','B', 0) == nearest_edge_keys['P'])
assert ("B", "D", 0) == nearest_edge_keys["P"] or (
"D",
"B",
0,
) == nearest_edge_keys["P"]
# **************************************************************************
Pedro L. Magalhães
committed
# *****************************************************************************
# *****************************************************************************
self,
AB_right_BC_wrong: bool = True,
create_reversed_edges: bool = False,
focus_on_node_P_only: bool = False,
revert_to_original_crs: bool = False,
use_multiple_addresses: bool = False,
):
network, node_keys, node_key_to_gdf_index_dict = self.get_network_A(
gdf=gdf,
country_code=country_code,
AB_right_BC_wrong=AB_right_BC_wrong,
use_multiple_addresses=use_multiple_addresses,
)
Pedro L. Magalhães
committed
# create reverse edges
Pedro L. Magalhães
committed
if create_reversed_edges:
previous_edge_keys = list(edge_key for edge_key in network.edges(keys=True))
Pedro L. Magalhães
committed
for edge_key in previous_edge_keys:
edge_dict = network.get_edge_data(
u=edge_key[0], v=edge_key[1], key=edge_key[2]
)
network.add_edge(
u_for_edge=edge_key[1], v_for_edge=edge_key[0], **edge_dict
)
Pedro L. Magalhães
committed
# find out which is the closest edge
Pedro L. Magalhães
committed
nearest_edge_keys, _, _ = gis_utils.identify_building_entrance_edges(
gdf_street_column=_osm.KEY_OSM_STREET,
network=network,
node_key_to_gdf_index_dict={"P": node_key_to_gdf_index_dict["P"]},
revert_to_original_crs=revert_to_original_crs,
)
Pedro L. Magalhães
committed
nearest_edge_keys, _, _ = gis_utils.identify_building_entrance_edges(
gdf_street_column=_osm.KEY_OSM_STREET,
network=network,
node_key_to_gdf_index_dict=node_key_to_gdf_index_dict,
crs=None,
revert_to_original_crs=revert_to_original_crs,
)
Pedro L. Magalhães
committed
# the closest edge should be AB
assert ("A", "B", 0) == nearest_edge_keys["P"]
Pedro L. Magalhães
committed
# the closest edge should be BC
assert ("B", "C", 0) == nearest_edge_keys["P"]
# *************************************************************************
# *************************************************************************
def test_identify_entrances_simple_no_driveway_closest(self):
# get problem details