Coverage for gws-app/gws/plugin/alkis/data/exporter.py: 0%

1"""ALKIS exporter. 

2 

3Export Flurstuecke to CSV or GeoJSON. 

4""" 

5 

6from typing import Iterable, Optional, cast 

7 

8import gws 

9import gws.base.feature 

10import gws.base.model 

11import gws.lib.intl 

12import gws.lib.mime 

13import gws.lib.jsonx 

14import gws.plugin.csv_helper 

15 

16from gws.lib.cli import ProgressIndicator 

17from . import types as dt 

18from . import index 

19 

20 

21class Config(gws.ConfigWithAccess): 

22 """Export configuration""" 

23 

24 type: str 

25 """Export type.""" 

26 title: Optional[str] 

27 """Title to display in the ui.""" 

28 model: Optional[gws.ext.config.model] 

29 """Export model.""" 

30 models: Optional[list[gws.ext.config.model]] 

31 """Export models.""" 

32 

33 

34class Args(gws.Data): 

35 """Arguments for the export operation.""" 

36 

37 fsList: Iterable[dt.Flurstueck] 

38 """Iterable of Flurstuecke to export.""" 

39 user: gws.User 

40 """User who requested the export.""" 

41 progress: Optional[ProgressIndicator] 

42 """Progress indicator to update during export.""" 

43 path: str 

44 """Path to save the export.""" 

45 models: list['Model'] 

46 """List of models to export.""" 

47 

48 

49_DEFAULT_FIELDS = [ 

50 gws.Config(type='text', name='fs_flurstueckskennzeichen', title='Flurstückskennzeichen'), 

51 gws.Config(type='text', name='fs_recs_gemeinde_text', title='Gemeinde'), 

52 gws.Config(type='text', name='fs_recs_gemarkung_code', title='Gemarkungsnummer'), 

53 gws.Config(type='text', name='fs_recs_gemarkung_text', title='Gemarkung'), 

54 gws.Config(type='text', name='fs_recs_flurnummer', title='Flurnummer'), 

55 gws.Config(type='text', name='fs_recs_zaehler', title='Zähler'), 

56 gws.Config(type='text', name='fs_recs_nenner', title='Nenner'), 

57 gws.Config(type='text', name='fs_recs_flurstuecksfolge', title='Folge'), 

58 gws.Config(type='float', name='fs_recs_amtlicheFlaeche', title='Fläche'), 

59 gws.Config(type='float', name='fs_recs_x', title='X'), 

60 gws.Config(type='float', name='fs_recs_y', title='Y'), 

61] 

62 

63 

64class Model(gws.base.model.Object): 

65 withEigentuemer: bool 

66 withBuchung: bool 

67 

68 def configure(self): 

69 self.configure_model() 

70 

71 

72class ModelProps(gws.Props): 

73 title: str 

74 

75 

76class Props(gws.Props): 

77 title: str 

78 models: list[ModelProps] 

79 

80 

81_READ_WRITE_PERMISSIONS = gws.Config(read='allow all', write='allow all') 

82 

83 

84class Object(gws.Node): 

85 models: list[Model] 

86 title: str 

87 type: str 

88 mimeType: str 

89 usedKeys: set[str] 

90 

91 def configure(self): 

92 self.type = self.cfg('type') or 'csv' 

93 if self.type == 'csv': 

94 self.mimeType = gws.lib.mime.CSV 

95 elif self.type == 'geojson': 

96 self.mimeType = gws.lib.mime.JSON 

97 else: 

98 raise gws.ConfigurationError(f'Unsupported export type: {self.type}') 

99 

100 self.title = self.cfg('title') or self.type 

101 

102 p = self.cfg('models') 

103 if not p: 

104 p = [self.cfg('model') or gws.Config(fields=_DEFAULT_FIELDS)] 

105 

106 self.models = [] 

107 for m in p: 

108 mod = cast( 

109 Model, 

110 self.create_child( 

111 Model, 

112 m, 

113 # NB need write permissions for `feature.to_record` 

114 permissions=_READ_WRITE_PERMISSIONS, 

115 ), 

116 ) 

117 mod.withEigentuemer = any('eigentuemer' in fld.name for fld in mod.fields) 

118 mod.withBuchung = any('buchung' in fld.name for fld in mod.fields) 

119 self.models.append(mod) 

120 

121 def props_with_flags(self, user: gws.User, withEigentuemer: bool, withBuchung: bool) -> Optional[Props]: 

122 if not user.can_use(self): 

123 return 

124 

125 models = [] 

126 for mod in self.get_models(user): 

127 if not user.can_use(mod): 

128 continue 

129 if mod.withEigentuemer and not withEigentuemer: 

130 continue 

131 if mod.withBuchung and not withBuchung: 

132 continue 

133 models.append(ModelProps(uid=mod.uid, title=mod.title)) 

134 if not models: 

135 return 

136 

137 return Props(uid=self.uid, title=self.title, models=models) 

138 

139 def get_models(self, user: gws.User, uids: Optional[list[str]] = None) -> list[Model]: 

140 if not uids: 

141 ms = self.models 

142 else: 

143 s = set(uids) 

144 ms = [] 

145 for m in self.models: 

146 if m.uid in s: 

147 ms.append(m) 

148 s.remove(m.uid) 

149 if s: 

150 gws.log.warning(f'Unknown model uids: {s}') 

151 

152 return [m for m in ms if user.can_use(m)] 

153 

154 def run(self, args: Args): 

155 """Export a Flurstueck list to a file.""" 

156 

157 if self.type == 'csv': 

158 return self._export_csv(args) 

159 if self.type == 'geojson': 

160 return self._export_geojson(args) 

161 raise gws.NotFoundError(f'Unsupported export format') 

162 

163 def _export_csv(self, args: Args): 

164 csv_helper = cast(gws.plugin.csv_helper.Object, self.root.app.helper('csv')) 

165 

166 with open(args.path, 'wb') as fp: 

167 writer = csv_helper.writer(gws.lib.intl.locale('de_DE'), stream_to=fp) 

168 for row in self._iter_rows(args): 

169 writer.write_dict(row) 

170 

171 def _export_geojson(self, args: Args): 

172 with open(args.path, 'wb') as fp: 

173 fp.write(b'{"type": "FeatureCollection", "features": [') 

174 comma = b'\n ' 

175 for row in self._iter_rows(args, with_geometry=True): 

176 shape = row.pop('fs_shape', None) 

177 d = dict( 

178 type='Feature', 

179 properties=row, 

180 geometry=cast(gws.Shape, shape).to_geojson() if shape else None, 

181 ) 

182 fp.write(comma + gws.lib.jsonx.to_string(d, ensure_ascii=False).encode('utf8')) 

183 comma = b',\n ' 

184 

185 fp.write(b'\n]}\n') 

186 

187 def _iter_rows(self, args: Args, with_geometry=False): 

188 """Iterate over a Flurstueck list and yield flat rows (dicts). 

189 

190 The Flurstueck structure, as created by our indexer, is deeply nested. 

191 We flatten it, creating a dict 'nested_key->value'. For list values, we repeat the dict 

192 for each item in the list, thus creating a product of all lists, e.g. 

193 

194 record: 

195 a:x, b:[1,2], c:[3,4] 

196 

197 flat list: 

198 a:x, b:1, c:3 

199 a:x, b:1, c:4 

200 a:x, b:2, c:3 

201 a:x, b:2, c:4 

202 

203 @TODO: with certain combinations of keys this can explode very quickly 

204 """ 

205 

206 if len(args.models) == 1: 

207 export_model = args.models[0] 

208 else: 

209 export_model = cast( 

210 Model, 

211 self.root.create_temporary( 

212 Model, 

213 gws.Config(fields=[]), 

214 permissions=_READ_WRITE_PERMISSIONS, 

215 ), 

216 ) 

217 for mod in args.models: 

218 export_model.fields.extend(mod.fields) 

219 

220 all_keys = set(fld.name for fld in export_model.fields) 

221 mc = gws.ModelContext( 

222 op=gws.ModelOperation.read, 

223 target=gws.ModelReadTarget.searchResults, 

224 user=args.user, 

225 ) 

226 row_hashes = set() 

227 

228 for fs in args.fsList: 

229 if args.progress: 

230 args.progress.update(1) 

231 

232 for atts in index.flatten_fs(fs, all_keys): 

233 # create a 'raw' feature from attributes and convert it to a record 

234 # so that dynamic fields can be computed 

235 

236 feature = gws.base.feature.new(model=export_model, attributes=atts) 

237 for fld in export_model.fields: 

238 fld.to_record(feature, mc) 

239 

240 # fmt: off 

241 row = { 

242 fld.title: feature.record.attributes.get(fld.name, '') 

243 for fld in export_model.fields 

244 if not fld.isHidden 

245 } 

246 # fmt: on 

247 

248 h = gws.u.sha256(row) 

249 if h in row_hashes: 

250 continue 

251 

252 row_hashes.add(h) 

253 if with_geometry: 

254 row['fs_shape'] = fs.shape 

255 yield row