Skip to content

Commit a4d2f56

Browse files
lonvialonvia
committed
add parameter for end date/ID to pyosmium_up_to_date
Also moves some code that is shared between the two tools into a separate file.
1 parent a2f1606 commit a4d2f56

File tree

6 files changed

+369
-155
lines changed

6 files changed

+369
-155
lines changed

‎pyproject.toml‎

Lines changed: 3 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -101,3 +101,6 @@ include = ['/src/**/*.py',
101101
'/contrib/protozero/LICENSE',
102102
'/contrib/protozero/README.md',
103103
]
104+
105+
[tool.pytest.ini_options]
106+
log_cli = false

‎src/osmium/tools/common.py‎

Lines changed: 92 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,92 @@
1+
# SPDX-License-Identifier: BSD-2-Clause
2+
#
3+
# This file is part of pyosmium. (https://osmcode.org/pyosmium/)
4+
#
5+
# Copyright (C) 2025 Sarah Hoffmann <[email protected]> and others.
6+
# For a full list of authors see the git log.
7+
from typing import Optional
8+
import logging
9+
from dataclasses import dataclass
10+
import datetime as dt
11+
from argparse import ArgumentTypeError
12+
13+
from ..replication import newest_change_from_file
14+
from ..replication.server import ReplicationServer
15+
from ..replication.utils import get_replication_header
16+
17+
18+
log = logging.getLogger()
19+
20+
21+
@dataclass
22+
class ReplicationStart:
23+
""" Represents the point where changeset download should begin.
24+
"""
25+
date: Optional[dt.datetime] = None
26+
seq_id: Optional[int] = None
27+
source: Optional[str] = None
28+
29+
def get_sequence(self, svr: ReplicationServer) -> Optional[int]:
30+
if self.seq_id is not None:
31+
log.debug("Using given sequence ID %d" % self.seq_id)
32+
return self.seq_id + 1
33+
34+
assert self.date is not None
35+
log.debug("Looking up sequence ID for timestamp %s" % self.date)
36+
return svr.timestamp_to_sequence(self.date)
37+
38+
def get_end_sequence(self, svr: ReplicationServer) -> Optional[int]:
39+
if self.seq_id is not None:
40+
log.debug("Using end sequence ID %d" % self.seq_id)
41+
return self.seq_id
42+
43+
assert self.date is not None
44+
log.debug("Looking up end sequence ID for timestamp %s" % self.date)
45+
return svr.timestamp_to_sequence(self.date)
46+
47+
@staticmethod
48+
def from_id(idstr: str) -> 'ReplicationStart':
49+
try:
50+
seq_id = int(idstr)
51+
except ValueError:
52+
raise ArgumentTypeError("Sequence id '%s' is not a number" % idstr)
53+
54+
if seq_id < -1:
55+
raise ArgumentTypeError("Sequence id '%s' is negative" % idstr)
56+
57+
return ReplicationStart(seq_id=seq_id)
58+
59+
@staticmethod
60+
def from_date(datestr: str) -> 'ReplicationStart':
61+
try:
62+
date = dt.datetime.strptime(datestr, "%Y-%m-%dT%H:%M:%SZ")
63+
date = date.replace(tzinfo=dt.timezone.utc)
64+
except ValueError:
65+
raise ArgumentTypeError(
66+
"Date needs to be in ISO8601 format (e.g. 2015-12-24T08:08:08Z).")
67+
68+
return ReplicationStart(date=date)
69+
70+
@staticmethod
71+
def from_osm_file(fname: str, ignore_headers: bool) -> 'ReplicationStart':
72+
if ignore_headers:
73+
ts = None
74+
seq = None
75+
url = None
76+
else:
77+
try:
78+
(url, seq, ts) = get_replication_header(fname)
79+
except RuntimeError as e:
80+
raise ArgumentTypeError(e)
81+
82+
if ts is None and seq is None:
83+
log.debug("OSM file has no replication headers. Looking for newest OSM object.")
84+
try:
85+
ts = newest_change_from_file(fname)
86+
except RuntimeError as e:
87+
raise ArgumentTypeError(e)
88+
89+
if ts is None:
90+
raise ArgumentTypeError("OSM file does not seem to contain valid data.")
91+
92+
return ReplicationStart(seq_id=seq, date=ts, source=url)

‎src/osmium/tools/pyosmium_get_changes.py‎

Lines changed: 7 additions & 84 deletions
Original file line numberDiff line numberDiff line change
@@ -29,97 +29,20 @@
2929
However, it can read cookies from a Netscape-style cookie jar file, send these
3030
cookies to the server and will save received cookies to the jar file.
3131
"""
32-
from typing import Optional, List
32+
from typing import List
3333
import sys
3434
import logging
3535
from textwrap import dedent as msgfmt
36-
37-
from argparse import ArgumentParser, RawDescriptionHelpFormatter, ArgumentTypeError
38-
import datetime as dt
39-
from dataclasses import dataclass
36+
from argparse import ArgumentParser, RawDescriptionHelpFormatter
4037
import http.cookiejar
4138

42-
from osmium.replication import server as rserv
43-
from osmium.replication import newest_change_from_file
44-
from osmium.replication.utils import get_replication_header
45-
from osmium.version import pyosmium_release
46-
from osmium import SimpleWriter
47-
48-
log = logging.getLogger()
39+
from ..replication import server as rserv
40+
from ..version import pyosmium_release
41+
from .. import SimpleWriter
42+
from .common import ReplicationStart
4943

5044

51-
@dataclass
52-
class ReplicationStart:
53-
""" Represents the point where changeset download should begin.
54-
"""
55-
date: Optional[dt.datetime] = None
56-
seq_id: Optional[int] = None
57-
source: Optional[str] = None
58-
59-
def get_sequence(self, svr: rserv.ReplicationServer) -> Optional[int]:
60-
if self.seq_id is not None:
61-
log.debug("Using given sequence ID %d" % self.seq_id)
62-
return self.seq_id + 1
63-
64-
assert self.date is not None
65-
log.debug("Looking up sequence ID for timestamp %s" % self.date)
66-
return svr.timestamp_to_sequence(self.date)
67-
68-
def get_end_sequence(self, svr: rserv.ReplicationServer) -> Optional[int]:
69-
if self.seq_id is not None:
70-
log.debug("Using end sequence ID %d" % self.seq_id)
71-
return self.seq_id
72-
73-
assert self.date is not None
74-
log.debug("Looking up end sequence ID for timestamp %s" % self.date)
75-
return svr.timestamp_to_sequence(self.date)
76-
77-
@staticmethod
78-
def from_id(idstr: str) -> 'ReplicationStart':
79-
try:
80-
seq_id = int(idstr)
81-
except ValueError:
82-
raise ArgumentTypeError("Sequence id '%s' is not a number" % idstr)
83-
84-
if seq_id < -1:
85-
raise ArgumentTypeError("Sequence id '%s' is negative" % idstr)
86-
87-
return ReplicationStart(seq_id=seq_id)
88-
89-
@staticmethod
90-
def from_date(datestr: str) -> 'ReplicationStart':
91-
try:
92-
date = dt.datetime.strptime(datestr, "%Y-%m-%dT%H:%M:%SZ")
93-
date = date.replace(tzinfo=dt.timezone.utc)
94-
except ValueError:
95-
raise ArgumentTypeError(
96-
"Date needs to be in ISO8601 format (e.g. 2015-12-24T08:08:08Z).")
97-
98-
return ReplicationStart(date=date)
99-
100-
@staticmethod
101-
def from_osm_file(fname: str, ignore_headers: bool) -> 'ReplicationStart':
102-
if ignore_headers:
103-
ts = None
104-
seq = None
105-
url = None
106-
else:
107-
try:
108-
(url, seq, ts) = get_replication_header(fname)
109-
except RuntimeError as e:
110-
raise ArgumentTypeError(e)
111-
112-
if ts is None and seq is None:
113-
log.debug("OSM file has no replication headers. Looking for newest OSM object.")
114-
try:
115-
ts = newest_change_from_file(fname)
116-
except RuntimeError as e:
117-
raise ArgumentTypeError(e)
118-
119-
if ts is None:
120-
raise ArgumentTypeError("OSM file does not seem to contain valid data.")
121-
122-
return ReplicationStart(seq_id=seq, date=ts, source=url)
45+
log = logging.getLogger()
12346

12447

12548
def write_end_sequence(fname: str, seqid: int) -> None:

0 commit comments

Comments
 (0)