Coverage for gws-app/gws/base/map/action.py: 62%

1"""Map related commands.""" 

2 

3from typing import Optional 

4 

5import gws 

6import gws.base.action 

7import gws.base.legend 

8import gws.lib.bounds 

9import gws.lib.crs 

10import gws.gis.render 

11import gws.lib.image 

12import gws.lib.intl 

13import gws.lib.jsonx 

14import gws.lib.mime 

15import gws.lib.uom 

16 

17gws.ext.new.action('map') 

18 

19 

20class Config(gws.base.action.Config): 

21 """Configuration for the map action.""" 

22 

23 pass 

24 

25 

26class Props(gws.base.action.Props): 

27 pass 

28 

29 

30class GetBoxRequest(gws.Request): 

31 bbox: gws.Extent 

32 width: int 

33 height: int 

34 layerUid: str 

35 crs: Optional[gws.CrsName] 

36 dpi: Optional[int] 

37 compositeLayerUids: Optional[list[str]] 

38 

39 

40class GetXyzRequest(gws.Request): 

41 layerUid: str 

42 x: int 

43 y: int 

44 z: int 

45 

46 

47class GetLegendRequest(gws.Request): 

48 layerUid: str 

49 

50 

51class ImageResponse(gws.Response): 

52 content: bytes 

53 mime: str 

54 

55 

56class DescribeLayerRequest(gws.Request): 

57 layerUid: str 

58 

59 

60class DescribeLayerResponse(gws.Request): 

61 content: str 

62 

63 

64class GetFeaturesRequest(gws.Request): 

65 bbox: Optional[gws.Extent] 

66 layerUid: str 

67 modelUid: Optional[str] 

68 crs: Optional[gws.CrsName] 

69 resolution: Optional[float] 

70 limit: int = 0 

71 views: Optional[list[str]] 

72 

73 

74class GetFeaturesResponse(gws.Response): 

75 features: list[gws.FeatureProps] 

76 

77 

78_GET_FEATURES_LIMIT = 10000 

79 

80 

81class Object(gws.base.action.Object): 

82 _empty_pixel = gws.lib.mime.PNG, gws.lib.image.empty_pixel() 

83 

84 @gws.ext.command.api('mapGetBox') 

85 def api_get_box(self, req: gws.WebRequester, p: GetBoxRequest) -> ImageResponse: 

86 """Get a part of the map inside a bounding box""" 

87 mime, content = self._get_box(req, p) 

88 return ImageResponse(mime=mime, content=content) 

89 

90 @gws.ext.command.get('mapGetBox') 

91 def http_get_box(self, req: gws.WebRequester, p: GetBoxRequest) -> gws.ContentResponse: 

92 mime, content = self._get_box(req, p) 

93 return gws.ContentResponse(mime=mime, content=content) 

94 

95 @gws.ext.command.api('mapGetXYZ') 

96 def api_get_xyz(self, req: gws.WebRequester, p: GetXyzRequest) -> ImageResponse: 

97 """Get an XYZ tile""" 

98 mime, content = self._get_xyz(req, p) 

99 return ImageResponse(mime=mime, content=content) 

100 

101 @gws.ext.command.get('mapGetXYZ') 

102 def http_get_xyz(self, req: gws.WebRequester, p: GetXyzRequest) -> gws.ContentResponse: 

103 mime, content = self._get_xyz(req, p) 

104 return gws.ContentResponse(mime=mime, content=content) 

105 

106 @gws.ext.command.api('mapGetLegend') 

107 def api_get_legend(self, req: gws.WebRequester, p: GetLegendRequest) -> ImageResponse: 

108 """Get a legend for a layer""" 

109 mime, content = self._get_legend(req, p) 

110 return ImageResponse(mime=mime, content=content) 

111 

112 @gws.ext.command.get('mapGetLegend') 

113 def http_get_legend(self, req: gws.WebRequester, p: GetLegendRequest) -> gws.ContentResponse: 

114 mime, content = self._get_legend(req, p) 

115 return gws.ContentResponse(mime=mime, content=content) 

116 

117 @gws.ext.command.api('mapDescribeLayer') 

118 def describe_layer(self, req: gws.WebRequester, p: DescribeLayerRequest) -> DescribeLayerResponse: 

119 project = req.user.require_project(p.projectUid) 

120 layer = req.user.require_layer(p.layerUid) 

121 tpl = self.root.app.templateMgr.find_template('layer.description', where=[layer, project], user=req.user) 

122 

123 if not tpl: 

124 return DescribeLayerResponse(content='') 

125 

126 res = tpl.render(gws.TemplateRenderInput(args={'layer': layer}, locale=gws.lib.intl.locale(p.localeUid, project.localeUids), user=req.user)) 

127 

128 return DescribeLayerResponse(content=res.content) 

129 

130 @gws.ext.command.api('mapGetFeatures') 

131 def api_get_features(self, req: gws.WebRequester, p: GetFeaturesRequest) -> GetFeaturesResponse: 

132 """Get a list of features in a bounding box""" 

133 

134 propses = self._get_features(req, p) 

135 return GetFeaturesResponse(features=propses) 

136 

137 @gws.ext.command.get('mapGetFeatures') 

138 def http_get_features(self, req: gws.WebRequester, p: GetFeaturesRequest) -> gws.ContentResponse: 

139 # @TODO the response should be geojson FeatureCollection 

140 

141 propses = self._get_features(req, p) 

142 js = gws.lib.jsonx.to_string({'features': propses}) 

143 

144 return gws.ContentResponse(mime=gws.lib.mime.JSON, content=js) 

145 

146 ## 

147 

148 def _get_box(self, req: gws.WebRequester, p: GetBoxRequest): 

149 layer = req.user.require_layer(p.layerUid) 

150 lri = gws.LayerRenderInput(type=gws.LayerRenderInputType.box, user=req.user, extraParams={}) 

151 

152 if p.compositeLayerUids: 

153 lri.extraParams['compositeLayerUids'] = p.compositeLayerUids 

154 

155 lri.view = gws.gis.render.map_view_from_bbox( 

156 crs=gws.lib.crs.get(p.crs) or layer.mapCrs, 

157 bbox=p.bbox, 

158 size=(p.width, p.height, gws.Uom.px), 

159 dpi=gws.lib.uom.OGC_SCREEN_PPI, 

160 rotation=0, 

161 ) 

162 

163 try: 

164 lro = layer.render(lri) 

165 if lro and lro.content: 

166 return gws.lib.mime.PNG, lro.content 

167 except Exception: 

168 gws.log.exception() 

169 

170 return self._empty_pixel 

171 

172 def _get_xyz(self, req: gws.WebRequester, p: GetXyzRequest): 

173 layer = req.user.require_layer(p.layerUid) 

174 lri = gws.LayerRenderInput(type=gws.LayerRenderInputType.xyz, user=req.user, x=p.x, y=p.y, z=p.z) 

175 lro = None 

176 

177 gws.debug.time_start(f'RENDER_XYZ layer={p.layerUid} lri={lri!r}') 

178 try: 

179 lro = layer.render(lri) 

180 except: 

181 gws.log.exception() 

182 gws.debug.time_end() 

183 

184 if not lro: 

185 return self._empty_pixel 

186 

187 content = lro.content 

188 

189 # for public tiled layers, write tiles to the web cache 

190 # so they will be subsequently served directly by nginx 

191 

192 # if content and gws.u.is_public_object(layer) and layer.has_cache: 

193 # path = layer.url_path('tile') 

194 # path = path.replace('{x}', str(p.x)) 

195 # path = path.replace('{y}', str(p.y)) 

196 # path = path.replace('{z}', str(p.z)) 

197 # gws.gis.cache.store_in_web_cache(path, content) 

198 

199 return gws.lib.mime.PNG, content 

200 

201 def _get_legend(self, req: gws.WebRequester, p: GetLegendRequest): 

202 layer = req.user.require_layer(p.layerUid) 

203 lro = layer.render_legend() 

204 content = gws.base.legend.output_to_bytes(lro) 

205 if content: 

206 return lro.mime, content 

207 return self._empty_pixel 

208 

209 def _image_response(self, lro: gws.LayerRenderOutput) -> ImageResponse: 

210 # @TODO content-dependent mime type 

211 # @TODO in-image errors 

212 if lro and lro.content: 

213 return ImageResponse(mime='image/png', content=lro.content) 

214 return ImageResponse(mime='image/png', content=gws.lib.image.empty_pixel()) 

215 

216 def _get_features(self, req: gws.WebRequester, p: GetFeaturesRequest) -> list[gws.FeatureProps]: 

217 layer = req.user.require_layer(p.layerUid) 

218 project = layer.find_closest(gws.ext.object.project) 

219 

220 crs = gws.lib.crs.get(p.crs) or layer.mapCrs 

221 

222 bounds = layer.bounds 

223 if p.bbox: 

224 bounds = gws.lib.bounds.from_extent(p.bbox, crs) 

225 

226 search = gws.SearchQuery( 

227 bounds=bounds, 

228 project=project, 

229 layers=[layer], 

230 limit=_GET_FEATURES_LIMIT, 

231 ) 

232 

233 features = layer.find_features(search, req.user) 

234 if not features: 

235 return [] 

236 

237 tpl = self.root.app.templateMgr.find_template(f'feature.label', where=[layer, project], user=req.user) 

238 if tpl: 

239 for feature in features: 

240 feature.render_views([tpl], project=project, layer=self, user=req.user) 

241 

242 mc = gws.ModelContext( 

243 op=gws.ModelOperation.read, 

244 target=gws.ModelReadTarget.map, 

245 user=req.user, 

246 ) 

247 

248 return [f.model.feature_to_view_props(f, mc) for f in features]