Coverage for gws-app/gws/plugin/ows_client/wms/provider.py: 0%
90 statements
« prev ^ index » next coverage.py v7.16.2, created at 2026-10-05 13:35 +0200
« prev ^ index » next coverage.py v7.16.2, created at 2026-10-05 13:35 +0200
1"""WMS service provider."""
3from typing import Optional, cast
5import gws
6import gws.base.ows.client
7import gws.config.util
8import gws.lib.crs
9import gws.lib.extent
10import gws.gis.source
13from . import caps
16class Config(gws.base.ows.client.provider.Config):
17 """Connection to a WMS service."""
19 bottomFirst: bool = False
20 """The service lists layers in its capabilities from bottom to top."""
21 maxRequestPixels: int = 4096
22 """Max. size of a single map request in pixels. (added in 8.5)"""
25class Object(gws.base.ows.client.provider.Object):
26 """WMS service provider.
28 Reads the capabilities of a WMS service and runs GetMap and GetFeatureInfo
29 requests for the layers, finders and models that use the service.
30 """
32 protocol = gws.OwsProtocol.WMS
34 maxRequestPixels: int
35 """Max. width and height of a single GetMap request in pixels; the grabber fetches larger images in chunks."""
37 def configure(self):
38 self.maxRequestPixels = self.cfg('maxRequestPixels')
40 cc = caps.parse(self.get_capabilities(), self.cfg('bottomFirst', default=False))
42 self.metadata = cc.metadata
43 self.sourceLayers = cc.sourceLayers
44 self.version = cc.version
46 self.configure_operations(cc.operations)
48 def cache_hash(self):
49 return gws.u.sha256([
50 super().cache_hash(),
51 self.cfg('bottomFirst', default=False),
52 ])
54 def get_map(self, bounds: gws.Bounds, width: int, height: int, source_layers: list[gws.SourceLayer], mime_type: str) -> bytes:
55 """Fetch a map image with GetMap.
57 The source layers are given top-first and sent bottom-first, as WMS expects.
58 The image is requested with a transparent background.
60 Args:
61 bounds: Bounds of the image; their CRS is the request CRS.
62 width: Image width in pixels.
63 height: Image height in pixels.
64 source_layers: Source layers to render, topmost first.
65 mime_type: Image format to request.
67 Returns:
68 The image content.
70 Raises:
71 ``gws.ExternalServiceError``: If the service has no GetMap operation
72 or the response is not an image.
73 """
74 v3 = self.version >= '1.3'
76 bbox = bounds.extent
77 always_xy = self.alwaysXY or not v3
78 if bounds.crs.isYX and not always_xy:
79 bbox = gws.lib.extent.swap_xy(bbox)
81 layer_names = list(reversed([sl.name for sl in source_layers]))
83 params = {
84 'BBOX': bbox,
85 'CRS' if v3 else 'SRS': bounds.crs.to_string(gws.CrsFormat.epsg),
86 'WIDTH': width,
87 'HEIGHT': height,
88 'LAYERS': layer_names,
89 'STYLES': [''] * len(layer_names),
90 'FORMAT': mime_type,
91 'TRANSPARENT': 'TRUE',
92 'VERSION': self.version,
93 }
95 op = self.get_operation(gws.OwsVerb.GetMap)
96 if not op:
97 raise gws.ExternalServiceError(f'no GetMap operation in {self.url!r}')
99 args = self.prepare_operation(op, params=params)
100 res = gws.base.ows.client.request.get(args)
101 if not res.content_type.startswith('image/'):
102 raise gws.ExternalServiceError(f'GetMap failed: content type {res.content_type!r} url={self.url!r}')
103 return res.content
105 DEFAULT_GET_FEATURE_LIMIT = 100
106 """Value of ``FEATURE_COUNT`` in GetFeatureInfo requests, if the search has no limit."""
108 def create_leaf_layer_config(self, source_layers):
109 """Create the configuration of a ``wmsflat`` layer for the given source layers.
111 Used by the ``wms`` tree layer to create its leaf layers.
113 Args:
114 source_layers: Source layers to render in the layer.
116 Returns:
117 A layer configuration that uses this provider.
118 """
119 return dict(
120 type='wmsflat',
121 _defaultProvider=self,
122 _defaultSourceLayers=source_layers,
123 )
125 def get_features(self, search, source_layers):
126 v3 = self.version >= '1.3'
128 shape = search.shape
129 if not shape or shape.type != gws.GeometryType.point:
130 return []
132 request_crs = self.forceCrs
133 if not request_crs:
134 request_crs = gws.lib.crs.best_match(
135 shape.crs,
136 gws.gis.source.combined_crs_list(source_layers))
138 box_size_m = 500
139 box_size_deg = 1
140 box_size_px = 500
142 size = None
144 if shape.crs.uom == gws.Uom.m:
145 size = box_size_px * search.resolution
146 if shape.crs.uom == gws.Uom.deg:
147 # @TODO use search.resolution here as well
148 size = box_size_deg
149 if not size:
150 gws.log.debug('cannot request crs {crs!r}, unsupported unit')
151 return []
153 bbox = (
154 shape.x - (size / 2),
155 shape.y - (size / 2),
156 shape.x + (size / 2),
157 shape.y + (size / 2),
158 )
160 bbox = gws.lib.extent.transform(bbox, shape.crs, request_crs)
162 always_xy = self.alwaysXY or not v3
163 if request_crs.isYX and not always_xy:
164 bbox = gws.lib.extent.swap_xy(bbox)
166 layer_names = [sl.name for sl in source_layers]
168 params = {
169 'BBOX': bbox,
170 'CRS' if v3 else 'SRS': request_crs.to_string(gws.CrsFormat.epsg),
171 'WIDTH': box_size_px,
172 'HEIGHT': box_size_px,
173 'I' if v3 else 'X': box_size_px >> 1,
174 'J' if v3 else 'Y': box_size_px >> 1,
175 'LAYERS': layer_names,
176 'QUERY_LAYERS': layer_names,
177 'STYLES': [''] * len(layer_names),
178 'VERSION': self.version,
179 'FEATURE_COUNT': search.limit or self.DEFAULT_GET_FEATURE_LIMIT,
180 }
182 if search.extraParams:
183 params = gws.u.merge(params, gws.u.to_upper_dict(search.extraParams))
185 op = self.get_operation(gws.OwsVerb.GetFeatureInfo)
186 if not op:
187 return []
189 if op.preferredFormat:
190 params.setdefault('INFO_FORMAT', op.preferredFormat)
192 args = self.prepare_operation(op, params=params)
193 text = gws.base.ows.client.request.get_text(args)
195 try:
196 records = gws.base.ows.client.featureinfo.parse(text, default_crs=request_crs, always_xy=self.alwaysXY)
197 except gws.Error as exc:
198 gws.log.error(f'get_features: parse error: {exc!r}')
199 return []
201 gws.log.debug(f'get_features: FOUND={len(records)} params={params!r}')
203 for rec in records:
204 if rec.shape:
205 rec.shape = rec.shape.transformed_to(shape.crs)
207 return records