Skip to content

Commit

Permalink
Merge pull request #2050 from openoereb/change-legend-entries-retrieval
Browse files Browse the repository at this point in the history 
change-legend-entries-retrieval
  • Loading branch information
@svamaa
svamaa authored Oct 17, 2024
2 parents a6c972f + ef97229 commit b917447
Show file tree
Hide file tree
Showing 4 changed files with 286 additions and 41 deletions.
58 changes: 38 additions & 20 deletions pyramid_oereb/contrib/data_sources/interlis_2_3/sources/plr.py
View file
Original file line number Diff line number Diff line change
Expand Up @@ -470,20 +470,33 @@ def collect_related_geometries_by_real_estate(self, session, real_estate):
.selectinload(self.models.MultilingualUri.localised_uri)
).all()

def collect_legend_entries_by_bbox(self, session, bbox, law_status):
def get_legend_entries_from_db(self, session, legend_entry_ids):
"""
Retrieves the legend entries for a list of t_id-values.
Args:
session (sqlalchemy.orm.Session): The requested clean session instance ready for use
legend_entry_ids (list): list of the egend entry ids
Returns:
list: the query result represented as a list.
"""
return session.query(self.legend_entry_model).filter(
self.legend_entry_model.t_id.in_(legend_entry_ids)).all()

def collect_legend_entries_by_bbox(self, session, bbox):
"""
Extracts all legend entries in the topic which have spatial relation with the passed bounding box of
visible extent.
Args:
session (sqlalchemy.orm.Session): The requested clean session instance ready for use
bbox (shapely.geometry.base.BaseGeometry): The bbox to search the records.
law_status (str): String of the law status for which the legend entries should be queried.
Returns:
list: The result of the related geometries unique by the public law restriction id and law status
"""
distinct_legend_entry_ids = []
# Select the legend entries of all plr within bbox
geometries = session.query(self._model_).filter(
or_(
self._model_.point.ST_Intersects(from_shape(bbox, srid=Config.get('srid'))),
Expand All @@ -493,13 +506,29 @@ def collect_legend_entries_by_bbox(self, session, bbox, law_status):
selectinload(self.models.Geometry.public_law_restriction)
).all()

# Compile a list of unique legend entry ids for each law status
legend_entry_ids = dict()
for geometry in geometries:
if geometry.public_law_restriction.legend_entry_id not in distinct_legend_entry_ids\
and geometry.public_law_restriction.law_status == law_status:
distinct_legend_entry_ids.append(geometry.public_law_restriction.legend_entry_id)
if geometry.public_law_restriction.law_status not in legend_entry_ids:
legend_entry_ids[geometry.public_law_restriction.law_status] = {
geometry.public_law_restriction.legend_entry_id
}
else:
legend_entry_ids[geometry.public_law_restriction.law_status].add(
geometry.public_law_restriction.legend_entry_id
)

return session.query(self.legend_entry_model).filter(
self.legend_entry_model.t_id.in_((distinct_legend_entry_ids))).all()
# Retrieve legend entries
legend_entries_from_db = []
for law_status in legend_entry_ids:
legend_entries_from_db.append(
[
self.get_legend_entries_from_db(session, list(legend_entry_ids[law_status])),
law_status
]
)

return legend_entries_from_db

def read(self, params, real_estate, bbox):
"""
Expand Down Expand Up @@ -537,20 +566,9 @@ def read(self, params, real_estate, bbox):
else:
# We found spatially related elements. This means we need to extract the actual plr
# information related to the found geometries.
law_status_of_geometry = []
# get distinct values of law_status for all geometries found
for geometry in geometry_results:
if (geometry.public_law_restriction.law_status not in law_status_of_geometry):
law_status_of_geometry.append(geometry.public_law_restriction.law_status)

legend_entries_from_db = []
# get legend_entries per law_status
for law_status in law_status_of_geometry:
legend_entry_with_law_status = [
self.collect_legend_entries_by_bbox(session, bbox, law_status),
law_status
]
legend_entries_from_db.append(legend_entry_with_law_status)
legend_entries_from_db = self.collect_legend_entries_by_bbox(session, bbox)

self.records = []
for geometry_result in geometry_results:
Expand Down
62 changes: 41 additions & 21 deletions pyramid_oereb/contrib/data_sources/standard/sources/plr.py
View file
Original file line number Diff line number Diff line change
Expand Up @@ -614,30 +614,62 @@ def collect_related_geometries_by_real_estate(self, session, real_estate):
.selectinload(self.models.PublicLawRestriction.responsible_office),
).all()

def collect_legend_entries_by_bbox(self, session, bbox, law_status):
def get_legend_entries_from_db(self, session, legend_entry_ids):
"""
Retrieves the legend entries for a list of id-values.
Args:
session (sqlalchemy.orm.Session): The requested clean session instance ready for use
legend_entry_ids (list): list of the egend entry ids
Returns:
list: the query result represented as a list.
"""

return session.query(self.legend_entry_model).filter(
self.legend_entry_model.id.in_(legend_entry_ids)).all()

def collect_legend_entries_by_bbox(self, session, bbox):
"""
Extracts all legend entries in the topic which have spatial relation with the passed bounding box of
visible extent.
Args:
session (sqlalchemy.orm.Session): The requested clean session instance ready for use
bbox (shapely.geometry.base.BaseGeometry): The bbox to search the records.
law_status (str): String of the law status for which the legend entries should be queried.
Returns:
list: The result of the related geometries unique by the public law restriction id and law status
"""

distinct_legend_entry_ids = []
# Select the legend entries of all plr within bbox
geometries = self.handle_collection(session, bbox).options(
selectinload(self.models.Geometry.public_law_restriction)
).all()

# Compile a list of unique legend entry ids for each law status
legend_entry_ids = dict()
for geometry in geometries:
if geometry.public_law_restriction.legend_entry_id not in distinct_legend_entry_ids \
and geometry.public_law_restriction.law_status == law_status:
distinct_legend_entry_ids.append(geometry.public_law_restriction.legend_entry_id)
return session.query(self.legend_entry_model).filter(
self.legend_entry_model.id.in_((distinct_legend_entry_ids))).all()
if geometry.public_law_restriction.law_status not in legend_entry_ids:
legend_entry_ids[geometry.public_law_restriction.law_status] = {
geometry.public_law_restriction.legend_entry_id
}
else:
legend_entry_ids[geometry.public_law_restriction.law_status].add(
geometry.public_law_restriction.legend_entry_id
)

# Retrieve legend entries
legend_entries_from_db = []
for law_status in legend_entry_ids:
legend_entries_from_db.append(
[
self.get_legend_entries_from_db(session, list(legend_entry_ids[law_status])),
law_status
]
)

return legend_entries_from_db

def read(self, params, real_estate, bbox): # pylint: disable=W:0221
"""
Expand Down Expand Up @@ -674,20 +706,8 @@ def read(self, params, real_estate, bbox): # pylint: disable=W:0221
# We found spatially related elements. This means we need to extract the actual plr
# information related to the found geometries.

law_status_of_geometry = []
# get distinct values of law_status for all geometries found
for geometry in geometry_results:
if (geometry.public_law_restriction.law_status not in law_status_of_geometry):
law_status_of_geometry.append(geometry.public_law_restriction.law_status)

legend_entries_from_db = []
# get legend_entries per law_status
for law_status in law_status_of_geometry:
legend_entry_with_law_status = [
self.collect_legend_entries_by_bbox(session, bbox, law_status),
law_status
]
legend_entries_from_db.append(legend_entry_with_law_status)
legend_entries_from_db = self.collect_legend_entries_by_bbox(session, bbox)

self.records = []
for geometry_result in geometry_results:
Expand Down
120 changes: 120 additions & 0 deletions tests/contrib.data_sources.interlis_2_3/sources/test_plr.py
View file
Original file line number Diff line number Diff line change
Expand Up @@ -20,6 +20,7 @@
from pyramid_oereb.contrib.data_sources.interlis_2_3.sources.plr import (
StandardThemeConfigParser
)
from pyramid_oereb.contrib.data_sources.interlis_2_3.sources.plr import DatabaseSource


@pytest.fixture(scope='session')
Expand Down Expand Up @@ -209,6 +210,21 @@ def oblique_limit_real_estate_record():
)


@pytest.fixture
def plr_source_params(db_connection):
yield {
"source": {
"class": "pyramid_oereb.contrib.data_sources.interlis_2_3.sources.plr.DatabaseSource",
"params": {
"db_connection": db_connection,
"model_factory": "pyramid_oereb.contrib.data_sources.interlis_2_3."
"models.theme.model_factory_integer_pk",
"schema_name": "land_use_plans"
}
}
}


# @pytest.fixture
# def interlis_real_estate():
# theme = ThemeRecord('code', dict(), 100)
Expand Down Expand Up @@ -286,3 +302,107 @@ def test_related_geometries(processor_data, pyramid_oereb_test_config, interlis_
assert len(extract_raw.real_estate.public_law_restrictions) == nb_results
extract = processor.plr_tolerance_check(extract_raw)
assert len(extract.real_estate.public_law_restrictions) == nb_results


def mock_session_object_query_geometries(items_list):
class PublicLawRestrictionTest():
def __init__(self, law_status, legend_entry_id):
self.law_status = law_status
self.legend_entry_id = legend_entry_id

class GeometryTest():
def __init__(self, public_law_restriction):
self.public_law_restriction = public_law_restriction

geometries = []
for item in items_list:
geometries.append(GeometryTest(PublicLawRestrictionTest(item[0], item[1])))

class AllTest():
def all():
return iter(geometries)

class DistinctTest():
def __init__(self):
pass

def options(arg2):
return AllTest

class FilterTest():
def __init__(self):
pass

def distinct(arg2):
return DistinctTest

class QueryTest():
def __init__(self):
pass

def filter(arg3):
return FilterTest

class SessionTest():
def __init__(self):
pass

def query(arg1):
return QueryTest

return SessionTest


def get_return_vals_of_get_legend_entries_from_db(arg1, arg2, list_of_ids):
return_value = []
for id in list_of_ids:
return_value.append((id, ))
return return_value


@pytest.mark.parametrize('idx,items_list', [
(0, [
["inForce", 1],
["changeWithoutPreEffect", 1],
["changeWithoutPreEffect", 2],
["inForce", 3],
["inForce", 4],
["inForce", 3],
["changeWithoutPreEffect", 6],
["inForce", 7],
["changeWithoutPreEffect", 2],
["inForce", 9],
["changeWithoutPreEffect", 7]
]),
(1, [
["inForce", 1],
["inForce", 3],
["inForce", 4],
["inForce", 3],
["inForce", 7],
["inForce", 9],
])
])
def test_collect_legend_entries_by_bbox(idx, items_list, plr_source_params):
with (
patch.object(
DatabaseSource,
'get_legend_entries_from_db',
get_return_vals_of_get_legend_entries_from_db
)
):
source = DatabaseSource(**plr_source_params)
result = source.collect_legend_entries_by_bbox(
mock_session_object_query_geometries(items_list),
Polygon(((0., 0.), (0., 1.), (1., 1.), (1., 0.), (0., 0.))))

if idx == 0:
assert len(result) == 2
assert sorted([x[0] for x in result if x[1] == 'inForce'][0]) == \
[(1, ), (3, ), (4, ), (7, ), (9, )]
assert sorted([x[0] for x in result if x[1] == 'changeWithoutPreEffect'][0]) == \
[(1, ), (2, ), (6, ), (7, )]
if idx == 1:
assert len(result) == 1
assert sorted([x[0] for x in result if x[1] == 'inForce'][0]) == \
[(1, ), (3, ), (4, ), (7, ), (9, )]
Loading

0 comments on commit b917447

Please sign in to comment.