Skip to content

Modules

Base classes for API clients.

WMSClient

Bases: ABC

Abstract base class for WMS service clients.

Source code in dovwms/base.py 
 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
class WMSClient(ABC):
    """Abstract base class for WMS service clients."""

    def __init__(self, base_url: str, wms_version: str = "1.3.0"):
        """Initialize the WMS client.

        The reason for the property and a setter implementation is to allow lazy connection
        to the WMS service at the point it is needed.

        Arguments:
            base_url: Base URL of the WMS service
            wms_version: WMS protocol version to use
        """
        self.base_url = base_url
        self.wms_version = wms_version
        self._wms: Optional[WebMapService] = None

    @property
    def wms(self) -> WebMapService:
        """Get the WMS connection, establishing it if needed."""
        if self._wms is None:
            self.connect_wms()
        return self._wms

    @wms.setter
    def wms(self, value: WebMapService) -> None:
        """Set the WMS connection.

        The setter allows injecting a mock WMS for testing.
        """
        self._wms = value

    def connect_wms(self) -> WebMapService:
        """Connect to the WMS service and return the connected WebMapService.

        Returns:
            The connected WebMapService instance.
        """
        wms_url = self.base_url if self.base_url.endswith("/wms") else f"{self.base_url}/wms"
        try:
            self._wms = WebMapService(wms_url, version=self.wms_version)
            logger.info("Connected to WMS service %s (%d layers available)", wms_url, len(self._wms.contents))
        except Exception:
            logger.exception("Failed to connect to WMS service at %s", wms_url)
        else:
            return self._wms
            raise

    def list_wms_layers(self, filter_func: Optional[Callable[[str, str], bool]] = None) -> dict[str, str]:
        """List available WMS layers from the service, optionally filtered.

        Arguments:
            filter_func: Optional function to filter layers. Takes layer name and title
                       as arguments and returns bool.

        Returns:
            Dictionary of layer names and titles
        """
        layers = {
            name: layer.title
            for name, layer in self.wms.contents.items()
            if filter_func is None or filter_func(name, layer.title)
        }
        return layers

    def check_layer_exists(self, layer_name: str) -> bool:
        """Check if a layer exists in the WMS service.

        Arguments:
            layer_name: Name of the layer to check

        Returns:
            True if layer exists, False otherwise
        """
        return layer_name in self.wms.contents

    @abstractmethod
    def parse_feature_info(self, content: str, **kwargs: Any) -> dict[str, Any]:
        """Parse GetFeatureInfo response content.

        This method should be implemented by subclasses to handle
        service-specific response formats. Preferably, in these method, each data
        type (e.g., soil texture, elevation) gets its own parsing logic in a dedicated
        private method.

        Arguments:
            content: Raw response content as string
            **kwargs: Additional parsing parameters

        Returns:
            Parsed content in appropriate format
        """
        pass

wms property writable

Get the WMS connection, establishing it if needed.

__init__(base_url, wms_version='1.3.0')

Initialize the WMS client.

The reason for the property and a setter implementation is to allow lazy connection to the WMS service at the point it is needed.

Parameters:

Name Type Description Default
base_url str

Base URL of the WMS service

 required
wms_version str

WMS protocol version to use

 '1.3.0'
Source code in dovwms/base.py 
19
20
21
22
23
24
25
26
27
28
29
30
31
def __init__(self, base_url: str, wms_version: str = "1.3.0"):
    """Initialize the WMS client.

    The reason for the property and a setter implementation is to allow lazy connection
    to the WMS service at the point it is needed.

    Arguments:
        base_url: Base URL of the WMS service
        wms_version: WMS protocol version to use
    """
    self.base_url = base_url
    self.wms_version = wms_version
    self._wms: Optional[WebMapService] = None

check_layer_exists(layer_name)

Check if a layer exists in the WMS service.

Parameters:

Name Type Description Default
layer_name str

Name of the layer to check

 required

Returns:

Type Description
bool

True if layer exists, False otherwise
Source code in dovwms/base.py 
81
82
83
84
85
86
87
88
89
90
def check_layer_exists(self, layer_name: str) -> bool:
    """Check if a layer exists in the WMS service.

    Arguments:
        layer_name: Name of the layer to check

    Returns:
        True if layer exists, False otherwise
    """
    return layer_name in self.wms.contents

connect_wms()

Connect to the WMS service and return the connected WebMapService.

Returns:

Type Description
WebMapService

The connected WebMapService instance.
Source code in dovwms/base.py 
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
def connect_wms(self) -> WebMapService:
    """Connect to the WMS service and return the connected WebMapService.

    Returns:
        The connected WebMapService instance.
    """
    wms_url = self.base_url if self.base_url.endswith("/wms") else f"{self.base_url}/wms"
    try:
        self._wms = WebMapService(wms_url, version=self.wms_version)
        logger.info("Connected to WMS service %s (%d layers available)", wms_url, len(self._wms.contents))
    except Exception:
        logger.exception("Failed to connect to WMS service at %s", wms_url)
    else:
        return self._wms
        raise

list_wms_layers(filter_func=None)

List available WMS layers from the service, optionally filtered.

Parameters:

Name Type Description Default
filter_func Optional[Callable[[str, str], bool]]

Optional function to filter layers. Takes layer name and title as arguments and returns bool.

 None

Returns:

Type Description
dict[str, str]

Dictionary of layer names and titles
Source code in dovwms/base.py 
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
def list_wms_layers(self, filter_func: Optional[Callable[[str, str], bool]] = None) -> dict[str, str]:
    """List available WMS layers from the service, optionally filtered.

    Arguments:
        filter_func: Optional function to filter layers. Takes layer name and title
                   as arguments and returns bool.

    Returns:
        Dictionary of layer names and titles
    """
    layers = {
        name: layer.title
        for name, layer in self.wms.contents.items()
        if filter_func is None or filter_func(name, layer.title)
    }
    return layers

parse_feature_info(content, **kwargs) abstractmethod

Parse GetFeatureInfo response content.

This method should be implemented by subclasses to handle service-specific response formats. Preferably, in these method, each data type (e.g., soil texture, elevation) gets its own parsing logic in a dedicated private method.

Parameters:

Name Type Description Default
content str

Raw response content as string

 required
**kwargs Any

Additional parsing parameters

 {}

Returns:

Type Description
dict[str, Any]

Parsed content in appropriate format
Source code in dovwms/base.py 
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
@abstractmethod
def parse_feature_info(self, content: str, **kwargs: Any) -> dict[str, Any]:
    """Parse GetFeatureInfo response content.

    This method should be implemented by subclasses to handle
    service-specific response formats. Preferably, in these method, each data
    type (e.g., soil texture, elevation) gets its own parsing logic in a dedicated
    private method.

    Arguments:
        content: Raw response content as string
        **kwargs: Additional parsing parameters

    Returns:
        Parsed content in appropriate format
    """
    pass

Client for the Belgian DOV (Databank Ondergrond Vlaanderen) API.

DOVClient

Bases: WMSClient

Client for fetching soil data from the Belgian DOV API.

Source code in dovwms/dov.py 
 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
class DOVClient(WMSClient):
    """Client for fetching soil data from the Belgian DOV API."""

    def __init__(self) -> None:
        """Lazy-initialize the DOV client.

        Connects to the DOV Geoserver WMS service for accessing soil data
        and related geological information. Open for further expasion.
        """
        super().__init__(base_url="https://www.dov.vlaanderen.be/geoserver")

    def list_wms_layers(self, filter_func: Optional[Callable[[str, str], bool]] = None) -> dict[str, str]:
        """List available WMS layers from the DOV service.

        Arguments:
            filter_func: Optional function to filter layers. Takes layer name and title
                     as arguments and returns bool. If None, only soil-related layers
                     are returned.

        Returns:
            Dictionary of layer names and titles
        """

        def soil_filter(name: str, title: str) -> bool:
            if filter_func is not None:
                return filter_func(name, title)
            return "bodem" in name.lower()

        return super().list_wms_layers(filter_func=soil_filter)

    def parse_feature_info(self, content: str, **kwargs: Any) -> dict[str, Any]:
        """Parse GetFeatureInfo response from DOV WMS.

        The parsing method depends on the content type and query type:
        - For soil texture: Parses JSON response with layer properties
        - For other queries: Returns raw content for specific handling

        Arguments:
            content: Raw response content
            **kwargs: Additional parameters:
                - content_type: Expected content type
                - query_type: Type of query (e.g., 'texture', 'properties')

        Returns:
            Parsed content in appropriate format
        """
        content_type = kwargs.get("content_type", "application/json")
        query_type = kwargs.get("query_type", "properties")

        if content_type == "application/json" and query_type == "texture":
            return self._parse_texture_response(content)

        return {"content": content}

    def _parse_texture_response(self, data: Any) -> dict[str, Any]:
        """Parse the WMS GetFeatureInfo response to extract texture fractions.

        Always returns a dictionary with a single key "layers" containing a
        list of layer dictionaries. This makes the parser output stable and
        lets callers attach additional information (e.g. elevation) to the
        returned dict without special-casing list vs dict.

        Args:
            data: data from a WMS GetFeatureInfo response (JSON string)

        Returns:
            Dict with key "layers" mapping to a list of layer dicts. If no
            features were found, returns {"layers": []}.
        """
        json_data = json.loads(data)
        features = json_data.get("features", [])

        if not features:
            return {"layers": []}

        properties = [feature.get("properties") for feature in features]

        # create Layer objects for each of the profiles
        depth_keys = [k for k in properties[0] if not k.endswith("_betrouwbaarheid")]

        # Map Dutch depth notation to layer info
        depth_mapping = {
            "_0_-_10_cm": (0, 10, "Layer_0-10cm"),
            "_10_-_30_cm": (10, 30, "Layer_10-30cm"),
            "_30_-_60_cm": (30, 60, "Layer_30-60cm"),
            "_60_-_100_cm": (60, 100, "Layer_60-100cm"),
            "_100_-_150_cm": (100, 150, "Layer_100-150cm"),
        }

        layers = []

        for depth_key in depth_keys:
            ci_key = f"{depth_key}_betrouwbaarheid"

            # Extract texture percentages and confidence intervals
            clay_pct = properties[0][depth_key]
            clay_mtd = {
                "source": "DOV WMS, bdbstat:fractie_klei_basisdata_bodemkartering",
                "uncertainty": properties[0][ci_key],
            }
            silt_pct = properties[1][depth_key]
            silt_mtd = {
                "source": "DOV WMS, bdbstat:fractie_leem_basisdata_bodemkartering",
                "uncertainty": properties[1][ci_key],
            }
            sand_pct = properties[2][depth_key]
            sand_mtd = {
                "source": "DOV WMS, bdbstat:fractie_zand_basisdata_bodemkartering",
                "uncertainty": properties[2][ci_key],
            }

            # Get depth info
            top_depth, bottom_depth, layer_name = depth_mapping[depth_key]

            # Create SoilLayer object
            layer = {
                "name": layer_name,
                "layer_top": top_depth,
                "layer_bottom": bottom_depth,
                "sand_content": sand_pct,
                "silt_content": silt_pct,
                "clay_content": clay_pct,
                "metadata": {
                    "sand_content": sand_mtd,
                    "silt_content": silt_mtd,
                    "clay_content": clay_mtd,
                },
            }

            layers.append(layer)

        return {"layers": layers}

    def fetch_profile(
        self, location: Point, fetch_elevation: bool = False, crs: str = "EPSG:31370"
    ) -> Optional[dict[str, Any]]:
        """Fetch soil texture information from the DOV WMS at a specific location.

        This method queries the DOV WMS service for clay, silt, and sand content
        at different depths at the specified location. The data is used to create
        a SoilProfile object with appropriate layers.

        Args:
            location: Point object with x, y coordinates
            fetch_elevation: Whether to fetch the elevation of the location from Geopunt.
            crs: Coordinate reference system

        Returns:
            Dictionary with texture data ("layers" key and optional "elevation") or None if data not found
        """
        # Texture layers from DOV bodemanalysie service.
        wms_layers = [
            "bdbstat:fractie_klei_basisdata_bodemkartering",  # clay
            "bdbstat:fractie_leem_basisdata_bodemkartering",  # silt
            "bdbstat:fractie_zand_basisdata_bodemkartering",  # sand
        ]

        # Verify layers exist
        for layer_name in wms_layers:
            if not self.check_layer_exists(layer_name):
                logger.warning("Layer %s not found", layer_name)
                return None

        # Define query area
        buffer = 0.0001
        bbox = (location.x - buffer, location.y - buffer, location.x + buffer, location.y + buffer)

        try:
            # Query texture data
            response = self.wms.getfeatureinfo(
                layers=wms_layers,
                query_layers=wms_layers,
                srs=crs,
                bbox=bbox,
                size=(100, 100),
                info_format="application/json",
                xy=(50, 50),  # center pixel
            )

            result = self.parse_feature_info(response.read(), content_type="application/json", query_type="texture")

            if fetch_elevation:
                elevation = get_elevation(location, crs)
                result["elevation"] = elevation
        except Exception:
            logger.exception("Failed to fetch profile")
            return None
        else:
            return result

__init__()

Lazy-initialize the DOV client.

Connects to the DOV Geoserver WMS service for accessing soil data and related geological information. Open for further expasion.

Source code in dovwms/dov.py 
18
19
20
21
22
23
24
def __init__(self) -> None:
    """Lazy-initialize the DOV client.

    Connects to the DOV Geoserver WMS service for accessing soil data
    and related geological information. Open for further expasion.
    """
    super().__init__(base_url="https://www.dov.vlaanderen.be/geoserver")

fetch_profile(location, fetch_elevation=False, crs='EPSG:31370')

Fetch soil texture information from the DOV WMS at a specific location.

This method queries the DOV WMS service for clay, silt, and sand content at different depths at the specified location. The data is used to create a SoilProfile object with appropriate layers.

Parameters:

Name Type Description Default
location Point

Point object with x, y coordinates

 required
fetch_elevation bool

Whether to fetch the elevation of the location from Geopunt.

 False
crs str

Coordinate reference system

 'EPSG:31370'

Returns:

Type Description
Optional[dict[str, Any]]

Dictionary with texture data ("layers" key and optional "elevation") or None if data not found
Source code in dovwms/dov.py 
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
def fetch_profile(
    self, location: Point, fetch_elevation: bool = False, crs: str = "EPSG:31370"
) -> Optional[dict[str, Any]]:
    """Fetch soil texture information from the DOV WMS at a specific location.

    This method queries the DOV WMS service for clay, silt, and sand content
    at different depths at the specified location. The data is used to create
    a SoilProfile object with appropriate layers.

    Args:
        location: Point object with x, y coordinates
        fetch_elevation: Whether to fetch the elevation of the location from Geopunt.
        crs: Coordinate reference system

    Returns:
        Dictionary with texture data ("layers" key and optional "elevation") or None if data not found
    """
    # Texture layers from DOV bodemanalysie service.
    wms_layers = [
        "bdbstat:fractie_klei_basisdata_bodemkartering",  # clay
        "bdbstat:fractie_leem_basisdata_bodemkartering",  # silt
        "bdbstat:fractie_zand_basisdata_bodemkartering",  # sand
    ]

    # Verify layers exist
    for layer_name in wms_layers:
        if not self.check_layer_exists(layer_name):
            logger.warning("Layer %s not found", layer_name)
            return None

    # Define query area
    buffer = 0.0001
    bbox = (location.x - buffer, location.y - buffer, location.x + buffer, location.y + buffer)

    try:
        # Query texture data
        response = self.wms.getfeatureinfo(
            layers=wms_layers,
            query_layers=wms_layers,
            srs=crs,
            bbox=bbox,
            size=(100, 100),
            info_format="application/json",
            xy=(50, 50),  # center pixel
        )

        result = self.parse_feature_info(response.read(), content_type="application/json", query_type="texture")

        if fetch_elevation:
            elevation = get_elevation(location, crs)
            result["elevation"] = elevation
    except Exception:
        logger.exception("Failed to fetch profile")
        return None
    else:
        return result

list_wms_layers(filter_func=None)

List available WMS layers from the DOV service.

Parameters:

Name Type Description Default
filter_func Optional[Callable[[str, str], bool]]

Optional function to filter layers. Takes layer name and title as arguments and returns bool. If None, only soil-related layers are returned.

 None

Returns:

Type Description
dict[str, str]

Dictionary of layer names and titles
Source code in dovwms/dov.py 
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
def list_wms_layers(self, filter_func: Optional[Callable[[str, str], bool]] = None) -> dict[str, str]:
    """List available WMS layers from the DOV service.

    Arguments:
        filter_func: Optional function to filter layers. Takes layer name and title
                 as arguments and returns bool. If None, only soil-related layers
                 are returned.

    Returns:
        Dictionary of layer names and titles
    """

    def soil_filter(name: str, title: str) -> bool:
        if filter_func is not None:
            return filter_func(name, title)
        return "bodem" in name.lower()

    return super().list_wms_layers(filter_func=soil_filter)

parse_feature_info(content, **kwargs)

Parse GetFeatureInfo response from DOV WMS.

The parsing method depends on the content type and query type: - For soil texture: Parses JSON response with layer properties - For other queries: Returns raw content for specific handling

Parameters:

Name Type Description Default
content str

Raw response content

 required
**kwargs Any

Additional parameters: - content_type: Expected content type - query_type: Type of query (e.g., 'texture', 'properties')

 {}

Returns:

Type Description
dict[str, Any]

Parsed content in appropriate format
Source code in dovwms/dov.py 
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
def parse_feature_info(self, content: str, **kwargs: Any) -> dict[str, Any]:
    """Parse GetFeatureInfo response from DOV WMS.

    The parsing method depends on the content type and query type:
    - For soil texture: Parses JSON response with layer properties
    - For other queries: Returns raw content for specific handling

    Arguments:
        content: Raw response content
        **kwargs: Additional parameters:
            - content_type: Expected content type
            - query_type: Type of query (e.g., 'texture', 'properties')

    Returns:
        Parsed content in appropriate format
    """
    content_type = kwargs.get("content_type", "application/json")
    query_type = kwargs.get("query_type", "properties")

    if content_type == "application/json" and query_type == "texture":
        return self._parse_texture_response(content)

    return {"content": content}

get_profile_from_dov(x, y, crs='EPSG:31370', fetch_elevation=True, profile_name=None)

Convenience function to fetch a soil profile from DOV at given coordinates.

This function handles all the necessary client setup and coordinate conversion to get a soil profile from the DOV service. It's a simpler alternative to creating and managing DOV and Geopunt clients manually.

Parameters:

Name Type Description Default
x float

X-coordinate in the specified CRS (default Lambert72)

 required
y float

Y-coordinate in the specified CRS (default Lambert72)

 required
profile_name Optional[str]

Optional name for the profile. If None, will use coordinates

 None
crs str

Coordinate reference system of the input coordinates

 'EPSG:31370'
fetch_elevation bool

elevation data

 True

Returns:

Type Description
Optional[dict[str, Any]]

SoilProfile object with texture and optional elevation data,
Optional[dict[str, Any]]

or None if the data couldn't be fetched
Example

profile = get_profile_from_dov(247172.56, 204590.58) print(f"Elevation: {profile.elevation:.2f}m") print(f"Number of layers: {len(profile.layers)}")

Source code in dovwms/dov.py 
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
240
241
242
243
244
245
246
247
248
249
250
251
def get_profile_from_dov(
    x: float, y: float, crs: str = "EPSG:31370", fetch_elevation: bool = True, profile_name: Optional[str] = None
) -> Optional[dict[str, Any]]:
    """Convenience function to fetch a soil profile from DOV at given coordinates.

    This function handles all the necessary client setup and coordinate conversion
    to get a soil profile from the DOV service. It's a simpler alternative to
    creating and managing DOV and Geopunt clients manually.

    Args:
        x: X-coordinate in the specified CRS (default Lambert72)
        y: Y-coordinate in the specified CRS (default Lambert72)
        profile_name: Optional name for the profile. If None, will use coordinates
        crs: Coordinate reference system of the input coordinates
        fetch_elevation: elevation data

    Returns:
        SoilProfile object with texture and optional elevation data,
        or None if the data couldn't be fetched

    Example:
        >>> profile = get_profile_from_dov(247172.56, 204590.58)
        >>> print(f"Elevation: {profile.elevation:.2f}m")
        >>> print(f"Number of layers: {len(profile.layers)}")
    """
    try:
        # Create location point
        location = Point(x, y)

        # Use coordinates for profile name if none provided (kept for backward compatibility,
        # but DOVClient.fetch_profile does not currently accept a profile_name parameter)
        if profile_name is None:
            profile_name = f"Profile_{x:.0f}_{y:.0f}"

        # Create DOV client
        client = DOVClient()

        # Fetch profile. Use the public DOVClient API: (location, fetch_elevation, crs)
        profile = client.fetch_profile(location, fetch_elevation=fetch_elevation, crs=crs)
    except Exception:
        logger.exception("Failed to get profile from DOV")
        return None
    else:
        return profile
        logger.exception("Error fetching profile from DOV")
        return None

GeopuntClient

Bases: WMSClient

Client for fetching data from the Geopunt API.

Source code in dovwms/geopunt.py 
 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
class GeopuntClient(WMSClient):
    """Client for fetching data from the Geopunt API."""

    def __init__(self) -> None:
        super().__init__(base_url="https://geo.api.vlaanderen.be/DHMV")

    def parse_feature_info(self, content: str, **kwargs: Any) -> dict[str, Any]:
        """Parse GetFeatureInfo response from Geopunt WMS.

        The parsing method depends on the content type and query type:
        - For elevation: Parses semicolon-separated response for elevation value
        - For other queries: Returns raw content for specific handling.
        - One substantial change.

        Args:
            content: Raw response content
            **kwargs: Additional parameters:
                - content_type: Expected content type
                - query_type: Type of query (e.g., 'elevation')

        Returns:
            Dict containing either {'elevation': float or None} or {'content': str}
        """
        query_type = kwargs.get("query_type", "elevation")

        if query_type == "elevation":
            elevation = self._parse_elevation_response(content)
            return {"elevation": elevation}

        return {"content": content}

    def _parse_elevation_response(self, content: str) -> Optional[float]:
        """Parse elevation data from GetFeatureInfo response.

        Args:
            content: Raw response content from WMS GetFeatureInfo

        Returns:
            Elevation in meters or None if parsing fails

        Note:
            Response format example:
            "@DHMVII_DTM_1m Stretched value;Pixel Value; 32.360001;32.360001;"
        """
        try:
            values = content.strip().split(";")
            if len(values) >= 3:
                return float(values[2].strip())
        except (ValueError, IndexError) as e:
            logger.warning("Error parsing elevation data: %s", e)
        return None

    def fetch_elevation(
        self, location: Point, crs: str = "EPSG:31370", layer_name: str = "DHMVII_DTM_1m"
    ) -> Optional[dict[str, Any]]:
        """Fetch elevation data from the Geopunt WMS at a specific location.

        Args:
            location: Point object with x, y coordinates
            crs: Coordinate reference system
        Returns:
            Elevation in meters or None if not found
        """
        # Layer name for Digital Terrain Model (1m resolution)

        if not self.check_layer_exists(layer_name):
            try:
                available = list(self.wms.contents.keys())
            except Exception:
                available = []
            logger.warning("Layer %s not found. Available layers: %s", layer_name, available)
            return None

        buffer = 0.0001
        bbox = (location.x - buffer, location.y - buffer, location.x + buffer, location.y + buffer)

        # Image size and center pixel
        img_width = img_height = 256
        pixel_x = pixel_y = img_width // 2

        try:
            # Make GetFeatureInfo request
            response = self.wms.getfeatureinfo(
                layers=[layer_name],
                query_layers=[layer_name],
                info_format="text/plain",
                srs=crs,
                bbox=bbox,
                size=(img_width, img_height),
                xy=(pixel_x, pixel_y),
            )

            # Parse response using the base class method
            content = response.read().decode("utf-8")
            elevation = self.parse_feature_info(content, content_type="text/plain", query_type="elevation")

            if elevation is not None:
                logger.info("Fetched elevation from Geopunt API")
        except Exception:
            logger.exception("Failed to fetch elevation from Geopunt API")
            return None
        else:
            return elevation

fetch_elevation(location, crs='EPSG:31370', layer_name='DHMVII_DTM_1m')

Fetch elevation data from the Geopunt WMS at a specific location.

Parameters:

Name Type Description Default
location Point

Point object with x, y coordinates

 required
crs str

Coordinate reference system

 'EPSG:31370'

Returns: Elevation in meters or None if not found

Source code in dovwms/geopunt.py 
 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
def fetch_elevation(
    self, location: Point, crs: str = "EPSG:31370", layer_name: str = "DHMVII_DTM_1m"
) -> Optional[dict[str, Any]]:
    """Fetch elevation data from the Geopunt WMS at a specific location.

    Args:
        location: Point object with x, y coordinates
        crs: Coordinate reference system
    Returns:
        Elevation in meters or None if not found
    """
    # Layer name for Digital Terrain Model (1m resolution)

    if not self.check_layer_exists(layer_name):
        try:
            available = list(self.wms.contents.keys())
        except Exception:
            available = []
        logger.warning("Layer %s not found. Available layers: %s", layer_name, available)
        return None

    buffer = 0.0001
    bbox = (location.x - buffer, location.y - buffer, location.x + buffer, location.y + buffer)

    # Image size and center pixel
    img_width = img_height = 256
    pixel_x = pixel_y = img_width // 2

    try:
        # Make GetFeatureInfo request
        response = self.wms.getfeatureinfo(
            layers=[layer_name],
            query_layers=[layer_name],
            info_format="text/plain",
            srs=crs,
            bbox=bbox,
            size=(img_width, img_height),
            xy=(pixel_x, pixel_y),
        )

        # Parse response using the base class method
        content = response.read().decode("utf-8")
        elevation = self.parse_feature_info(content, content_type="text/plain", query_type="elevation")

        if elevation is not None:
            logger.info("Fetched elevation from Geopunt API")
    except Exception:
        logger.exception("Failed to fetch elevation from Geopunt API")
        return None
    else:
        return elevation

parse_feature_info(content, **kwargs)

Parse GetFeatureInfo response from Geopunt WMS.

The parsing method depends on the content type and query type: - For elevation: Parses semicolon-separated response for elevation value - For other queries: Returns raw content for specific handling. - One substantial change.

Parameters:

Name Type Description Default
content str

Raw response content

 required
**kwargs Any

Additional parameters: - content_type: Expected content type - query_type: Type of query (e.g., 'elevation')

 {}

Returns:

Type Description
dict[str, Any]

Dict containing either {'elevation': float or None} or {'content': str}
Source code in dovwms/geopunt.py 
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
def parse_feature_info(self, content: str, **kwargs: Any) -> dict[str, Any]:
    """Parse GetFeatureInfo response from Geopunt WMS.

    The parsing method depends on the content type and query type:
    - For elevation: Parses semicolon-separated response for elevation value
    - For other queries: Returns raw content for specific handling.
    - One substantial change.

    Args:
        content: Raw response content
        **kwargs: Additional parameters:
            - content_type: Expected content type
            - query_type: Type of query (e.g., 'elevation')

    Returns:
        Dict containing either {'elevation': float or None} or {'content': str}
    """
    query_type = kwargs.get("query_type", "elevation")

    if query_type == "elevation":
        elevation = self._parse_elevation_response(content)
        return {"elevation": elevation}

    return {"content": content}

get_elevation(location, crs='EPSG:31370', layer_name='DHMVII_DTM_1m')

Convenience wrapper to fetch elevation using the GeopuntClient.

This helper creates a GeopuntClient, requests the elevation for the provided location and returns the value. Tests can patch this function to avoid instantiating the client or making network calls.

Source code in dovwms/geopunt.py 
118
119
120
121
122
123
124
125
126
127
128
def get_elevation(
    location: Point, crs: str = "EPSG:31370", layer_name: str = "DHMVII_DTM_1m"
) -> Optional[dict[str, Any]]:
    """Convenience wrapper to fetch elevation using the GeopuntClient.

    This helper creates a GeopuntClient, requests the elevation for the
    provided location and returns the value. Tests can patch this function
    to avoid instantiating the client or making network calls.
    """
    client = GeopuntClient()
    return client.fetch_elevation(location, crs, layer_name)