Skip to content
Closed
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
31 changes: 24 additions & 7 deletions src/osm_osw_reformatter/serializer/osw/osw_normalizer.py
Original file line number Diff line number Diff line change
Expand Up @@ -18,6 +18,11 @@ class OSWWayNormalizer:
"trunk_link"
)

CLIMB_VALUES = (
"up",
"down",
)

def __init__(self, tags):
self.tags = tags

Expand Down Expand Up @@ -67,7 +72,16 @@ def normalize(self):
raise ValueError("This is an invalid way")

def _normalize_way(self, keep_keys={}, defaults = {}):
generic_keep_keys = {"highway": str, "width": float, "surface": surface, "name": str, "description": str, "foot": foot}
generic_keep_keys = {
"highway": str,
"width": float,
"surface": surface,
"name": str,
"description": str,
"foot": foot,
"incline": incline,
"length": float,
}
generic_defaults = {}

new_tags = _normalize(self.tags, {**generic_keep_keys, **keep_keys}, {**generic_defaults, **defaults})
Expand All @@ -81,8 +95,8 @@ def _normalize_pedestrian(self, keep_keys = {}, defaults = {}):
return new_tags

def _normalize_stairs(self, keep_keys = {}, defaults = {}):
generic_keep_keys = {"step_count": int, "incline": ["climb", climb]}
generic_defaults = {"foot": "yes"}
generic_keep_keys = {"step_count": int, "climb": climb}
generic_defaults = {"highway": "steps"}

new_tags = self._normalize_way({**generic_keep_keys, **keep_keys}, {**generic_defaults, **defaults})
return new_tags
Expand Down Expand Up @@ -584,13 +598,16 @@ def crossing_markings(tag_value, tags):
return None

def climb(tag_value, tags):
if tag_value.lower() not in (
"up",
"down"
):
if tag_value.lower() not in OSWWayNormalizer.CLIMB_VALUES:
return None
else:
return tag_value.lower()

def incline(tag_value, tags):
try:
return float(str(tag_value).rstrip('%'))
except (ValueError, TypeError):
return None

def foot(tag_value, tags):
if tag_value.lower() not in (
Expand Down
12 changes: 12 additions & 0 deletions tests/unit_tests/test_files/incline-test.xml
Original file line number Diff line number Diff line change
@@ -0,0 +1,12 @@
<?xml version="1.0" encoding="UTF-8"?>
<osm version="0.6" generator="test" upload="false">
<node id="1" lat="0.0" lon="0.0" />
<node id="2" lat="0.0" lon="0.1" />
<way id="1">
<nd ref="1"/>
<nd ref="2"/>
<tag k="highway" v="footway"/>
<tag k="incline" v="0.1"/>
<tag k="_id" v="1"/>
</way>
</osm>
28 changes: 28 additions & 0 deletions tests/unit_tests/test_osm2osw/test_osm2osw.py
Original file line number Diff line number Diff line change
Expand Up @@ -10,6 +10,7 @@
OUTPUT_DIR = os.path.join(os.path.dirname(os.path.dirname(ROOT_DIR)), 'output')
TEST_FILE = os.path.join(ROOT_DIR, 'test_files/wa.microsoft.osm.pbf')
TEST_WIDTH_FILE = os.path.join(ROOT_DIR, 'test_files/width-test.xml')
TEST_INCLINE_FILE = os.path.join(ROOT_DIR, 'test_files/incline-test.xml')


def is_valid_float(value):
Expand Down Expand Up @@ -218,6 +219,33 @@ async def run_test():

asyncio.run(run_test())

def test_retains_incline_tag(self):
osm_file_path = TEST_INCLINE_FILE

async def run_test():
osm2osw = OSM2OSW(osm_file=osm_file_path, workdir=OUTPUT_DIR, prefix='test')
result = await osm2osw.convert()
self.assertTrue(result.status)

found_incline = False
for file_path in result.generated_files:
if file_path.endswith('edges.geojson'):
with open(file_path) as f:
geojson = json.load(f)
for feature in geojson.get('features', []):
props = feature.get('properties', {})
if 'incline' in props:
self.assertIsInstance(props['incline'], (int, float))
found_incline = True
break

for file_path in result.generated_files:
os.remove(file_path)

self.assertTrue(found_incline, 'Incline tag not found in output edges')

asyncio.run(run_test())


if __name__ == '__main__':
unittest.main()
33 changes: 33 additions & 0 deletions tests/unit_tests/test_osm_compliance/test_osm_compliance.py
Original file line number Diff line number Diff line change
Expand Up @@ -36,3 +36,36 @@ async def test_output_is_osm_compliant(self):
os.remove(f)
os.remove(zip_path)
formatter.cleanup()

async def test_incline_tag_preserved(self):
osw2osm = OSW2OSM(
zip_file_path=TEST_DATA_WITH_INCLINE_ZIP_FILE,
workdir=OUTPUT_DIR,
prefix='incline'
)
result = osw2osm.convert()
osm_file = result.generated_files

formatter = Formatter(workdir=OUTPUT_DIR, file_path=osm_file, prefix='incline')
res = await formatter.osm2osw()
osw_files = res.generated_files

found_incline = False
for f in osw_files:
if f.endswith('.geojson'):
with open(f) as fh:
data = json.load(fh)
for feature in data.get('features', []):
props = feature.get('properties', {})
if 'incline' in props:
found_incline = True
break
if found_incline:
break

self.assertTrue(found_incline, 'No incline tag found in OSW output')

os.remove(osm_file)
for f in osw_files:
os.remove(f)
formatter.cleanup()
66 changes: 63 additions & 3 deletions tests/unit_tests/test_serializer/test_osw_normalizer.py
Original file line number Diff line number Diff line change
@@ -1,6 +1,21 @@
import unittest
from src.osm_osw_reformatter.serializer.osw.osw_normalizer import OSWWayNormalizer, OSWNodeNormalizer, \
OSWPointNormalizer, tactile_paving, surface, crossing_markings, climb, _normalize
import importlib.util
from pathlib import Path

module_path = Path(__file__).resolve().parents[3] / 'src/osm_osw_reformatter/serializer/osw/osw_normalizer.py'
spec = importlib.util.spec_from_file_location('osw_normalizer', module_path)
osw_normalizer = importlib.util.module_from_spec(spec)
spec.loader.exec_module(osw_normalizer)

OSWWayNormalizer = osw_normalizer.OSWWayNormalizer
OSWNodeNormalizer = osw_normalizer.OSWNodeNormalizer
OSWPointNormalizer = osw_normalizer.OSWPointNormalizer
tactile_paving = osw_normalizer.tactile_paving
surface = osw_normalizer.surface
crossing_markings = osw_normalizer.crossing_markings
climb = osw_normalizer.climb
incline = osw_normalizer.incline
_normalize = osw_normalizer._normalize


class TestOSWWayNormalizer(unittest.TestCase):
Expand Down Expand Up @@ -29,6 +44,11 @@ def test_is_stairs(self):
normalizer = OSWWayNormalizer(tags)
self.assertTrue(normalizer.is_stairs())

def test_is_stairs_with_invalid_climb(self):
tags = {'highway': 'steps', 'climb': 'left'}
normalizer = OSWWayNormalizer(tags)
self.assertTrue(normalizer.is_stairs())

def test_is_pedestrian(self):
tags = {'highway': 'pedestrian'}
normalizer = OSWWayNormalizer(tags)
Expand All @@ -53,6 +73,41 @@ def test_normalize_crossing(self):
expected = {'highway': 'footway', 'footway': 'crossing', 'foot': 'yes'}
self.assertEqual(result, expected)

def test_normalize_incline(self):
tags = {'highway': 'footway', 'incline': '10%'}
normalizer = OSWWayNormalizer(tags)
result = normalizer.normalize()
expected = {'highway': 'footway', 'incline': 10.0, 'foot': 'yes'}
self.assertEqual(result, expected)

def test_normalize_length(self):
tags = {'highway': 'footway', 'length': '12'}
normalizer = OSWWayNormalizer(tags)
result = normalizer.normalize()
expected = {'highway': 'footway', 'length': 12.0, 'foot': 'yes'}
self.assertEqual(result, expected)

def test_normalize_stairs_keeps_climb(self):
tags = {'highway': 'steps', 'climb': 'down', 'step_count': '3'}
normalizer = OSWWayNormalizer(tags)
result = normalizer.normalize()
expected = {'highway': 'steps', 'climb': 'down', 'step_count': 3}
self.assertEqual(result, expected)

def test_normalize_stairs_defaults_highway_and_no_foot(self):
tags = {'climb': 'up'}
normalizer = OSWWayNormalizer(tags)
result = normalizer._normalize_stairs()
expected = {'highway': 'steps', 'climb': 'up'}
self.assertEqual(result, expected)

def test_normalize_stairs_drops_invalid_climb(self):
tags = {'highway': 'steps', 'climb': 'left'}
normalizer = OSWWayNormalizer(tags)
result = normalizer.normalize()
expected = {'highway': 'steps'}
self.assertEqual(result, expected)

def test_normalize_invalid_way(self):
tags = {'highway': 'invalid_type'}
normalizer = OSWWayNormalizer(tags)
Expand Down Expand Up @@ -121,11 +176,16 @@ def test_crossing_markings(self):
self.assertEqual(crossing_markings('dots', {'crossing:markings': 'dots'}), 'dots')
self.assertIsNone(crossing_markings('invalid_value', {'crossing:markings': 'invalid_value'}))

def test_incline(self):
def test_climb(self):
self.assertEqual(climb('up', {}), 'up')
self.assertEqual(climb('down', {}), 'down')
self.assertIsNone(climb('invalid_value', {}))

def test_incline(self):
self.assertEqual(incline('10%', {}), 10.0)
self.assertEqual(incline('0.5', {}), 0.5)
self.assertIsNone(incline('steep', {}))


class TestNormalizeWidthField(unittest.TestCase):
def test_removes_width_when_value_is_nan_string(self):
Expand Down