Coverage for gws-app/gws/plugin/alkis/data/exporter.py: 0%
126 statements
« prev ^ index » next coverage.py v7.16.2, created at 2026-10-05 13:35 +0200
1"""Export of Flurstuecke to CSV or GeoJSON."""
3from typing import Iterable, Optional, cast
5import gws
6import gws.base.feature
7import gws.base.model
8import gws.lib.intl
9import gws.lib.mime
10import gws.lib.jsonx
11import gws.plugin.csv_helper
13from gws.lib.cli import ProgressIndicator
14from . import types as dt
15from . import index
18class Config(gws.ConfigWithAccess):
19 """Export of Flurstück data to CSV or GeoJSON."""
21 type: str
22 """Export type."""
23 title: Optional[str]
24 """Title to display in the UI."""
25 model: Optional[gws.ext.config.model]
26 """Fields to export."""
27 models: Optional[list[gws.ext.config.model]]
28 """Field sets the user can choose from."""
31class Args(gws.Data):
32 """Arguments for the export operation."""
34 fsList: Iterable[dt.Flurstueck]
35 """Iterable of Flurstuecke to export."""
36 user: gws.User
37 """User who requested the export."""
38 progress: Optional[ProgressIndicator]
39 """Progress indicator to update during export."""
40 path: str
41 """Path to save the export."""
42 models: list['Model']
43 """List of models to export."""
46_DEFAULT_FIELDS = [
47 gws.Config(type='text', name='fs_flurstueckskennzeichen', title='Flurstückskennzeichen'),
48 gws.Config(type='text', name='fs_recs_gemeinde_text', title='Gemeinde'),
49 gws.Config(type='text', name='fs_recs_gemarkung_code', title='Gemarkungsnummer'),
50 gws.Config(type='text', name='fs_recs_gemarkung_text', title='Gemarkung'),
51 gws.Config(type='text', name='fs_recs_flurnummer', title='Flurnummer'),
52 gws.Config(type='text', name='fs_recs_zaehler', title='Zähler'),
53 gws.Config(type='text', name='fs_recs_nenner', title='Nenner'),
54 gws.Config(type='text', name='fs_recs_flurstuecksfolge', title='Folge'),
55 gws.Config(type='float', name='fs_recs_amtlicheFlaeche', title='Fläche'),
56 gws.Config(type='float', name='fs_recs_x', title='X'),
57 gws.Config(type='float', name='fs_recs_y', title='Y'),
58]
61class Model(gws.base.model.Object):
62 """Export model, a set of fields written to the export file.
64 The field names are flat Flurstueck keys, see ``index.flatten_fs``.
65 """
67 withEigentuemer: bool
68 """Whether the model has fields with owner (Eigentuemer) data."""
69 withBuchung: bool
70 """Whether the model has fields with land register (Buchung) data."""
72 def configure(self):
73 self.configure_model()
76class ModelProps(gws.Props):
77 title: str
80class Props(gws.Props):
81 title: str
82 models: list[ModelProps]
85_READ_WRITE_PERMISSIONS = gws.Config(read='allow all', write='allow all')
88class Object(gws.Node):
89 """ALKIS exporter.
91 Writes Flurstuecke to a CSV or GeoJSON file, using export models the user
92 can choose from.
93 """
95 models: list[Model]
96 """Export models the user can choose from."""
97 title: str
98 """Title to display in the UI."""
99 type: str
100 """Export type, ``csv`` or ``geojson``."""
101 mimeType: str
102 """Mime type of the export file."""
104 def configure(self):
105 self.type = self.cfg('type') or 'csv'
106 if self.type == 'csv':
107 self.mimeType = gws.lib.mime.CSV
108 elif self.type == 'geojson':
109 self.mimeType = gws.lib.mime.JSON
110 else:
111 raise gws.ConfigurationError(f'Unsupported export type: {self.type}')
113 self.title = self.cfg('title') or self.type
115 p = self.cfg('models')
116 if not p:
117 p = [self.cfg('model') or gws.Config(fields=_DEFAULT_FIELDS)]
119 self.models = []
120 for m in p:
121 mod = cast(
122 Model,
123 self.create_child(
124 Model,
125 m,
126 # NB need write permissions for `feature.to_record`
127 permissions=_READ_WRITE_PERMISSIONS,
128 ),
129 )
130 mod.withEigentuemer = any('eigentuemer' in fld.name for fld in mod.fields)
131 mod.withBuchung = any('buchung' in fld.name for fld in mod.fields)
132 self.models.append(mod)
134 def props_with_flags(self, user: gws.User, withEigentuemer: bool, withBuchung: bool) -> Optional[Props]:
135 """Return exporter properties with the models available to the user.
137 Models with owner or land register fields are left out if the user
138 has no access to that data.
140 Args:
141 user: The user.
142 withEigentuemer: Whether the user may see owner data.
143 withBuchung: Whether the user may see land register data.
145 Returns:
146 Exporter properties, or ``None`` if the user cannot use the exporter
147 or no model is available.
148 """
150 if not user.can_use(self):
151 return
153 models = []
154 for mod in self.get_models(user):
155 if not user.can_use(mod):
156 continue
157 if mod.withEigentuemer and not withEigentuemer:
158 continue
159 if mod.withBuchung and not withBuchung:
160 continue
161 models.append(ModelProps(uid=mod.uid, title=mod.title))
162 if not models:
163 return
165 return Props(uid=self.uid, title=self.title, models=models)
167 def get_models(self, user: gws.User, uids: Optional[list[str]] = None) -> list[Model]:
168 """Return export models the user can use.
170 Unknown uids are logged and ignored.
172 Args:
173 user: The user.
174 uids: Model uids to select. If empty, all models are considered.
176 Returns:
177 A list of models.
178 """
180 if not uids:
181 ms = self.models
182 else:
183 s = set(uids)
184 ms = []
185 for m in self.models:
186 if m.uid in s:
187 ms.append(m)
188 s.remove(m.uid)
189 if s:
190 gws.log.warning(f'Unknown model uids: {s}')
192 return [m for m in ms if user.can_use(m)]
194 def run(self, args: Args):
195 """Export a list of Flurstuecke to a file.
197 Args:
198 args: Export arguments, including the Flurstuecke, the models and the output path.
200 Raises:
201 ``gws.NotFoundError``: If the export type is not supported.
202 """
204 if self.type == 'csv':
205 return self._export_csv(args)
206 if self.type == 'geojson':
207 return self._export_geojson(args)
208 raise gws.NotFoundError(f'Unsupported export format')
210 def _export_csv(self, args: Args):
211 """Write rows to a CSV file."""
213 csv_helper = cast(gws.plugin.csv_helper.Object, self.root.app.helper('csv'))
215 with open(args.path, 'wb') as fp:
216 writer = csv_helper.writer(gws.lib.intl.locale('de_DE'), stream_to=fp)
217 for row in self._iter_rows(args):
218 writer.write_dict(row)
220 def _export_geojson(self, args: Args):
221 """Write rows to a GeoJSON feature collection."""
223 with open(args.path, 'wb') as fp:
224 fp.write(b'{"type": "FeatureCollection", "features": [')
225 comma = b'\n '
226 for row in self._iter_rows(args, with_geometry=True):
227 shape = row.pop('fs_shape', None)
228 d = dict(
229 type='Feature',
230 properties=row,
231 geometry=cast(gws.Shape, shape).to_geojson() if shape else None,
232 )
233 fp.write(comma + gws.lib.jsonx.to_string(d, ensure_ascii=False).encode('utf8'))
234 comma = b',\n '
236 fp.write(b'\n]}\n')
238 def _iter_rows(self, args: Args, with_geometry=False):
239 """Yield flat, deduplicated rows for the Flurstuecke, keyed by field titles."""
241 if len(args.models) == 1:
242 export_model = args.models[0]
243 else:
244 export_model = cast(
245 Model,
246 self.root.create_temporary(
247 Model,
248 gws.Config(fields=[]),
249 permissions=_READ_WRITE_PERMISSIONS,
250 ),
251 )
252 for mod in args.models:
253 export_model.fields.extend(mod.fields)
255 all_keys = set(fld.name for fld in export_model.fields)
256 mc = gws.ModelContext(
257 op=gws.ModelOperation.read,
258 target=gws.ModelReadTarget.searchResults,
259 user=args.user,
260 )
261 row_hashes = set()
263 for fs in args.fsList:
264 if args.progress:
265 args.progress.update(1)
267 for atts in index.flatten_fs(fs, all_keys):
268 # create a 'raw' feature from attributes and convert it to a record
269 # so that dynamic fields can be computed
271 feature = gws.base.feature.new(model=export_model, attributes=atts)
272 for fld in export_model.fields:
273 fld.to_record(feature, mc)
275 # fmt: off
276 row = {
277 fld.title: feature.record.attributes.get(fld.name, '')
278 for fld in export_model.fields
279 if not fld.isHidden
280 }
281 # fmt: on
283 h = gws.u.sha256(row)
284 if h in row_hashes:
285 continue
287 row_hashes.add(h)
288 if with_geometry:
289 row['fs_shape'] = fs.shape
290 yield row