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

1"""Command line commands for the ALKIS index and export.""" 

2 

3from typing import Optional, cast 

4 

5import gws 

6import gws.base.action 

7import gws.lib.jsonx 

8import gws.lib.osx 

9from gws.lib.cli import ProgressIndicator 

10 

11from .data import types as dt 

12from .data import exporter, index, indexer 

13from . import action 

14 

15 

16class CreateIndexParams(gws.CliParams): 

17 """Parameters for the ``alkisIndex`` command.""" 

18 

19 projectUid: Optional[str] 

20 """Project uid.""" 

21 force: bool = False 

22 """Drop and rebuild the index even if it is complete.""" 

23 

24 

25class StatusParams(gws.CliParams): 

26 """Parameters for the ``alkisStatus`` command.""" 

27 

28 projectUid: Optional[str] 

29 """Project uid.""" 

30 

31 

32class DumpParams(gws.CliParams): 

33 """Parameters for the ``alkisDump`` command.""" 

34 

35 projectUid: Optional[str] 

36 """Project uid.""" 

37 fs: str 

38 """Flurstueck uids.""" 

39 path: str 

40 """Path to save the dump.""" 

41 

42 

43# @TODO options to filter data and configure groups 

44 

45 

46class ExportParams(gws.CliParams): 

47 """Parameters for the ``alkisExport`` command.""" 

48 

49 projectUid: Optional[str] 

50 """Project uid.""" 

51 exporterUid: Optional[str] 

52 """Exporter uid, the first usable exporter if empty.""" 

53 modelUids: Optional[str] 

54 """Uids of the export models, all usable models if empty.""" 

55 path: str 

56 """Path to save the export.""" 

57 

58 

59@gws.ext.object.cli('alkis') 

60class Object(gws.Node): 

61 """ALKIS command line commands.""" 

62 

63 act: action.Object 

64 """The ``alkis`` action the commands work with.""" 

65 

66 def _prepare(self, project_uid): 

67 """Load the configuration and find the ``alkis`` action, exit if there is none.""" 

68 

69 root = gws.load_root() 

70 self.act = cast(action.Object, gws.base.action.get_action_for_cli(root, 'alkis', project_uid)) 

71 if not self.act: 

72 exit(1) 

73 

74 @gws.ext.command.cli('alkisIndex') 

75 def do_index(self, p: CreateIndexParams): 

76 """Create the ALKIS index.""" 

77 

78 self._prepare(p.projectUid) 

79 

80 s = self.act.ix.status() 

81 if s.complete and not p.force: 

82 gws.log.info(f'ALKIS index ok') 

83 return 

84 

85 gws.log.info(f'indexing db={self.act.db.uid} dataSchema={self.act.dataSchema} indexSchema={self.act.indexSchema}') 

86 indexer.run(self.act.ix, self.act.dataSchema, with_force=p.force) 

87 

88 @gws.ext.command.cli('alkisStatus') 

89 def do_status(self, p: StatusParams): 

90 """Display the status of the ALKIS index.""" 

91 

92 self._prepare(p.projectUid) 

93 

94 s = self.act.ix.status() 

95 if s.complete: 

96 gws.log.info(f'ALKIS index ok') 

97 return 

98 if s.missing: 

99 gws.log.error(f'ALKIS index not found') 

100 return 

101 

102 gws.log.warning(f'ALKIS index incomplete: basic={s.basic} eigentuemer={s.eigentuemer} buchung={s.buchung}') 

103 

104 @gws.ext.command.cli('alkisExport') 

105 def do_export(self, p: ExportParams): 

106 """Export all parcels in the ALKIS index.""" 

107 

108 self._prepare(p.projectUid) 

109 

110 s = self.act.ix.status() 

111 if s.missing: 

112 gws.log.error(f'ALKIS index missing') 

113 exit(1) 

114 

115 sys_user = self.act.root.app.authMgr.systemUser 

116 exp = self.act.get_exporter(p.exporterUid, sys_user) 

117 if not exp: 

118 gws.log.error(f'ALKIS exporter not found') 

119 exit(3) 

120 

121 models = exp.get_models(sys_user, gws.u.to_list(p.modelUids)) 

122 if not models: 

123 gws.log.error(f'no models found for ALKIS exporter') 

124 exit(4) 

125 

126 qo = dt.FlurstueckQueryOptions( 

127 withEigentuemer=True, 

128 withBuchung=True, 

129 withHistorySearch=False, 

130 withHistoryDisplay=False, 

131 displayThemes=[ 

132 'lage', 

133 'gebaeude', 

134 'nutzung', 

135 'festlegung', 

136 'bewertung', 

137 'buchung', 

138 'eigentuemer', 

139 ], 

140 pageSize=1000, 

141 ) 

142 

143 total = self.act.ix.count_all(qo) 

144 fs = self.act.ix.iter_all(qo) 

145 

146 with ProgressIndicator('export', total) as progress: 

147 exp.run( 

148 exporter.Args( 

149 fsList=fs, 

150 user=sys_user, 

151 models=models, 

152 path=p.path, 

153 progress=progress, 

154 ) 

155 ) 

156 

157 @gws.ext.command.cli('alkisKeys') 

158 def do_keys(self, p: gws.CliParams): 

159 """Print the keys available for ALKIS exports.""" 

160 

161 d = index.all_flat_keys() 

162 for key, typ in d.items(): 

163 print(f'{typ.__name__:7} {key}') 

164 

165 @gws.ext.command.cli('alkisDump') 

166 def do_dump(self, p: DumpParams): 

167 """Write the internal representation of parcels to a JSON file.""" 

168 

169 self._prepare(p.projectUid) 

170 

171 s = self.act.ix.status() 

172 if s.missing: 

173 gws.log.error(f'ALKIS index missing') 

174 exit(1) 

175 

176 qo = dt.FlurstueckQueryOptions( 

177 withEigentuemer=True, 

178 withBuchung=True, 

179 withHistorySearch=False, 

180 withHistoryDisplay=True, 

181 displayThemes=[ 

182 'lage', 

183 'gebaeude', 

184 'nutzung', 

185 'festlegung', 

186 'bewertung', 

187 'buchung', 

188 'eigentuemer', 

189 ], 

190 ) 

191 

192 fs = self.act.ix.load_flurstueck(gws.u.to_list(p.fs), qo) 

193 js = [index.serialize(f, encode_enum_pairs=False) for f in fs] 

194 gws.u.write_file(p.path, gws.lib.jsonx.to_pretty_string(js))