Coverage for gws-app/gws/plugin/ows_client/wfs/provider.py: 63%

1"""WFS service provider.""" 

2 

3from typing import Optional, cast 

4 

5import gws 

6import gws.base.ows.client 

7import gws.lib.shape 

8import gws.config.util 

9import gws.lib.bounds 

10import gws.lib.crs 

11import gws.lib.extent 

12import gws.gis.source 

13 

14from . import caps 

15 

16 

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

18 """Connection to a WFS service.""" 

19 

20 withBboxCrs: Optional[bool] 

21 """Add the CRS to the bounding box parameter.""" 

22 

23 

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

25 """WFS service provider. 

26 

27 Reads the capabilities of a WFS service and runs GetFeature requests for 

28 the layers, finders and models that use the service. 

29 """ 

30 

31 protocol = gws.OwsProtocol.WFS 

32 withBboxCrs: bool 

33 """Append the CRS to the ``BBOX`` parameter. Defaults to ``True`` for WFS 2.""" 

34 isWfs2: bool 

35 """The service version is 2 or higher.""" 

36 

37 def configure(self): 

38 cc = caps.parse(self.get_capabilities()) 

39 

40 self.metadata = cc.metadata 

41 self.sourceLayers = cc.sourceLayers 

42 self.version = cc.version 

43 self.isWfs2 = self.version >= '2' 

44 

45 self.configure_operations(cc.operations) 

46 

47 # use bbox with crs for wfs 2 by default 

48 # see also comments in qgis/qgswfsfeatureiterator.cpp buildURL 

49 p = self.cfg('withBboxCrs') 

50 self.withBboxCrs = self.isWfs2 if p is None else p 

51 

52 DEFAULT_GET_FEATURE_LIMIT = 100 

53 """Max. number of features in GetFeature requests, if the search has no limit.""" 

54 

55 def create_leaf_layer_config(self, source_layers): 

56 """Create the configuration of a ``wfsflat`` layer for the given feature types. 

57 

58 Used by the ``wfs`` tree layer to create its child layers. 

59 

60 Args: 

61 source_layers: Feature types to show in the layer. 

62 

63 Returns: 

64 A layer configuration that uses this provider. 

65 """ 

66 return dict( 

67 type='wfsflat', 

68 _defaultProvider=self, 

69 _defaultSourceLayers=source_layers, 

70 ) 

71 

72 def get_features(self, search, source_layers, mime=''): 

73 bounds = search.bounds 

74 search_shape = None 

75 

76 if search.shape: 

77 geometry_tolerance = 0.0 

78 

79 if search.tolerance: 

80 n, u = search.tolerance 

81 geometry_tolerance = n * (search.resolution or 1) if u == 'px' else n 

82 

83 search_shape = search.shape.tolerance_polygon(geometry_tolerance) 

84 bounds = search_shape.bounds() 

85 

86 if not bounds: 

87 gws.log.warning('get_features: no bounds or shape given') 

88 return [] 

89 

90 request_crs = self.forceCrs or gws.lib.crs.WGS84 

91 

92 bbox = gws.lib.bounds.transform(bounds, request_crs).extent 

93 if request_crs.isYX and not self.alwaysXY: 

94 bbox = gws.lib.extent.swap_xy(bbox) 

95 bbox = ','.join(str(k) for k in bbox) 

96 

97 srs = request_crs.urn 

98 if self.withBboxCrs: 

99 bbox += ',' + srs 

100 

101 params = { 

102 'BBOX': bbox, 

103 'COUNT' if self.isWfs2 else 'MAXFEATURES': search.limit or self.DEFAULT_GET_FEATURE_LIMIT, 

104 'SRSNAME': srs, 

105 'TYPENAMES' if self.isWfs2 else 'TYPENAME': [sl.name for sl in source_layers], 

106 'VERSION': self.version, 

107 } 

108 

109 if search.extraParams: 

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

111 

112 op = self.get_operation(gws.OwsVerb.GetFeature) 

113 if not op: 

114 return [] 

115 

116 mime = mime or op.defaultMimeType 

117 if mime: 

118 params.setdefault('OUTPUTFORMAT', mime) 

119 

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

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

122 

123 try: 

124 records = gws.base.ows.client.featureinfo.parse(text, default_crs=request_crs, always_xy=self.alwaysXY) 

125 except gws.Error as exc: 

126 gws.log.error(f'get_features: parse error: {exc!r}') 

127 return [] 

128 

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

130 

131 for rec in records: 

132 if rec.shape: 

133 rec.shape = rec.shape.transformed_to(bounds.crs) 

134 

135 if not search_shape: 

136 return records 

137 

138 filtered = [ 

139 rec for rec in records 

140 if not rec.shape or rec.shape.intersects(search_shape) 

141 ] 

142 

143 gws.log.debug(f'get_features: FILTERED={len(filtered)}') 

144 return filtered