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

1"""Nominatim model.""" 

2 

3from typing import Optional 

4 

5import gws 

6import gws.base.model 

7import gws.base.feature 

8import gws.lib.shape 

9import gws.lib.net 

10import gws.lib.jsonx 

11import gws.gis.source 

12import gws.lib.crs 

13import gws.lib.bounds 

14 

15 

16@gws.ext.config.model('nominatim') 

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

18 """Data model for places and addresses from Nominatim.""" 

19 

20 country: Optional[str] 

21 """Countries to limit the results to.""" 

22 language: Optional[str] 

23 """Preferred language of the results.""" 

24 

25 

26@gws.ext.object.model('nominatim') 

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

28 """Nominatim model object. 

29 

30 A read-only model whose features are results of the Nominatim search API. 

31 """ 

32 

33 country: str 

34 """Countries to limit the results to.""" 

35 language: str 

36 """Preferred language of the results.""" 

37 

38 serviceUrl = 'https://nominatim.openstreetmap.org/search' 

39 

40 def configure(self): 

41 self.country = self.cfg('country') 

42 self.language = self.cfg('language') 

43 self.uidName = 'uid' 

44 self.geometryName = 'geometry' 

45 self.loadingStrategy = gws.FeatureLoadingStrategy.all 

46 

47 def props(self, user): 

48 return gws.u.merge( 

49 super().props(user), 

50 canCreate=False, 

51 canDelete=False, 

52 canWrite=False, 

53 ) 

54 

55 def find_features(self, search, user, **kwargs): 

56 params = { 

57 'q': search.keyword, 

58 'addressdetails': 1, 

59 'polygon_geojson': 1, 

60 'format': 'json', 

61 'bounded': 1, 

62 'dedupe': 1, 

63 'limit': search.limit, 

64 } 

65 

66 if self.language: 

67 params['accept-language'] = self.cfg('language') 

68 

69 if self.country: 

70 params['countrycodes'] = self.cfg('country') 

71 

72 params['viewbox'] = gws.lib.bounds.transform(search.shape.bounds(), gws.lib.crs.WGS84).extent 

73 

74 features = [] 

75 

76 for rec in self._query(params): 

77 uid = rec.get('place_id') or rec.get('osm_id') 

78 geom = rec.pop('geojson', {}) 

79 

80 if not geom: 

81 gws.log.debug(f'SKIP {uid}: no geometry') 

82 continue 

83 

84 shape = gws.lib.shape.from_geojson(geom, gws.lib.crs.WGS84, always_xy=True).transformed_to(search.shape.crs) 

85 if not shape.intersects(search.shape): 

86 gws.log.debug(f'SKIP {uid}: no intersection') 

87 continue 

88 

89 features.append(self.feature_from_record( 

90 gws.FeatureRecord(uid=uid, shape=shape, attributes=self._normalize(rec)), 

91 user)) 

92 

93 return sorted(features, key=lambda f: (f.get('name'), f.get('osm_class'), f.get('osm_type'))) 

94 

95 def _query(self, params) -> list[dict]: 

96 """Send a search request to the service and return the results, or an empty list on error.""" 

97 try: 

98 res = gws.lib.net.http_request(self.serviceUrl, params=params) 

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

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

101 gws.log.error('nominatim request error', e) 

102 return [] 

103 

104 def _normalize(self, rec): 

105 """Flatten the address, add name and OSM class/type attributes, and drop coordinate attributes.""" 

106 # merge the address subrec into the main 

107 

108 if 'address' in rec: 

109 for k, v in rec.pop('address').items(): 

110 rec['address_' + k] = v 

111 

112 # ensure basic address fields 

113 

114 rec['address_road'] = rec.get('address_road') or '' 

115 rec['address_building'] = rec.get('address_building') or rec.get('address_house_number') or '' 

116 rec['address_city'] = rec.get('address_city') or rec.get('address_town') or rec.get('address_village') or '' 

117 rec['address_country'] = rec.get('address_country') or '' 

118 

119 # find out the "name" 

120 # the problem is there's no fixed key for the name (the key depends on the class) 

121 # however, display_name seems to always start with the 'name' 

122 

123 dn = rec.get('display_name', '') 

124 rec['name'] = dn.split(',')[0].strip() 

125 

126 # rename 'class' and 'type' for easier templating 

127 

128 rec['osm_class'] = rec.pop('class', '') 

129 rec['osm_type'] = rec.pop('type', '') 

130 

131 # remove geographic attributes 

132 

133 for k in 'boundingbox', 'lat', 'lon': 

134 rec.pop(k, None) 

135 

136 return rec