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

1"""WMS service provider.""" 

2 

3from typing import Optional, cast 

4 

5import gws 

6import gws.base.ows.client 

7import gws.config.util 

8import gws.lib.crs 

9import gws.lib.extent 

10import gws.gis.source 

11 

12 

13from . import caps 

14 

15 

16class Config(gws.base.ows.client.provider.Config): 

17 """Connection to a WMS service.""" 

18 

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)""" 

23 

24 

25class Object(gws.base.ows.client.provider.Object): 

26 """WMS service provider. 

27 

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 """ 

31 

32 protocol = gws.OwsProtocol.WMS 

33 

34 maxRequestPixels: int 

35 """Max. width and height of a single GetMap request in pixels; the grabber fetches larger images in chunks.""" 

36 

37 def configure(self): 

38 self.maxRequestPixels = self.cfg('maxRequestPixels') 

39 

40 cc = caps.parse(self.get_capabilities(), self.cfg('bottomFirst', default=False)) 

41 

42 self.metadata = cc.metadata 

43 self.sourceLayers = cc.sourceLayers 

44 self.version = cc.version 

45 

46 self.configure_operations(cc.operations) 

47 

48 def cache_hash(self): 

49 return gws.u.sha256([ 

50 super().cache_hash(), 

51 self.cfg('bottomFirst', default=False), 

52 ]) 

53 

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. 

56 

57 The source layers are given top-first and sent bottom-first, as WMS expects. 

58 The image is requested with a transparent background. 

59 

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. 

66 

67 Returns: 

68 The image content. 

69 

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' 

75 

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) 

80 

81 layer_names = list(reversed([sl.name for sl in source_layers])) 

82 

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 } 

94 

95 op = self.get_operation(gws.OwsVerb.GetMap) 

96 if not op: 

97 raise gws.ExternalServiceError(f'no GetMap operation in {self.url!r}') 

98 

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 

104 

105 DEFAULT_GET_FEATURE_LIMIT = 100 

106 """Value of ``FEATURE_COUNT`` in GetFeatureInfo requests, if the search has no limit.""" 

107 

108 def create_leaf_layer_config(self, source_layers): 

109 """Create the configuration of a ``wmsflat`` layer for the given source layers. 

110 

111 Used by the ``wms`` tree layer to create its leaf layers. 

112 

113 Args: 

114 source_layers: Source layers to render in the layer. 

115 

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 ) 

124 

125 def get_features(self, search, source_layers): 

126 v3 = self.version >= '1.3' 

127 

128 shape = search.shape 

129 if not shape or shape.type != gws.GeometryType.point: 

130 return [] 

131 

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)) 

137 

138 box_size_m = 500 

139 box_size_deg = 1 

140 box_size_px = 500 

141 

142 size = None 

143 

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 [] 

152 

153 bbox = ( 

154 shape.x - (size / 2), 

155 shape.y - (size / 2), 

156 shape.x + (size / 2), 

157 shape.y + (size / 2), 

158 ) 

159 

160 bbox = gws.lib.extent.transform(bbox, shape.crs, request_crs) 

161 

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) 

165 

166 layer_names = [sl.name for sl in source_layers] 

167 

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 } 

181 

182 if search.extraParams: 

183 params = gws.u.merge(params, gws.u.to_upper_dict(search.extraParams)) 

184 

185 op = self.get_operation(gws.OwsVerb.GetFeatureInfo) 

186 if not op: 

187 return [] 

188 

189 if op.preferredFormat: 

190 params.setdefault('INFO_FORMAT', op.preferredFormat) 

191 

192 args = self.prepare_operation(op, params=params) 

193 text = gws.base.ows.client.request.get_text(args) 

194 

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 [] 

200 

201 gws.log.debug(f'get_features: FOUND={len(records)} params={params!r}') 

202 

203 for rec in records: 

204 if rec.shape: 

205 rec.shape = rec.shape.transformed_to(shape.crs) 

206 

207 return records