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. (added in 8.2)""" 

26 title: Optional[str] 

27 """Title to display in the ui. (added in 8.2)""" 

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

29 """Export model. (added in 8.2)""" 

30 

31 

32class Args(gws.Data): 

33 """Arguments for the export operation.""" 

34 

35 fsList: Iterable[dt.Flurstueck] 

36 """Iterable of Flurstuecke to export.""" 

37 user: gws.User 

38 """User who requested the export.""" 

39 progress: Optional[ProgressIndicator] 

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

41 path: str 

42 """Path to save the export.""" 

43 

44 

45_DEFAULT_FIELDS = [ 

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

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

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

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

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

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

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

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

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

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

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

57] 

58 

59 

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

61 def configure(self): 

62 self.configure_model() 

63 

64 

65class Object(gws.Node): 

66 model: Model 

67 title: str 

68 type: str 

69 mimeType: str 

70 usedKeys: set[str] 

71 withEigentuemer: bool 

72 withBuchung: bool 

73 

74 def configure(self): 

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

76 if self.type == 'csv': 

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

78 elif self.type == 'geojson': 

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

80 else: 

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

82 

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

84 

85 p = self.cfg('model') or gws.Config(fields=_DEFAULT_FIELDS) 

86 self.model = cast( 

87 Model, 

88 self.create_child( 

89 Model, 

90 p, 

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

92 permissions=gws.Config(read='allow all', write='allow all'), 

93 ), 

94 ) 

95 

96 self.withEigentuemer = any('namensnummer' in fld.name for fld in self.model.fields) 

97 self.withBuchung = any('buchung' in fld.name for fld in self.model.fields) 

98 

99 def run(self, args: Args): 

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

101 

102 if self.type == 'csv': 

103 return self._export_csv(args) 

104 if self.type == 'geojson': 

105 return self._export_geojson(args) 

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

107 

108 def _export_csv(self, args: Args): 

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

110 

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

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

113 for row in self._iter_rows(args): 

114 writer.write_dict(row) 

115 

116 def _export_geojson(self, args: Args): 

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

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

119 comma = b'\n ' 

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

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

122 d = dict( 

123 type='Feature', 

124 properties=row, 

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

126 ) 

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

128 comma = b',\n ' 

129 

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

131 

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

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

134 

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

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

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

138 

139 record: 

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

141 

142 flat list: 

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

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

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

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

147 

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

149 """ 

150 

151 all_keys = set(fld.name for fld in self.model.fields) 

152 mc = gws.ModelContext(op=gws.ModelOperation.read, target=gws.ModelReadTarget.searchResults, user=args.user) 

153 row_hashes = set() 

154 

155 for fs in args.fsList: 

156 if args.progress: 

157 args.progress.update(1) 

158 

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

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

161 # so that dynamic fields can be computed 

162 

163 feature = gws.base.feature.new(model=self.model, attributes=atts) 

164 for fld in self.model.fields: 

165 fld.to_record(feature, mc) 

166 

167 # fmt: off 

168 row = { 

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

170 for fld in self.model.fields 

171 if not fld.isHidden 

172 } 

173 # fmt: on 

174 

175 h = gws.u.sha256(row) 

176 if h in row_hashes: 

177 continue 

178 

179 row_hashes.add(h) 

180 if with_geometry: 

181 row['fs_shape'] = fs.shape 

182 yield row