Github actions: add tests for linting and type checking

.flake8

@@ -0,0 +1,5 @@
+[flake8]
+max-line-length = 100
+max-doc-length = 100
+per-file-ignores =
+    __init__.py: F401

.github/workflows/ci.yml

@@ -12,7 +12,16 @@ jobs:
             - name: Install packages
               run: |
                 sudo apt-get update -y -qq
-                sudo apt-get install -y -qq libboost-dev libexpat1-dev zlib1g-dev libbz2-dev libproj-dev libgeos-dev liblz4-dev
+                sudo apt-get install -y -qq libboost-dev libexpat1-dev zlib1g-dev libbz2-dev libproj-dev libgeos-dev liblz4-dev pipx
+                pipx install mypy
+                pipx inject mypy types-requests
+                pipx install flake8
+
+            - name: Lint package
+              run: flake8 src examples
+
+            - name: Typecheck package
+              run: mypy src
 
             - name: Set up Python 3.7
               uses: actions/setup-python@v5

examples/amenity_list.py

@@ -13,6 +13,7 @@
 
 wkbfab = osmium.geom.WKBFactory()
 
+
 class AmenityListHandler(osmium.SimpleHandler):
 
     def print_amenity(self, tags, lon, lat):
@@ -37,6 +38,7 @@ def main(osmfile):
 
     return 0
 
+
 if __name__ == '__main__':
     if len(sys.argv) != 2:
         print("Usage: python %s <osmfile>" % sys.argv[0])

examples/convert.py

@@ -24,4 +24,3 @@
             writer.add_relation(obj)
 
     writer.close()
-

examples/convert_to_geojson.py

@@ -10,6 +10,7 @@
 
 geojsonfab = osmium.geom.GeoJSONFactory()
 
+
 class GeoJsonWriter(osmium.SimpleHandler):
 
     def __init__(self):
@@ -48,7 +49,8 @@ def print_object(self, geojson, tags):
 def main(osmfile):
     handler = GeoJsonWriter()
 
-    handler.apply_file(osmfile,filters=[osmium.filter.EmptyTagFilter().enable_for(osmium.osm.NODE)])
+    handler.apply_file(osmfile,
+                       filters=[osmium.filter.EmptyTagFilter().enable_for(osmium.osm.NODE)])
     handler.finish()
 
     return 0

examples/filter_coastlines.py

@@ -32,7 +32,8 @@
     way_filter = osmium.filter.KeyFilter('natural').enable_for(osmium.osm.WAY)
 
     # We need nodes and ways in the second pass.
-    for obj in osmium.FileProcessor(sys.argv[1], osmium.osm.WAY | osmium.osm.NODE).with_filter(way_filter):
+    for obj in osmium.FileProcessor(sys.argv[1], osmium.osm.WAY | osmium.osm.NODE) \
+                     .with_filter(way_filter):
         if obj.is_node() and obj.id in nodes:
             # Strip the object of tags along the way
             writer.add_node(obj.replace(tags={}))

examples/normalize_boolean.py

@@ -6,6 +6,7 @@
 import osmium
 import sys
 
+
 class BoolNormalizer(osmium.SimpleHandler):
 
     def __init__(self, writer):

examples/osm_diff_stats.py

@@ -7,6 +7,7 @@
 import osmium
 import sys
 
+
 class Stats:
 
     def __init__(self):
@@ -40,6 +41,7 @@ def main(osmfile):
 
     return 0
 
+
 if __name__ == '__main__':
     if len(sys.argv) != 2:
         print("Usage: python %s <osmfile>" % sys.argv[0])

examples/osm_file_stats.py

@@ -6,6 +6,7 @@
 import osmium
 import sys
 
+
 class FileStatsHandler(osmium.SimpleHandler):
 
     def __init__(self):
@@ -35,6 +36,7 @@ def main(osmfile):
 
     return 0
 
+
 if __name__ == '__main__':
     if len(sys.argv) != 2:
         print("Usage: python %s <osmfile>" % sys.argv[0])

examples/osm_replication_stats.py

@@ -9,7 +9,8 @@
 import datetime as dt
 import osmium.replication.server as rserv
 
-class Stats(object):
+
+class Stats:
 
     def __init__(self):
         self.added = 0
@@ -24,12 +25,12 @@ def add(self, o):
         else:
             self.modified += 1
 
-
     def outstats(self, prefix):
         print("%s added: %d" % (prefix, self.added))
         print("%s modified: %d" % (prefix, self.modified))
         print("%s deleted: %d" % (prefix, self.deleted))
 
+
 class FileStatsHandler(osmium.SimpleHandler):
     def __init__(self):
         super(FileStatsHandler, self).__init__()
@@ -54,8 +55,7 @@ def relation(self, r):
 
     server_url = sys.argv[1]
     start = dt.datetime.strptime(sys.argv[2], "%Y-%m-%dT%H:%M:%SZ")
-    if sys.version_info >= (3,0):
-        start = start.replace(tzinfo=dt.timezone.utc)
+    start = start.replace(tzinfo=dt.timezone.utc)
     maxkb = min(int(sys.argv[3]), 10 * 1024)
 
     repserv = rserv.ReplicationServer(server_url)

examples/osm_url_stats.py

@@ -13,7 +13,6 @@
         print("Usage: python osm_url_stats.py <osmfile>")
         sys.exit(-1)
 
-
     data = urlrequest.urlopen(sys.argv[1]).read()
 
     counter = {'n': 0, 'w': 0, 'r': 0}

examples/pub_names.py

@@ -4,6 +4,7 @@
 import osmium
 import sys
 
+
 def main(osmfile):
     for obj in osmium.FileProcessor(osmfile)\
                      .with_filter(osmium.filter.KeyFilter('amenity'))\
@@ -13,6 +14,7 @@ def main(osmfile):
 
     return 0
 
+
 if __name__ == '__main__':
     if len(sys.argv) != 2:
         print("Usage: python %s <osmfile>" % sys.argv[0])

examples/road_length.py

@@ -6,13 +6,14 @@
 import osmium
 import sys
 
+
 def main(osmfile):
     total = 0.0
     # As we need the way geometry, the node locations need to be cached.
     # This is enabled with the with_locations() function.
     for obj in osmium.FileProcessor(osmfile, osmium.osm.NODE | osmium.osm.WAY)\
-                .with_locations()\
-                .with_filter(osmium.filter.KeyFilter('highway')):
+                     .with_locations()\
+                     .with_filter(osmium.filter.KeyFilter('highway')):
         if obj.is_way():
             try:
                 total += osmium.geom.haversine_distance(obj.nodes)
@@ -25,6 +26,7 @@ def main(osmfile):
 
     return 0
 
+
 if __name__ == '__main__':
     if len(sys.argv) != 2:
         print("Usage: python %s <osmfile>" % sys.argv[0])

examples/use_nodecache.py

@@ -4,15 +4,18 @@
 import osmium
 import sys
 
+
 class WayHandler:
 
     def __init__(self, idx):
         self.idx = idx
 
     def way(self, w):
+        locations = []
         for n in w.nodes:
-            loc = idx.get(n.ref) # note that cache is used here
-        print("%d %s" % (w.id, len(w.nodes)))
+            locations.append(idx.get(n.ref))  # note that cache is used here
+        print(w.id, len(w.nodes), locations)
+
 
 if len(sys.argv) != 3:
     print("Usage: python use_nodecache.py <osm file> <node cache>")

src/osmium/back_reference_writer.py

@@ -14,6 +14,7 @@
 from osmium.file_processor import FileProcessor, zip_processors
 from osmium import IdTracker
 
+
 class BackReferenceWriter:
     """ Writer that adds referenced objects, so that all written
         objects are reference-complete.
@@ -28,7 +29,7 @@ class BackReferenceWriter:
 
     def __init__(self, outfile: Union[str, 'os.PathLike[str]', File],
                  ref_src: Union[str, 'os.PathLike[str]', File, FileBuffer],
-                 overwrite: bool=False, remove_tags: bool=True,
+                 overwrite: bool = False, remove_tags: bool = True,
                  relation_depth: int = 0):
         """ Create a new writer.
 

src/osmium/file_processor.py

@@ -2,24 +2,25 @@
 #
 # This file is part of pyosmium. (https://osmcode.org/pyosmium/)
 #
-# Copyright (C) 2024 Sarah Hoffmann <[email protected]> and others.
+# Copyright (C) 2025 Sarah Hoffmann <[email protected]> and others.
 # For a full list of authors see the git log.
-from typing import Iterable, Iterator, Tuple, Any, Union, Optional, List
+from typing import Iterable, Iterator, Tuple, Union, Optional, List
 import os
 
 import osmium
 from osmium.index import LocationTable
 from osmium.io import File, FileBuffer
 from osmium.osm.types import OSMEntity
 
+
 class FileProcessor:
     """ A processor that reads an OSM file in a streaming fashion,
         optionally pre-filters the data, enhances it with geometry information,
         returning the data via an iterator.
     """
 
     def __init__(self, indata: Union[File, FileBuffer, str, 'os.PathLike[str]'],
-                 entities: osmium.osm.osm_entity_bits=osmium.osm.ALL) -> None:
+                 entities: osmium.osm.osm_entity_bits = osmium.osm.ALL) -> None:
         """ Initialise a new file processor for the given input source _indata_.
             This may either be a filename, an instance of [File](IO.md#osmium.io.File)
             or buffered data in form of a [FileBuffer](IO.md#osmium.io.FileBuffer).
@@ -54,7 +55,7 @@ def node_location_storage(self) -> Optional[LocationTable]:
         """
         return self._node_store
 
-    def with_locations(self, storage: str='flex_mem') -> 'FileProcessor':
+    def with_locations(self, storage: str = 'flex_mem') -> 'FileProcessor':
         """ Enable caching of node locations. The file processor will keep
             the coordinates of all nodes that are read from the file in
             memory and automatically enhance the node list of ways with
@@ -77,7 +78,8 @@ def with_locations(self, storage: str='flex_mem') -> 'FileProcessor':
         elif storage is None or isinstance(storage, osmium.index.LocationTable):
             self._node_store = storage
         else:
-            raise TypeError("'storage' argument must be a LocationTable or a string describing the index")
+            raise TypeError("'storage' argument must be a LocationTable "
+                            "or a string describing the index")
 
         return self
 
@@ -117,7 +119,6 @@ def with_filter(self, filt: 'osmium._osmium.HandlerLike') -> 'FileProcessor':
         self._filters.append(filt)
         return self
 
-
     def handler_for_filtered(self, handler: 'osmium._osmium.HandlerLike') -> 'FileProcessor':
         """ Set a fallback handler for object that have been filtered out.
 
@@ -214,13 +215,12 @@ def next(self, nextid: Optional[Tuple[int, int]]) -> Tuple[int, int]:
             if self.comp == nextid:
                 self.current = next(self.iter, None)
                 if self.current is None:
-                    self.comp = (100, 0) # end of file marker. larger than any ID
+                    self.comp = (100, 0)  # end of file marker. larger than any ID
                 else:
                     self.comp = (TID[self.current.type_str()], self.current.id)
             assert self.comp is not None
             return self.comp
 
-
     iters = [_CompIter(p) for p in procs]
 
     nextid = min(i.next(None) for i in iters)