Skip to content
GitLab
Projects
Groups
Snippets
/
Help
Help
Support
Community forum
Keyboard shortcuts
?
Submit feedback
Contribute to GitLab
Sign in
Toggle navigation
Menu
Open sidebar
Dynamic Exposure
OpenBuildingMap
rabotnik-obm
Commits
b49f3f4c
Commit
b49f3f4c
authored
Jun 03, 2021
by
Marius Kriegerowski
Browse files
Added data dump to floorspace and tests
parent
9bd21411
Pipeline
#24310
passed with stage
in 1 minute and 48 seconds
Changes
4
Pipelines
2
Hide whitespace changes
Inline
Side-by-side
rabotnikobm/instance.py
View file @
b49f3f4c
...
...
@@ -63,10 +63,10 @@ async def main():
rule
=
GetFloorspace
(
storage_osmreplication
,
storage_obmbuildings
)
await
message_bus
.
subscribe
(
"building"
,
rule
.
evaluate
)
rule
=
GetBuildingLandUse
(
storage_osmreplication
,
storage_obmbuildings
)
rule
=
GetBuildingLandUse
(
storage_osmreplication
)
await
message_bus
.
subscribe
(
"building"
,
rule
.
evaluate
)
rule
=
GetPointsInBuilding
(
storage_osmreplication
,
storage_obmbuildings
)
rule
=
GetPointsInBuilding
(
storage_osmreplication
)
await
message_bus
.
subscribe
(
"building"
,
rule
.
evaluate
)
...
...
rabotnikobm/rules/get_floorspace.py
View file @
b49f3f4c
...
...
@@ -8,28 +8,42 @@ logger = logging.getLogger()
class
GetFloorspace
(
Rule
):
def
__init__
(
self
,
storage
:
RabotnikStorage
):
def
__init__
(
self
,
storage
:
RabotnikStorage
,
storage_target
:
RabotnikStorage
):
"""Calculate floorspace of buildings.
Args:
storage: Storage serving building data.
storage_target: Storage to fill in result.
"""
self
.
storage
=
storage
self
.
storage_target
=
storage_target
async
def
evaluate
(
self
,
payload
:
dict
)
->
tupl
e
:
async
def
evaluate
(
self
,
payload
:
dict
)
->
Non
e
:
building_id
=
payload
[
"building_id"
]
async
for
data
in
self
.
storage
.
iter_results
(
f
"""
query_data
=
f
"""
SELECT ST_Area(ST_Transform(geometry, 4326)) *
CAST(br.tags->'building:levels' AS float)
FROM osm_building_relations AS br
FROM osm_building_relations AS br
WHERE br.tags->'building:levels' IS NOT NULL
AND osm_id =
{
building_id
}
UNION ALL
SELECT ST_Area(ST_Transform(geometry, 4326)) *
CAST(bp.tags->'building:levels' AS float)
FROM osm_building_polygons AS bp
WHERE id =
{
building_id
}
;
WHERE bp.tags->'building:levels' IS NOT NULL
AND osm_id =
{
building_id
}
;
"""
write_data
=
"""
INSERT INTO obm_buildings (osm_id, floorspace) VALUES ({}, {})
"""
):
yield
data
async
for
data
in
self
.
storage
.
iter_results
(
query_data
):
assert
len
(
data
)
==
1
data
=
data
[
0
]
async
for
response
in
self
.
storage_target
.
iter_results
(
write_data
.
format
(
*
(
building_id
,
data
))
):
logger
.
debug
(
response
)
tests/conftest.py
View file @
b49f3f4c
...
...
@@ -18,6 +18,14 @@ def pytest_configure(config):
config
.
addinivalue_line
(
"markers"
,
"requires_storage: mark test that require a storage"
)
@
pytest
.
fixture
@
pytest
.
mark
.
asyncio
async
def
mock_storage
():
storage
=
rabotnik
.
Rabotnik
.
get_storage
(
"mockeddb"
)
storage
.
connect
()
yield
storage
@
pytest
.
fixture
@
pytest
.
mark
.
asyncio
async
def
connected_storage
(
pytestconfig
):
...
...
tests/test_get_floorspace.py
0 → 100644
View file @
b49f3f4c
#!/usr/bin/env python3
# Copyright (C) 2021:
# Helmholtz-Zentrum Potsdam Deutsches GeoForschungsZentrum GFZ
#
# This program is free software: you can redistribute it and/or modify it
# under the terms of the GNU Affero General Public License as published by
# the Free Software Foundation, either version 3 of the License, or (at
# your option) any later version.
#
# This program is distributed in the hope that it will be useful, but
# WITHOUT ANY WARRANTY; without even the implied warranty of
# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the GNU Affero
# General Public License for more details.
#
# You should have received a copy of the GNU Affero General Public License
# along with this program. If not, see http://www.gnu.org/licenses/.
import
pytest
from
rabotnikobm.rules.get_floorspace
import
(
GetFloorspace
,
)
@
pytest
.
mark
.
requires_storage
@
pytest
.
mark
.
asyncio
async
def
test_get_floorspace
(
connected_storage
,
mock_storage
):
c
=
mock_storage
.
connection
.
cursor
()
c
.
execute
(
"""CREATE TABLE obm_buildings (
osm_id INTEGER,
floorspace FLOAT
)
"""
)
rule
=
GetFloorspace
(
storage
=
connected_storage
,
storage_target
=
mock_storage
)
payload
=
{
"building_id"
:
848970775
}
await
rule
.
evaluate
(
payload
=
payload
)
data_out
=
mock_storage
.
get_results
(
"SELECT * FROM obm_buildings;"
)
assert
data_out
==
[(
848970775
,
2.1424237391628443e-09
)]
Write
Preview
Supports
Markdown
0%
Try again
or
attach a new file
.
Cancel
You are about to add
0
people
to the discussion. Proceed with caution.
Finish editing this message first!
Cancel
Please
register
or
sign in
to comment