diff --git a/lib/osm.cc b/lib/osm.cc
index efe23354..bef0e7c1 100644
--- a/lib/osm.cc
+++ b/lib/osm.cc
@@ -102,7 +102,8 @@ PYBIND11_MODULE(_osm, m) {
py::return_value_policy::reference_internal,
"Extend the box to include the given location. If the location "
"is invalid the box remains unchanged. If the box is invalid, it "
- "will contain only the location after the operation.")
+ "will contain only the location after the operation. "
+ "Returns a reference to itself.")
.def("extend",
(osmium::Box& (osmium::Box::*)(osmium::Box const &))
&osmium::Box::extend,
@@ -110,7 +111,8 @@ PYBIND11_MODULE(_osm, m) {
py::return_value_policy::reference_internal,
"Extend the box to include the given box. If the box to be added "
"is invalid the input box remains unchanged. If the input box is invalid, it "
- "will become equal to the box that was added.")
+ "will become equal to the box that was added. "
+ "Returns a reference to itself.")
.def("valid", &osmium::Box::valid,
"Check if the box coordinates are defined and with the usual bounds.")
.def("size", &osmium::Box::size,
diff --git a/lib/osmium.cc b/lib/osmium.cc
index 3d60c1b1..a5740576 100644
--- a/lib/osmium.cc
+++ b/lib/osmium.cc
@@ -37,7 +37,7 @@ PYBIND11_MODULE(_osmium, m) {
"Apply a chain of handlers.");
m.def("apply", [](osmium::io::Reader &rd, NodeLocationHandler &h)
{ py::gil_scoped_release release; osmium::apply(rd, h); },
- py::arg("reader"), py::arg("handler"),
+ py::arg("reader"), py::arg("node_handler"),
"Apply a chain of handlers.");
m.def("apply", [](osmium::io::Reader &rd, NodeLocationHandler &l,
BaseHandler &h)
diff --git a/setup.py b/setup.py
index 308200ec..b2f72422 100644
--- a/setup.py
+++ b/setup.py
@@ -166,6 +166,9 @@ def build_extension(self, ext):
ext_modules=[CMakeExtension('cmake_example')],
packages = ['osmium', 'osmium/osm', 'osmium/replication'],
package_dir = {'' : 'src'},
+ package_data = { 'osmium': ['py.typed', '*.pyi',
+ 'replication/_replication.pyi',
+ 'osm/_osm.pyi']},
python_requires = ">=3.6",
install_requires = ['requests'],
cmdclass=dict(build_ext=CMakeBuild, sdist=Pyosmium_sdist),
diff --git a/src/osmium/_osmium.pyi b/src/osmium/_osmium.pyi
new file mode 100644
index 00000000..9bb6a4df
--- /dev/null
+++ b/src/osmium/_osmium.pyi
@@ -0,0 +1,55 @@
+from typing import overload, ByteString, Union
+import os
+
+import osmium.index
+import osmium.io
+
+StrPath = Union[str, 'os.PathLike[str]']
+
+class InvalidLocationError(Exception): ...
+
+class NodeLocationsForWays:
+ def __init__(self, locations: osmium.index.LocationTable) -> None: ...
+ def ignore_errors(self) -> None: ...
+
+class BaseHandler: ...
+
+class SimpleHandler(BaseHandler):
+ def __init__(self) -> None: ...
+ def apply_buffer(self, buffer: Union[ByteString, str], format: str, locations: bool = ..., idx: str = ...) -> None: ...
+ def apply_file(self, filename: StrPath, locations: bool = ..., idx: str = ...) -> None: ...
+
+class MergeInputReader:
+ def __init__(self) -> None: ...
+ def add_buffer(self, buffer: Union[ByteString, str], format: str) -> int: ...
+ def add_file(self, file: str) -> int: ...
+ def apply(self, handler: BaseHandler, idx: str = ..., simplify: bool = ...) -> None: ...
+ def apply_to_reader(self, reader: osmium.io.Reader, writer: osmium.io.Writer, with_history: bool = ...) -> None: ...
+
+class WriteHandler(BaseHandler):
+ @overload
+ def __init__(self, filename: str, bufsz: int, filetype: str) -> None: ...
+ @overload
+ def __init__(self, filename: str, bufsz: int) -> None: ...
+ @overload
+ def __init__(self, filename: str) -> None: ...
+ def close(self) -> None: ...
+
+class SimpleWriter:
+ @overload
+ def __init__(self, filename: str, bufsz: int, header: osmium.io.Header) -> None: ...
+ @overload
+ def __init__(self, filename: str, bufsz: int) -> None: ...
+ @overload
+ def __init__(self, filename: str) -> None: ...
+ def add_node(self, node: object) -> None: ...
+ def add_relation(self, relation: object) -> None: ...
+ def add_way(self, way: object) -> None: ...
+ def close(self) -> None: ...
+
+@overload
+def apply(reader: osmium.io.Reader, handler: BaseHandler) -> None: ...
+@overload
+def apply(reader: osmium.io.Reader, node_handler: NodeLocationsForWays) -> None: ...
+@overload
+def apply(reader: osmium.io.Reader, node_handler: NodeLocationsForWays, handler: BaseHandler) -> None: ...
diff --git a/src/osmium/geom.pyi b/src/osmium/geom.pyi
new file mode 100644
index 00000000..c5742200
--- /dev/null
+++ b/src/osmium/geom.pyi
@@ -0,0 +1,99 @@
+from typing import ClassVar
+
+from typing import overload
+import osmium.osm
+ALL: use_nodes
+BACKWARD: direction
+FORWARD: direction
+UNIQUE: use_nodes
+
+
+class use_nodes:
+ ALL: ClassVar[use_nodes] = ...
+ UNIQUE: ClassVar[use_nodes] = ...
+ def __init__(self, value: int) -> None: ...
+ @property
+ def name(self) -> str: ...
+ @property
+ def value(self) -> int: ...
+
+class direction:
+ BACKWARD: ClassVar[direction] = ...
+ FORWARD: ClassVar[direction] = ...
+ def __init__(self, value: int) -> None: ...
+ @property
+ def name(self) -> str: ...
+ @property
+ def value(self) -> int: ...
+
+class Coordinates:
+ @overload
+ def __init__(self) -> None: ...
+ @overload
+ def __init__(self, cx: float, cy: float) -> None: ...
+ @overload
+ def __init__(self, location: osmium.osm.Location) -> None: ...
+ def valid(self) -> bool: ...
+ @property
+ def x(self) -> float: ...
+ @property
+ def y(self) -> float: ...
+
+
+class GeoJSONFactory:
+ def __init__(self) -> None: ...
+ @overload
+ def create_linestring(self, list: osmium.osm.WayNodeList, use_nodes: use_nodes = ..., direction: direction = ...) -> str: ...
+ @overload
+ def create_linestring(self, way: osmium.osm.Way, use_nodes: use_nodes = ..., direction: direction = ...) -> str: ...
+ def create_multipolygon(self, area: osmium.osm.Area) -> str: ...
+ @overload
+ def create_point(self, location: osmium.osm.Location) -> str: ...
+ @overload
+ def create_point(self, node: osmium.osm.Node) -> str: ...
+ @overload
+ def create_point(self, ref: osmium.osm.NodeRef) -> str: ...
+ @property
+ def epsg(self) -> int: ...
+ @property
+ def proj_string(self) -> str: ...
+
+class WKBFactory:
+ def __init__(self) -> None: ...
+ @overload
+ def create_linestring(self, list: osmium.osm.WayNodeList, use_nodes: use_nodes = ..., direction: direction = ...) -> str: ...
+ @overload
+ def create_linestring(self, way: osmium.osm.Way, use_nodes: use_nodes = ..., direction: direction = ...) -> str: ...
+ def create_multipolygon(self, area: osmium.osm.Area) -> str: ...
+ @overload
+ def create_point(self, location: osmium.osm.Location) -> str: ...
+ @overload
+ def create_point(self, node: osmium.osm.Node) -> str: ...
+ @overload
+ def create_point(self, ref: osmium.osm.NodeRef) -> str: ...
+ @property
+ def epsg(self) -> int: ...
+ @property
+ def proj_string(self) -> str: ...
+
+class WKTFactory:
+ def __init__(self) -> None: ...
+ @overload
+ def create_linestring(self, list: osmium.osm._osm.WayNodeList, use_nodes: use_nodes = ..., direction: direction = ...) -> str: ...
+ @overload
+ def create_linestring(self, way: osmium.osm._osm.Way, use_nodes: use_nodes = ..., direction: direction = ...) -> str: ...
+ def create_multipolygon(self, area: osmium.osm._osm.Area) -> str: ...
+ @overload
+ def create_point(self, location: osmium.osm._osm.Location) -> str: ...
+ @overload
+ def create_point(self, node: osmium.osm._osm.Node) -> str: ...
+ @overload
+ def create_point(self, ref: osmium.osm._osm.NodeRef) -> str: ...
+ @property
+ def epsg(self) -> int: ...
+ @property
+ def proj_string(self) -> str: ...
+
+def haversine_distance(list: osmium.osm.WayNodeList) -> float: ...
+def lonlat_to_mercator(coordinate: Coordinates) -> Coordinates: ...
+def mercator_to_lonlat(coordinate: Coordinates) -> Coordinates: ...
diff --git a/src/osmium/helper.py b/src/osmium/helper.py
index eec5f14e..24db5e8a 100644
--- a/src/osmium/helper.py
+++ b/src/osmium/helper.py
@@ -1,6 +1,17 @@
+from typing import Optional, Callable, TypeVar
+
from osmium._osmium import SimpleHandler
+from osmium.osm import Node, Way, Relation, Area, Changeset
+
+T = TypeVar('T')
+HandlerFunc = Optional[Callable[[T], None]]
+
-def make_simple_handler(node=None, way=None, relation=None, area=None, changeset=None):
+def make_simple_handler(node: HandlerFunc[Node] = None,
+ way: HandlerFunc[Way] = None,
+ relation: HandlerFunc[Relation] = None,
+ area: HandlerFunc[Area] = None,
+ changeset: HandlerFunc[Changeset] = None) -> SimpleHandler:
""" Convenience function that creates a `SimpleHandler` from a set of
callback functions. Each of the parameters takes an optional callable
that must expect a single positional parameter with the object being
@@ -10,14 +21,14 @@ class __HandlerWithCallbacks(SimpleHandler):
pass
if node is not None:
- __HandlerWithCallbacks.node = staticmethod(node)
+ setattr(__HandlerWithCallbacks, "node", staticmethod(node))
if way is not None:
- __HandlerWithCallbacks.way = staticmethod(way)
+ setattr(__HandlerWithCallbacks, "way", staticmethod(way))
if relation is not None:
- __HandlerWithCallbacks.relation = staticmethod(relation)
+ setattr(__HandlerWithCallbacks, "relation", staticmethod(relation))
if area is not None:
- __HandlerWithCallbacks.area = staticmethod(area)
+ setattr(__HandlerWithCallbacks, "area", staticmethod(area))
if changeset is not None:
- __HandlerWithCallbacks.changeset = staticmethod(changeset)
+ setattr(__HandlerWithCallbacks, "changeset", staticmethod(changeset))
return __HandlerWithCallbacks()
diff --git a/src/osmium/index.pyi b/src/osmium/index.pyi
new file mode 100644
index 00000000..e778fb3d
--- /dev/null
+++ b/src/osmium/index.pyi
@@ -0,0 +1,12 @@
+from typing import List
+
+import osmium.osm
+
+class LocationTable:
+ def clear(self) -> None: ...
+ def get(self, id: int) -> osmium.osm.Location: ...
+ def set(self, id: int, loc: osmium.osm.Location) -> None: ...
+ def used_memory(self) -> int: ...
+
+def create_map(map_type: str) -> LocationTable: ...
+def map_types() -> List[str]: ...
diff --git a/src/osmium/io.pyi b/src/osmium/io.pyi
new file mode 100644
index 00000000..5cba6247
--- /dev/null
+++ b/src/osmium/io.pyi
@@ -0,0 +1,41 @@
+from typing import Any
+
+from typing import overload
+
+import osmium.osm
+
+class File:
+ has_multiple_object_versions: bool
+ @overload
+ def __init__(self, filename: str) -> None: ...
+ @overload
+ def __init__(self, filename: str, format: str) -> None: ...
+ def parse_format(self, arg0: str) -> None: ...
+
+class Header:
+ has_multiple_object_versions: bool
+ def __init__(self) -> None: ...
+ def add_box(self, box: osmium.osm.Box) -> Header: ...
+ def box(self) -> osmium.osm.Box: ...
+ def get(self, key: str, default: str = ...) -> str: ...
+ def set(self, key: str, value: str) -> None: ...
+
+class Reader:
+ @overload
+ def __init__(self, filename: str) -> None: ...
+ @overload
+ def __init__(self, filename: str, types: osmium.osm.osm_entity_bits) -> None: ...
+ def close(self) -> None: ...
+ def eof(self) -> bool: ...
+ def header(self) -> Header: ...
+
+class Writer:
+ @overload
+ def __init__(self, filename: str) -> None: ...
+ @overload
+ def __init__(self, ffile: File) -> None: ...
+ @overload
+ def __init__(self, filename: str, header: Header) -> None: ...
+ @overload
+ def __init__(self, ffile: File, header: Header) -> None: ...
+ def close(self) -> int: ...
diff --git a/src/osmium/osm/__init__.py b/src/osmium/osm/__init__.py
index eb7e8afd..89ffdbf6 100644
--- a/src/osmium/osm/__init__.py
+++ b/src/osmium/osm/__init__.py
@@ -1,43 +1,24 @@
-import osmium.osm.mutable
+from typing import Any, Callable, Sequence
+
+from osmium.osm.mutable import create_mutable_node, create_mutable_way, create_mutable_relation
from ._osm import *
-def create_mutable_node(node, **args):
- """ Create a mutable node replacing the properties given in the
- named parameters. Note that this function only creates a shallow
- copy which is still bound to the scope of the original object.
- """
- return osmium.osm.mutable.Node(base=node, **args)
-
-def create_mutable_way(way, **args):
- """ Create a mutable way replacing the properties given in the
- named parameters. Note that this function only creates a shallow
- copy which is still bound to the scope of the original object.
- """
- return osmium.osm.mutable.Way(base=way, **args)
-
-def create_mutable_relation(rel, **args):
- """ Create a mutable relation replacing the properties given in the
- named parameters. Note that this function only creates a shallow
- copy which is still bound to the scope of the original object.
- """
- return osmium.osm.mutable.Relation(base=rel, **args)
-
-Node.replace = create_mutable_node
-Way.replace = create_mutable_way
-Relation.replace = create_mutable_relation
-
-def _make_repr(*attr_list):
+setattr(Node, 'replace', create_mutable_node)
+setattr(Way, 'replace', create_mutable_way)
+setattr(Relation, 'replace', create_mutable_relation)
+
+def _make_repr(*attrs: str) -> Callable[[object], str]:
fmt_string = 'osmium.osm.{0}('\
- + ', '.join([f'{x}={{1.{x}!r}}' for x in attr_list])\
+ + ', '.join([f'{x}={{1.{x}!r}}' for x in attrs])\
+ ')'
return lambda o: fmt_string.format(o.__class__.__name__, o)
-def _list_repr(obj):
+def _list_repr(obj: Sequence[Any]) -> str:
return 'osmium.osm.{}([{}])'.format(obj.__class__.__name__,
', '.join(map(repr, obj)))
-def _list_elipse(obj):
+def _list_elipse(obj: Sequence[Any]) -> str:
objects = ','.join(map(str, obj))
if len(objects) > 50:
objects = objects[:47] + '...'
diff --git a/src/osmium/osm/_osm.pyi b/src/osmium/osm/_osm.pyi
new file mode 100644
index 00000000..5c0528ab
--- /dev/null
+++ b/src/osmium/osm/_osm.pyi
@@ -0,0 +1,202 @@
+from typing import ClassVar, Iterator, Tuple, Optional, Any
+
+from typing import overload
+import datetime
+
+import osmium.osm.mutable
+
+ALL: osm_entity_bits
+AREA: osm_entity_bits
+CHANGESET: osm_entity_bits
+NODE: osm_entity_bits
+NOTHING: osm_entity_bits
+OBJECT: osm_entity_bits
+RELATION: osm_entity_bits
+WAY: osm_entity_bits
+
+
+class Location:
+ @overload
+ def __init__(self) -> None: ...
+ @overload
+ def __init__(self, lon: float, lat: float) -> None: ...
+ def lat_without_check(self) -> float: ...
+ def lon_without_check(self) -> float: ...
+ def valid(self) -> bool: ...
+ @property
+ def lat(self) -> float: ...
+ @property
+ def lon(self) -> float: ...
+ @property
+ def x(self) -> int: ...
+ @property
+ def y(self) -> int: ...
+
+class Box:
+ @overload
+ def __init__(self, minx: float, miny: float, maxx: float, maxy: float) -> None: ...
+ @overload
+ def __init__(self, bottom_left: Location, top_right: Location) -> None: ...
+ def contains(self, location: Location) -> bool: ...
+ @overload
+ def extend(self, location: Location) -> Box: ...
+ @overload
+ def extend(self, box: Box) -> Box: ...
+ def size(self) -> float: ...
+ def valid(self) -> bool: ...
+ @property
+ def bottom_left(self) -> Location: ...
+ @property
+ def top_right(self) -> Location: ...
+
+class TagIterator:
+ def __iter__(self) -> TagIterator: ...
+ def __len__(self) -> int: ...
+ def __next__(self) -> str: ...
+
+class Tag:
+ def __iter__(self) -> TagIterator: ...
+ @property
+ def k(self) -> str: ...
+ @property
+ def v(self) -> str: ...
+
+class TagList:
+ @overload
+ def get(self, key: str, default: str) -> str: ...
+ @overload
+ def get(self, key: str) -> Optional[str]: ...
+ def __contains__(self, key: str) -> bool: ...
+ def __getitem__(self, key: str) -> str: ...
+ def __iter__(self) -> Iterator[Tag]: ...
+ def __len__(self) -> int: ...
+
+class NodeRef:
+ @property
+ def lat(self) -> float: ...
+ @property
+ def location(self) -> Location: ...
+ @property
+ def lon(self) -> float: ...
+ @property
+ def ref(self) -> int: ...
+ @property
+ def x(self) -> int: ...
+ @property
+ def y(self) -> int: ...
+
+class RelationMember:
+ @property
+ def ref(self) -> int: ...
+ @property
+ def role(self) -> str: ...
+ @property
+ def type(self) -> str: ...
+
+class RelationMemberList:
+ def __iter__(self) -> Iterator[RelationMember]: ...
+ def __len__(self) -> int: ...
+
+class NodeRefList:
+ def ends_have_same_id(self) -> bool: ...
+ def ends_have_same_location(self) -> bool: ...
+ def is_closed(self) -> bool: ...
+ def __getitem__(self, idx: int) -> NodeRef: ...
+ def __iter__(self) -> Iterator[NodeRef]: ...
+ def __len__(self) -> int: ...
+
+class WayNodeList(NodeRefList):
+ pass
+
+class OuterRing(NodeRefList):
+ pass
+
+class InnerRing(NodeRefList):
+ pass
+
+class InnerRingIterator:
+ def __iter__(self) -> Iterator[InnerRing]: ...
+
+class OSMObject:
+ def positive_id(self) -> int: ...
+ def user_is_anonymous(self) -> bool: ...
+ @property
+ def changeset(self) -> int: ...
+ @property
+ def deleted(self) -> bool: ...
+ @property
+ def id(self) -> int: ...
+ @property
+ def tags(self) -> TagList: ...
+ @property
+ def timestamp(self) -> datetime.datetime: ...
+ @property
+ def uid(self) -> int: ...
+ @property
+ def user(self) -> str: ...
+ @property
+ def version(self) -> int: ...
+ @property
+ def visible(self) -> bool: ...
+
+class Node(OSMObject):
+ @property
+ def location(self) -> Location: ...
+ def replace(**args: Any) -> osmium.osm.mutable.Node: ...
+
+class Way(OSMObject):
+ def ends_have_same_id(self) -> bool: ...
+ def ends_have_same_location(self) -> bool: ...
+ def is_closed(self) -> bool: ...
+ @property
+ def nodes(self) -> WayNodeList: ...
+ def replace(**args: Any) -> osmium.osm.mutable.Way: ...
+
+class Relation(OSMObject):
+ @property
+ def members(self) -> RelationMemberList: ...
+ def replace(**args: Any) -> osmium.osm.mutable.Relation: ...
+
+class Area(OSMObject):
+ def from_way(self) -> bool: ...
+ def inner_rings(self, outer_ring: OuterRing) -> InnerRingIterator: ...
+ def is_multipolygon(self) -> bool: ...
+ def num_rings(self) -> Tuple[int,int]: ...
+ def orig_id(self) -> int: ...
+ def outer_rings(self) -> Iterator[OuterRing]: ...
+
+class Changeset:
+ def user_is_anonymous(self) -> bool: ...
+ @property
+ def bounds(self) -> Box: ...
+ @property
+ def closed_at(self) -> datetime.datetime: ...
+ @property
+ def created_at(self) -> datetime.datetime: ...
+ @property
+ def id(self) -> int: ...
+ @property
+ def num_changes(self) -> int: ...
+ @property
+ def open(self) -> bool: ...
+ @property
+ def tags(self) -> TagList: ...
+ @property
+ def uid(self) -> int: ...
+ @property
+ def user(self) -> str: ...
+
+class osm_entity_bits:
+ ALL: ClassVar[osm_entity_bits] = ...
+ AREA: ClassVar[osm_entity_bits] = ...
+ CHANGESET: ClassVar[osm_entity_bits] = ...
+ NODE: ClassVar[osm_entity_bits] = ...
+ NOTHING: ClassVar[osm_entity_bits] = ...
+ OBJECT: ClassVar[osm_entity_bits] = ...
+ RELATION: ClassVar[osm_entity_bits] = ...
+ WAY: ClassVar[osm_entity_bits] = ...
+ def __init__(self, value: int) -> None: ...
+ @property
+ def name(self) -> str: ...
+ @property
+ def value(self) -> int: ...
diff --git a/src/osmium/osm/mutable.py b/src/osmium/osm/mutable.py
index 992eae85..386cd43f 100644
--- a/src/osmium/osm/mutable.py
+++ b/src/osmium/osm/mutable.py
@@ -1,4 +1,21 @@
-class OSMObject(object):
+from typing import Optional, Union, Any, Mapping, Sequence, Tuple, TYPE_CHECKING
+from datetime import datetime
+
+if TYPE_CHECKING:
+ import osmium.osm
+
+ OSMObjectLike = Union['OSMObject', osmium.osm.OSMObject]
+ NodeLike = Union[Node, osmium.osm.Node]
+ WayLike = Union[Way, osmium.osm.Way]
+ RelationLike = Union[Relation, osmium.osm.Relation]
+
+ TagSequence = Union[osmium.osm.TagList, Mapping[str, str], Sequence[Tuple[str, str]]]
+ LocationLike = Union[osmium.osm.Location, Tuple[float, float]]
+ NodeSequence = Union[osmium.osm.NodeRefList, Sequence[Union[osmium.osm.NodeRef, int]]]
+ MemberSequence = Union[osmium.osm.RelationMemberList,
+ Sequence[Union[osmium.osm.RelationMember, Tuple[str, int, str]]]]
+
+class OSMObject:
"""Mutable version of ``osmium.osm.OSMObject``. It exposes the following
attributes ``id``, ``version``, ``visible``, ``changeset``, ``timestamp``,
``uid`` and ``tags``. Timestamps may be strings or datetime objects.
@@ -9,8 +26,11 @@ class OSMObject(object):
will be initialised first from the attributes of this base object.
"""
- def __init__(self, base=None, id=None, version=None, visible=None, changeset=None,
- timestamp=None, uid=None, tags=None, user=None):
+ def __init__(self, base: Optional['OSMObjectLike'] = None,
+ id: Optional[int] = None, version: Optional[int] = None,
+ visible: Optional[bool] = None, changeset: Optional[int] = None,
+ timestamp: Optional[datetime] = None, uid: Optional[int] = None,
+ tags: Optional['TagSequence'] = None, user: Optional[str] = None) -> None:
if base is None:
self.id = id
self.version = version
@@ -37,7 +57,9 @@ class Node(OSMObject):
may either be an `osmium.osm.Location` or a tuple of lon/lat coordinates.
"""
- def __init__(self, base=None, location=None, **attrs):
+ def __init__(self, base: Optional['NodeLike'] = None,
+ location: Optional['LocationLike'] = None,
+ **attrs: Any) -> None:
OSMObject.__init__(self, base=base, **attrs)
if base is None:
self.location = location
@@ -52,7 +74,8 @@ class Way(OSMObject):
``osmium.osm.NodeRef`` or simple node ids.
"""
- def __init__(self, base=None, nodes=None, **attrs):
+ def __init__(self, base: Optional['WayLike'] = None,
+ nodes: Optional['NodeSequence'] = None, **attrs: Any) -> None:
OSMObject.__init__(self, base=base, **attrs)
if base is None:
self.nodes = nodes
@@ -67,9 +90,33 @@ class Relation(OSMObject):
member type should be a single character 'n', 'w' or 'r'.
"""
- def __init__(self, base=None, members=None, **attrs):
+ def __init__(self, base: Optional['RelationLike'] = None,
+ members: Optional['MemberSequence'] = None, **attrs: Any) -> None:
OSMObject.__init__(self, base=base, **attrs)
if base is None:
self.members = members
else:
self.members = members if members is not None else base.members
+
+
+def create_mutable_node(node: 'NodeLike', **args: Any) -> Node:
+ """ Create a mutable node replacing the properties given in the
+ named parameters. Note that this function only creates a shallow
+ copy which is still bound to the scope of the original object.
+ """
+ return Node(base=node, **args)
+
+def create_mutable_way(way: 'WayLike', **args: Any) -> Way:
+ """ Create a mutable way replacing the properties given in the
+ named parameters. Note that this function only creates a shallow
+ copy which is still bound to the scope of the original object.
+ """
+ return Way(base=way, **args)
+
+def create_mutable_relation(rel: 'RelationLike', **args: Any) -> Relation:
+ """ Create a mutable relation replacing the properties given in the
+ named parameters. Note that this function only creates a shallow
+ copy which is still bound to the scope of the original object.
+ """
+ return Relation(base=rel, **args)
+
diff --git a/src/osmium/py.typed b/src/osmium/py.typed
new file mode 100644
index 00000000..e69de29b
diff --git a/src/osmium/replication/_replication.pyi b/src/osmium/replication/_replication.pyi
new file mode 100644
index 00000000..9f5a8532
--- /dev/null
+++ b/src/osmium/replication/_replication.pyi
@@ -0,0 +1,3 @@
+import datetime
+
+def newest_change_from_file(filename: str) -> datetime.datetime: ...
diff --git a/src/osmium/replication/server.py b/src/osmium/replication/server.py
index e03ffd73..f17eba4b 100644
--- a/src/osmium/replication/server.py
+++ b/src/osmium/replication/server.py
@@ -1,6 +1,6 @@
""" Helper functions to communicate with replication servers.
"""
-
+from typing import NamedTuple, Optional, Any, Iterator, cast, Mapping, Tuple
import requests
import urllib.request as urlrequest
from urllib.error import URLError
@@ -9,7 +9,7 @@
from contextlib import contextmanager
from math import ceil
-from osmium import MergeInputReader
+from osmium import MergeInputReader, BaseHandler
from osmium import io as oio
from osmium import version
@@ -18,8 +18,14 @@
LOG = logging.getLogger('pyosmium')
LOG.addHandler(logging.NullHandler())
-OsmosisState = namedtuple('OsmosisState', ['sequence', 'timestamp'])
-DownloadResult = namedtuple('DownloadResult', ['id', 'reader', 'newest'])
+class OsmosisState(NamedTuple):
+ sequence: int
+ timestamp: dt.datetime
+
+class DownloadResult(NamedTuple):
+ id: int
+ reader: MergeInputReader
+ newest: int
class ReplicationServer:
""" Represents a connection to a server that publishes replication data.
@@ -30,37 +36,37 @@ class ReplicationServer:
internally keeps a connection to the server making downloads faster.
"""
- def __init__(self, url, diff_type='osc.gz'):
+ def __init__(self, url: str, diff_type: str = 'osc.gz') -> None:
self.baseurl = url
self.diff_type = diff_type
- self.session = None
+ self.session: Optional[requests.Session] = None
- def close(self):
+ def close(self) -> None:
""" Close any open connection to the replication server.
"""
if self.session is not None:
self.session.close()
self.session = None
- def __enter__(self):
+ def __enter__(self) -> 'ReplicationServer':
self.session = requests.Session()
return self
- def __exit__(self, exc_type, exc_value, traceback):
+ def __exit__(self, exc_type: Any, exc_value: Any, traceback: Any) -> None:
self.close()
- def make_request(self, url):
+ def make_request(self, url: str) -> urlrequest.Request:
headers = {"User-Agent" : "pyosmium/{}".format(version.pyosmium_release)}
return urlrequest.Request(url, headers=headers)
- def open_url(self, url):
+ def open_url(self, url: urlrequest.Request) -> Any:
""" Download a resource from the given URL and return a byte sequence
of the content.
This method has no support for cookies or any special authentication
methods. If you need these, you have to provide your own custom URL
opener. Overwrite open_url() with a method that receives an
- urllib.Request object and returns a ByteIO-like object or a
+ urlrequest.Request object and returns a ByteIO-like object or a
requests.Response.
Example::
@@ -79,14 +85,14 @@ def open_url(self, url):
return self.session.get(url.get_full_url(), headers=headers, stream=True)
@contextmanager
- def _get_url_with_session():
+ def _get_url_with_session() -> Iterator[requests.Response]:
with requests.Session() as session:
request = session.get(url.get_full_url(), headers=headers, stream=True)
yield request
return _get_url_with_session()
- def collect_diffs(self, start_id, max_size=1024):
+ def collect_diffs(self, start_id: int, max_size: int = 1024) -> Optional[DownloadResult]:
""" Create a MergeInputReader and download diffs starting with sequence
id `start_id` into it. `max_size`
restricts the number of diffs that are downloaded. The download
@@ -131,7 +137,9 @@ def collect_diffs(self, start_id, max_size=1024):
return DownloadResult(current_id - 1, rd, newest.sequence)
- def apply_diffs(self, handler, start_id, max_size=1024, idx="", simplify=True):
+ def apply_diffs(self, handler: BaseHandler, start_id: int,
+ max_size: int = 1024, idx: str = "",
+ simplify: bool = True) -> Optional[int]:
""" Download diffs starting with sequence id `start_id`, merge them
together and then apply them to handler `handler`. `max_size`
restricts the number of diffs that are downloaded. The download
@@ -165,9 +173,11 @@ def apply_diffs(self, handler, start_id, max_size=1024, idx="", simplify=True):
return diffs.id
- def apply_diffs_to_file(self, infile, outfile, start_id, max_size=1024,
- set_replication_header=True, extra_headers=None,
- outformat=None):
+ def apply_diffs_to_file(self, infile: str, outfile: str,
+ start_id: int, max_size: int = 1024,
+ set_replication_header: bool = True,
+ extra_headers: Optional[Mapping[str, str]] = None,
+ outformat: Optional[str] = None) -> Optional[Tuple[int, int]]:
""" Download diffs starting with sequence id `start_id`, merge them
with the data from the OSM file named `infile` and write the result
into a file with the name `outfile`. The output file must not yet
@@ -230,7 +240,8 @@ def apply_diffs_to_file(self, infile, outfile, start_id, max_size=1024,
return (diffs.id, diffs.newest)
- def timestamp_to_sequence(self, timestamp, balanced_search=False):
+ def timestamp_to_sequence(self, timestamp: dt.datetime,
+ balanced_search: bool = False) -> Optional[int]:
""" Get the sequence number of the replication file that contains the
given timestamp. The search algorithm is optimised for replication
servers that publish updates in regular intervals. For servers
@@ -312,7 +323,7 @@ def timestamp_to_sequence(self, timestamp, balanced_search=False):
return lower.sequence
- def get_state_info(self, seq=None, retries=2):
+ def get_state_info(self, seq: Optional[int] = None, retries: int = 2) -> Optional[OsmosisState]:
""" Downloads and returns the state information for the given
sequence. If the download is successful, a namedtuple with
`sequence` and `timestamp` is returned, otherwise the function
@@ -359,7 +370,7 @@ def get_state_info(self, seq=None, retries=2):
return None
- def get_diff_block(self, seq):
+ def get_diff_block(self, seq: int) -> str:
""" Downloads the diff with the given sequence number and returns
it as a byte sequence. Throws a :code:`urllib.error.HTTPError`
if the file cannot be downloaded.
@@ -367,13 +378,13 @@ def get_diff_block(self, seq):
with self.open_url(self.make_request(self.get_diff_url(seq))) as resp:
if hasattr(resp, 'content'):
# generated by requests
- return resp.content
+ return cast(str, resp.content)
# generated by urllib.request
- return resp.read()
+ return cast(str, resp.read())
- def get_state_url(self, seq):
+ def get_state_url(self, seq: Optional[int]) -> str:
""" Returns the URL of the state.txt files for a given sequence id.
If seq is `None` the URL for the latest state info is returned,
@@ -387,7 +398,7 @@ def get_state_url(self, seq):
(self.baseurl, seq / 1000000, (seq % 1000000) / 1000, seq % 1000)
- def get_diff_url(self, seq):
+ def get_diff_url(self, seq: int) -> str:
""" Returns the URL to the diff file for the given sequence id.
"""
return '%s/%03i/%03i/%03i.%s' % \
diff --git a/src/osmium/replication/utils.py b/src/osmium/replication/utils.py
index c0fc2195..9623251d 100644
--- a/src/osmium/replication/utils.py
+++ b/src/osmium/replication/utils.py
@@ -1,5 +1,5 @@
""" Helper functions for change file handling. """
-
+from typing import NamedTuple, Optional
import logging
import datetime as dt
from collections import namedtuple
@@ -8,10 +8,13 @@
LOG = logging.getLogger('pyosmium')
-ReplicationHeader = namedtuple('ReplicationHeader',
- ['url', 'sequence', 'timestamp'])
+class ReplicationHeader(NamedTuple):
+ url: Optional[str]
+ sequence: Optional[int]
+ timestamp: Optional[dt.datetime]
+
-def get_replication_header(fname):
+def get_replication_header(fname: str) -> ReplicationHeader:
""" Scans the given file for an Osmosis replication header. It returns
a namedtuple with `url`, `sequence` and `timestamp`. Each or all fields
may be None, if the piece of information is not avilable. If any of
@@ -24,20 +27,21 @@ def get_replication_header(fname):
r = oreader(fname, NOTHING)
h = r.header()
- ts = h.get("osmosis_replication_timestamp")
- url = h.get("osmosis_replication_base_url")
+ tsstr = h.get("osmosis_replication_timestamp")
+ url: Optional[str] = h.get("osmosis_replication_base_url")
- if url or ts:
+ if url or tsstr:
LOG.debug("Replication information found in OSM file header.")
if url:
LOG.debug("Replication URL: %s", url)
# the sequence ID is only considered valid, if an URL is given
- seq = h.get("osmosis_replication_sequence_number")
- if seq:
- LOG.debug("Replication sequence: %s", seq)
+ seqstr = h.get("osmosis_replication_sequence_number")
+ seq: Optional[int]
+ if seqstr:
+ LOG.debug("Replication sequence: %s", seqstr)
try:
- seq = int(seq)
+ seq = int(seqstr)
if seq < 0:
LOG.warning("Sequence id '%d' in OSM file header is negative. Ignored.", seq)
seq = None
@@ -50,10 +54,10 @@ def get_replication_header(fname):
url = None
seq = None
- if ts:
- LOG.debug("Replication timestamp: %s", ts)
+ if tsstr:
+ LOG.debug("Replication timestamp: %s", tsstr)
try:
- ts = dt.datetime.strptime(ts, "%Y-%m-%dT%H:%M:%SZ")
+ ts = dt.datetime.strptime(tsstr, "%Y-%m-%dT%H:%M:%SZ")
ts = ts.replace(tzinfo=dt.timezone.utc)
except ValueError: