OSMChange

Core class for retrieving and parsing OpenStreetMap changesets in OSMChange format.

For working with the Augmented Diff format, we have AugmentedDiff and ContinuousAugmentedDiff.

Features

• Retrieves changesets from OSM replication servers
• Parses OSMChange XML format
• Handles create/modify/delete actions
• Supports both remote and local file sources
• Context manager support
• Sequence number management

Basic Usage

from osmdiff import OSMChange

# Create with sequence number
osm_change = OSMChange(sequence_number=12345)

# Retrieve and process changes
status = osm_change.retrieve()
if status == 200:
    creations = osm_change.actions["create"]
    modifications = osm_change.actions["modify"]
    deletions = osm_change.actions["delete"]
    print(f"Created: {len(creations)} features")
    print(f"Modified: {len(modifications)} features") 
    print(f"Deleted: {len(deletions)} features")

API Reference

Bases: object

Handles OpenStreetMap changesets in OSMChange format.

Parameters:

Name	Type	Description	Default
url	Optional[str]	Base URL of OSM replication server	None
frequency	str	Replication frequency (‘minute’, ‘hour’, or ‘day’)	'minute'
file	Optional[str]	Path to local OSMChange XML file	None
sequence_number	Optional[int]	Sequence number of the diff	None
timeout	Optional[int]	Request timeout in seconds	None

Note

Follows the OSM replication protocol.

Source code in src/osmdiff/osmchange.py

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

sequence_number property writable

frequency property writable

actions property

Get all actions combined in a single list.

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

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, "api/0.6/changesets/state")
    response = requests.get(
        state_url, timeout=self.timeout, headers=DEFAULT_HEADERS
    )
    if response.status_code != 200:
        return False

    # Parse XML response
    root = ElementTree.fromstring(response.content)
    state = root.find("state")
    if state is not None:
        seq = state.find("sequenceNumber")
        if seq is not None and seq.text:
            self._sequence_number = int(seq.text)
            return True
    return False

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

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
        # Handle both gzipped and plain XML responses
        content = r.content
        if content.startswith(b"\x1f\x8b"):  # Gzip magic number
            gzfile = GzipFile(fileobj=r.raw)
            xml = ElementTree.iterparse(gzfile, events=("start", "end"))
        else:
            xml = ElementTree.iterparse(r.raw, events=("start", "end"))
        self._parse_xml(xml)
        return r.status_code
    except ConnectionError:
        # FIXME catch this?
        return 0

__repr__()

Source code in src/osmdiff/osmchange.py

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

__enter__()

Source code in src/osmdiff/osmchange.py

def __enter__(self):
    return self

__exit__(exc_type, exc_val, exc_tb)

Clear all changes when exiting context.

Source code in src/osmdiff/osmchange.py

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

See Also

• AugmentedDiff - For augmented diffs with additional metadata