Coverage for gws-app/gws/plugin/gbd_geoservices/model.py: 0%

1"""The ``gbd_geoservices`` model.""" 

2 

3import gws 

4import gws.base.feature 

5import gws.base.model 

6import gws.lib.shape 

7import gws.lib.bounds 

8import gws.lib.crs 

9import gws.gis.source 

10import gws.lib.jsonx 

11import gws.lib.net 

12 

13 

14@gws.ext.config.model('gbd_geoservices') 

15class Config(gws.base.model.Config): 

16 """Read-only model for features from GBD Geoservices.""" 

17 

18 apiKey: str 

19 """API key for GBD Geoservices.""" 

20 

21 

22@gws.ext.object.model('gbd_geoservices') 

23class Object(gws.base.model.default_model.Object): 

24 """GBD Geoservices model. 

25 

26 Read-only model that loads features from the GBD Geoservices search 

27 service. 

28 """ 

29 

30 apiKey: str 

31 """API key for the service.""" 

32 

33 serviceUrl = 'https://geoservices.gbd-consult.de/search' 

34 

35 def configure(self): 

36 self.apiKey = self.cfg('apiKey') 

37 self.uidName = 'uid' 

38 self.geometryName = 'geometry' 

39 self.loadingStrategy = gws.FeatureLoadingStrategy.all 

40 

41 def props(self, user): 

42 return gws.u.merge( 

43 super().props(user), 

44 canCreate=False, 

45 canDelete=False, 

46 canWrite=False, 

47 ) 

48 

49 def find_features(self, search, mc, **kwargs): 

50 kw = (search.keyword or '').strip() 

51 

52 tolerance = 0.0 

53 if search.tolerance: 

54 n, u = search.tolerance 

55 tolerance = n * (search.resolution or 1) if u == 'px' else n 

56 

57 area = None 

58 if search.shape: 

59 area = search.shape.tolerance_polygon(tolerance).transformed_to(gws.lib.crs.WGS84) 

60 

61 request = {} 

62 

63 if kw: 

64 request['mode'] = 'phrase' 

65 request['text'] = kw 

66 if area: 

67 request['area'] = area.to_wkt(output_dimension=2) 

68 request['within'] = False 

69 elif area: 

70 request['mode'] = 'point' 

71 if tolerance > 0: 

72 request['radius'] = min(tolerance, _MAX_POINT_RADIUS) 

73 else: 

74 return [] 

75 

76 request['limit'] = min(search.limit or _MAX_LIMIT, _MAX_LIMIT) 

77 if area: 

78 c = area.centroid() 

79 request['near'] = [c.geom.x, c.geom.y] 

80 

81 res = self._request(self.serviceUrl, method='POST', json=request) 

82 if not res: 

83 return [] 

84 

85 results = res.get('results') or {} 

86 features = results.get('features') or [] 

87 

88 out = [] 

89 

90 for f in features: 

91 a = self._attributes(f['properties']) 

92 if not a: 

93 gws.log.warning(f'geoservices: skipping feature without name, address or category: {f.get("id")}') 

94 continue 

95 shape = gws.lib.shape.from_geojson(f['geometry'], gws.lib.crs.WGS84, always_xy=True) 

96 rec = gws.FeatureRecord(uid=f['id'], attributes=a, shape=shape) 

97 gws.log.debug(f'geoservices: feature record: {rec=}') 

98 out.append(self.feature_from_record(rec, mc)) 

99 

100 return out 

101 

102 def _request(self, url, **kwargs) -> dict: 

103 """Send a request to the service and return the decoded JSON, or an empty dict on errors.""" 

104 res = gws.lib.net.http_request( 

105 url, 

106 headers={'Authorization': f'Bearer {self.apiKey}'}, 

107 **kwargs, 

108 ) 

109 if not res.ok: 

110 gws.log.error(f'geoservices request error: {res.status_code} {res.text}') 

111 return {} 

112 gws.log.debug(f'{res.text=}') 

113 return gws.lib.jsonx.from_string(res.text) 

114 

115 def _attributes(self, props: dict) -> dict | None: 

116 """Create feature attributes from a result, or ``None`` if it has no name, address or category.""" 

117 name = props.get('name') or '' 

118 

119 addr1 = _join([props.get('address_street'), props.get('address_housenumber')]) 

120 addr2 = _join([props.get('address_postcode'), props.get('address_city')]) 

121 

122 category = props.get('category_name') or '' 

123 icon = props.get('icon') or '' 

124 places = [p['name'] for p in props.get('places') or []] 

125 places_line = ' • '.join(places) 

126 place = places[-1] if places else '' 

127 

128 a = dict(title='', subtitle='', teaser='', details='', icon=icon) 

129 

130 if name: 

131 a['title'] = name 

132 a['subtitle'] = category 

133 a['teaser'] = places_line 

134 if addr1: 

135 a['details'] = addr1 + '\n' + addr2 

136 else: 

137 a['details'] = places_line 

138 elif addr1: 

139 a['title'] = addr1 + '\n' + addr2 

140 a['subtitle'] = category 

141 elif category: 

142 a['title'] = category 

143 a['subtitle'] = '' 

144 a['teaser'] = place 

145 if addr1: 

146 a['details'] = addr1 + '\n' + addr2 

147 else: 

148 a['details'] = places_line 

149 else: 

150 return None 

151 

152 return a 

153 

154 

155_MAX_LIMIT = 100 

156_MAX_POINT_RADIUS = 10_000 

157 

158 

159def _join(parts): 

160 """Join the non-empty parts with single spaces.""" 

161 return ' '.join(' '.join(str(p) for p in parts if p).split())