Skip to content

OSMChange#

osmdiff.osmchange.OSMChange #

Bases: object

Class to represent an OSMChange object.

Parameters:

Name Type Description Default
url str | None

URL of the OSM replication server

 None
frequency str

frequency of the replication diff

 'minute'
file str | None

path to the XML file

 None
sequence_number int | None

sequence number of the diff

 None
timeout int | None

request timeout

 None

Attributes:

Name Type Description
base_url str

URL of the OSM replication server
timeout int

request timeout
create list

list of created OSM objects
modify list

list of modified OSM objects
delete list

list of deleted OSM objects

Raises:

Type Description
Exception

If an invalid sequence number is provided
ValueError

If frequency is not one of the valid options
Example 
# Create an OSMChange object with a URL and frequency
osmchange = OSMChange(
    url="https://osm.example.com",
    frequency="minute",
    file="path/to/osmchange.xml",
    sequence_number=123456789,
)
Source code in src/osmdiff/osmchange.py 
 12
 13
 14
 15
 16
 17
 18
 19
 20
 21
 22
 23
 24
 25
 26
 27
 28
 29
 30
 31
 32
 33
 34
 35
 36
 37
 38
 39
 40
 41
 42
 43
 44
 45
 46
 47
 48
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
class OSMChange(object):
    """
    Class to represent an OSMChange object.

    Parameters:
        url (str | None): URL of the OSM replication server
        frequency (str): frequency of the replication diff
        file (str | None): path to the XML file
        sequence_number (int | None): sequence number of the diff
        timeout (int | None): request timeout

    Attributes:
        base_url (str): URL of the OSM replication server
        timeout (int): request timeout
        create (list): list of created OSM objects
        modify (list): list of modified OSM objects
        delete (list): list of deleted OSM objects

    Raises:
        Exception: If an invalid sequence number is provided
        ValueError: If frequency is not one of the valid options

    Example:
        ```python
        # Create an OSMChange object with a URL and frequency
        osmchange = OSMChange(
            url="https://osm.example.com",
            frequency="minute",
            file="path/to/osmchange.xml",
            sequence_number=123456789,
        )
        ```
    """

    def __init__(
        self,
        url: Optional[str] = None,
        frequency: str = "minute",
        file: Optional[str] = None,
        sequence_number: Optional[int] = None,
        timeout: Optional[int] = None,
    ):
        # Initialize with defaults from config
        self.base_url = url or API_CONFIG["osm"]["base_url"]
        self.timeout = timeout or API_CONFIG["osm"]["timeout"]

        self.create = []
        self.modify = []
        self.delete = []

        if file:
            with open(file, "r") as fh:
                xml = ElementTree.iterparse(fh, events=("start", "end"))
                self._parse_xml(xml)
        else:
            self._frequency = frequency
            self._sequence_number = sequence_number

    def get_state(self) -> bool:
        """
        Retrieve the current state from the OSM API.

        Returns:
            bool: True if state was successfully retrieved, False otherwise

        Raises:
            requests.RequestException: If the API request fails
        """
        state_url = urljoin(self.base_url, self._frequency, "state.txt")
        response = requests.get(
            state_url, timeout=self.timeout, headers=DEFAULT_HEADERS
        )
        if response.status_code != 200:
            return False
        for line in response.text.split("\n"):
            if line.startswith("sequenceNumber"):
                self._sequence_number = int(line[15:])
        return True

    def _build_sequence_url(self) -> str:
        seqno = str(self._sequence_number).zfill(9)
        url = urljoin(
            self.base_url,
            self._frequency,
            seqno[:3],
            seqno[3:6],
            "{}{}".format(seqno[6:], ".osc.gz"),
        )
        return url

    def _parse_xml(self, xml) -> None:
        for event, elem in xml:
            if elem.tag in ("create", "modify", "delete"):
                self._build_action(elem)

    def _build_action(self, elem: ElementTree.Element) -> None:
        """
        Build OSM objects from XML elements and add them to the appropriate list.

        Args:
            elem (ElementTree.Element): XML element containing OSM objects
        """
        for thing in elem:
            o = OSMObject.from_xml(thing)
            getattr(self, elem.tag).append(o)  # Use getattr instead of __getattribute__

    def retrieve(self, clear_cache: bool = False, timeout: Optional[int] = None) -> int:
        """
        Retrieve the OSM diff corresponding to the OSMChange sequence_number.

        Parameters:
            clear_cache (bool): clear the cache
            timeout (int): request timeout

        Returns:
            int: HTTP status code

        Raises:
            Exception: If an invalid sequence number is provided
        """
        if not self._sequence_number:
            raise Exception("invalid sequence number")
        if clear_cache:
            self.create, self.modify, self.delete = ([], [], [])
        try:
            r = requests.get(
                self._build_sequence_url(),
                stream=True,
                timeout=timeout or self.timeout,
                headers=DEFAULT_HEADERS,
            )
            if r.status_code != 200:
                return r.status_code
            gzfile = GzipFile(fileobj=r.raw)
            xml = ElementTree.iterparse(gzfile, events=("start", "end"))
            self._parse_xml(xml)
            return r.status_code
        except ConnectionError:
            # FIXME catch this?
            return 0

    @classmethod
    def from_xml(cls, xml: ElementTree.Element) -> "OSMChange":
        """
        Initialize OSMChange object from an XML object.

        If you used this method before version 0.3, please note that this
        method now takes an XML object. If you want to initialize from a file,\
        use the from_xml_file method.

        Parameters:
            xml (ElementTree.Element): XML object

        Returns:
            OSMChange: OSMChange object
        """
        new_osmchange_obj = cls()
        new_osmchange_obj._parse_xml(xml)
        return new_osmchange_obj

    @classmethod
    def from_xml_file(cls, path) -> "OSMChange":
        """
        Initialize OSMChange object from an XML file.

        Parameters:
            path (str): path to the XML file

        Returns:
            OSMChange: OSMChange object
        """
        with open(path, "r") as fh:
            xml = ElementTree.iterparse(fh, events=("start", "end"))
            return cls.from_xml(xml)

    @property
    def sequence_number(self) -> int:
        return self._sequence_number

    @sequence_number.setter
    def sequence_number(self, value):
        try:
            # value can be none
            if value is None:
                self._sequence_number = None
                return
            self._sequence_number = int(value)
        except ValueError:
            raise ValueError(
                "sequence_number must be an integer or parsable as an integer"
            )

    @property
    def frequency(self) -> str:
        return self._frequency

    @frequency.setter
    def frequency(self, f: str) -> None:
        """
        Set the frequency for OSM changes.

        Args:
            f (str): Frequency ('minute', 'hour', or 'day')

        Raises:
            ValueError: If frequency is not one of the valid options
        """
        VALID_FREQUENCIES = {"minute", "hour", "day"}
        if f not in VALID_FREQUENCIES:
            raise ValueError(
                f"Frequency must be one of: {', '.join(VALID_FREQUENCIES)}"
            )
        self._frequency = f

    def __repr__(self):
        return "OSMChange ({create} created, {modify} modified, \
{delete} deleted)".format(
            create=len(self.create), modify=len(self.modify), delete=len(self.delete)
        )

    def __enter__(self):
        return self

    def __exit__(self, exc_type, exc_val, exc_tb):
        """Clear all changes when exiting context."""
        self.create.clear()
        self.modify.clear()
        self.delete.clear()

__exit__(exc_type, exc_val, exc_tb) #

Clear all changes when exiting context.

Source code in src/osmdiff/osmchange.py 
235
236
237
238
239
def __exit__(self, exc_type, exc_val, exc_tb):
    """Clear all changes when exiting context."""
    self.create.clear()
    self.modify.clear()
    self.delete.clear()

from_xml(xml) classmethod #

Initialize OSMChange object from an XML object.

If you used this method before version 0.3, please note that this method now takes an XML object. If you want to initialize from a file, use the from_xml_file method.

Parameters:

Name Type Description Default
xml Element

XML object

 required

Returns:

Name Type Description
OSMChange OSMChange

OSMChange object
Source code in src/osmdiff/osmchange.py 
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
@classmethod
def from_xml(cls, xml: ElementTree.Element) -> "OSMChange":
    """
    Initialize OSMChange object from an XML object.

    If you used this method before version 0.3, please note that this
    method now takes an XML object. If you want to initialize from a file,\
    use the from_xml_file method.

    Parameters:
        xml (ElementTree.Element): XML object

    Returns:
        OSMChange: OSMChange object
    """
    new_osmchange_obj = cls()
    new_osmchange_obj._parse_xml(xml)
    return new_osmchange_obj

from_xml_file(path) classmethod #

Initialize OSMChange object from an XML file.

Parameters:

Name Type Description Default
path str

path to the XML file

 required

Returns:

Name Type Description
OSMChange OSMChange

OSMChange object
Source code in src/osmdiff/osmchange.py 
172
173
174
175
176
177
178
179
180
181
182
183
184
185
@classmethod
def from_xml_file(cls, path) -> "OSMChange":
    """
    Initialize OSMChange object from an XML file.

    Parameters:
        path (str): path to the XML file

    Returns:
        OSMChange: OSMChange object
    """
    with open(path, "r") as fh:
        xml = ElementTree.iterparse(fh, events=("start", "end"))
        return cls.from_xml(xml)

get_state() #

Retrieve the current state from the OSM API.

Returns:

Name Type Description
bool bool

True if state was successfully retrieved, False otherwise

Raises:

Type Description
RequestException

If the API request fails
Source code in src/osmdiff/osmchange.py 
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
def get_state(self) -> bool:
    """
    Retrieve the current state from the OSM API.

    Returns:
        bool: True if state was successfully retrieved, False otherwise

    Raises:
        requests.RequestException: If the API request fails
    """
    state_url = urljoin(self.base_url, self._frequency, "state.txt")
    response = requests.get(
        state_url, timeout=self.timeout, headers=DEFAULT_HEADERS
    )
    if response.status_code != 200:
        return False
    for line in response.text.split("\n"):
        if line.startswith("sequenceNumber"):
            self._sequence_number = int(line[15:])
    return True

retrieve(clear_cache=False, timeout=None) #

Retrieve the OSM diff corresponding to the OSMChange sequence_number.

Parameters:

Name Type Description Default
clear_cache bool

clear the cache

 False
timeout int

request timeout

 None

Returns:

Name Type Description
int int

HTTP status code

Raises:

Type Description
Exception

If an invalid sequence number is provided
Source code in src/osmdiff/osmchange.py 
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
def retrieve(self, clear_cache: bool = False, timeout: Optional[int] = None) -> int:
    """
    Retrieve the OSM diff corresponding to the OSMChange sequence_number.

    Parameters:
        clear_cache (bool): clear the cache
        timeout (int): request timeout

    Returns:
        int: HTTP status code

    Raises:
        Exception: If an invalid sequence number is provided
    """
    if not self._sequence_number:
        raise Exception("invalid sequence number")
    if clear_cache:
        self.create, self.modify, self.delete = ([], [], [])
    try:
        r = requests.get(
            self._build_sequence_url(),
            stream=True,
            timeout=timeout or self.timeout,
            headers=DEFAULT_HEADERS,
        )
        if r.status_code != 200:
            return r.status_code
        gzfile = GzipFile(fileobj=r.raw)
        xml = ElementTree.iterparse(gzfile, events=("start", "end"))
        self._parse_xml(xml)
        return r.status_code
    except ConnectionError:
        # FIXME catch this?
        return 0