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

1"""GBD Geoservices model.""" 

2 

3import re 

4 

5import gws 

6import gws.base.feature 

7import gws.base.model 

8import gws.base.shape 

9import gws.gis.bounds 

10import gws.gis.crs 

11import gws.gis.source 

12import gws.lib.jsonx 

13import gws.lib.net 

14 

15gws.ext.new.model('gbd_geoservices') 

16 

17 

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

19 """GBD Geoservices model.""" 

20 

21 apiKey: str 

22 

23 

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

25 """GBD Geoservices model.""" 

26 

27 apiKey: str 

28 

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

30 

31 def configure(self): 

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

33 self.uidName = 'uid' 

34 self.geometryName = 'geometry' 

35 self.loadingStrategy = gws.FeatureLoadingStrategy.all 

36 

37 def props(self, user): 

38 return gws.u.merge( 

39 super().props(user), 

40 canCreate=False, 

41 canDelete=False, 

42 canWrite=False, 

43 ) 

44 

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

46 request = { 

47 'page': 0, 

48 'tags': {} 

49 } 

50 

51 if search.shape: 

52 geometry_tolerance = 0.0 

53 

54 if search.tolerance: 

55 n, u = search.tolerance 

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

57 

58 search_shape = search.shape.tolerance_polygon(geometry_tolerance) 

59 request['viewbox'] = gws.gis.bounds.wgs_extent(search_shape.bounds()) 

60 

61 kw = search.keyword or '' 

62 use_address = False 

63 if kw: 

64 request['intersect'] = 1 

65 # crude heuristics to check if this is an "address" or a "name" 

66 if re.search(r'\s\d', kw): 

67 request['address'] = kw 

68 use_address = True 

69 else: 

70 request['name'] = kw 

71 

72 features = {} 

73 

74 res = self._query(request) 

75 fs = res['results']['features'] 

76 

77 for f in fs: 

78 rec = gws.FeatureRecord( 

79 uid=f['id'], 

80 attributes={ 

81 k.replace(':', '_'): v 

82 for k, v in sorted(f['properties'].items()) 

83 }, 

84 shape=gws.base.shape.from_geojson(f['geometry'], gws.gis.crs.WGS84, always_xy=True), 

85 ) 

86 if search.shape and search.shape.type == gws.GeometryType.polygon and not rec['shape'].intersects(search.shape): 

87 continue 

88 

89 a = rec['attributes'] 

90 

91 if use_address and 'addr_housenumber' not in a: 

92 continue 

93 if not use_address and 'name' not in a: 

94 continue 

95 

96 address = ' '.join([ 

97 a.get('addr_street', ''), 

98 a.get('addr_housenumber', ''), 

99 a.get('addr_postcode', ''), 

100 a.get('addr_city', ''), 

101 ]) 

102 address = ' '.join(address.split()) 

103 

104 if use_address: 

105 a['title'] = address 

106 else: 

107 a['title'] = a.get('name') 

108 if address: 

109 a['title'] += ' (' + address + ')' 

110 

111 features[a['title']] = self.feature_from_record(rec, mc) 

112 

113 return [f for _, f in sorted(features.items())] 

114 

115 def _query(self, request) -> dict: 

116 try: 

117 res = gws.lib.net.http_request( 

118 self.serviceUrl, 

119 method='POST', 

120 headers={'x-api-key': self.apiKey}, 

121 json=request 

122 ) 

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

124 except gws.lib.net.Error as e: 

125 gws.log.error('geoservices request error', e) 

126 return {}