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 cache: bool = False 

24 """Use object cache.""" 

25 

26 

27class StatusParams(gws.CliParams): 

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

29 

30 projectUid: Optional[str] 

31 """Project uid.""" 

32 

33 

34class DumpParams(gws.CliParams): 

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

36 

37 projectUid: Optional[str] 

38 """Project uid.""" 

39 fs: str 

40 """Flurstueck uids.""" 

41 path: str 

42 """Path to save the dump.""" 

43 

44 

45# @TODO options to filter data and configure groups 

46 

47 

48class ExportParams(gws.CliParams): 

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

50 

51 projectUid: Optional[str] 

52 """Project uid.""" 

53 exporterUid: Optional[str] 

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

55 modelUids: Optional[str] 

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

57 path: str 

58 """Path to save the export.""" 

59 

60 

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

62class Object(gws.Node): 

63 """ALKIS command line commands.""" 

64 

65 act: action.Object 

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

67 

68 def _prepare(self, project_uid): 

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

70 

71 root = gws.load_root() 

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

73 if not self.act: 

74 exit(1) 

75 

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

77 def do_index(self, p: CreateIndexParams): 

78 """Create the ALKIS index.""" 

79 

80 self._prepare(p.projectUid) 

81 

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

83 if s.complete and not p.force: 

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

85 return 

86 

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

88 indexer.run(self.act.ix, self.act.dataSchema, with_force=p.force, with_cache=p.cache) 

89 

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

91 def do_status(self, p: StatusParams): 

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

93 

94 self._prepare(p.projectUid) 

95 

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

97 if s.complete: 

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

99 return 

100 if s.missing: 

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

102 return 

103 

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

105 

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

107 def do_export(self, p: ExportParams): 

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

109 

110 self._prepare(p.projectUid) 

111 

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

113 if s.missing: 

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

115 exit(1) 

116 

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

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

119 if not exp: 

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

121 exit(3) 

122 

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

124 if not models: 

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

126 exit(4) 

127 

128 qo = dt.FlurstueckQueryOptions( 

129 withEigentuemer=True, 

130 withBuchung=True, 

131 withHistorySearch=False, 

132 withHistoryDisplay=False, 

133 displayThemes=[ 

134 'lage', 

135 'gebaeude', 

136 'nutzung', 

137 'festlegung', 

138 'bewertung', 

139 'buchung', 

140 'eigentuemer', 

141 ], 

142 pageSize=1000, 

143 ) 

144 

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

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

147 

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

149 exp.run( 

150 exporter.Args( 

151 fsList=fs, 

152 user=sys_user, 

153 models=models, 

154 path=p.path, 

155 progress=progress, 

156 ) 

157 ) 

158 

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

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

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

162 

163 d = index.all_flat_keys() 

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

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

166 

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

168 def do_dump(self, p: DumpParams): 

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

170 

171 self._prepare(p.projectUid) 

172 

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

174 if s.missing: 

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

176 exit(1) 

177 

178 qo = dt.FlurstueckQueryOptions( 

179 withEigentuemer=True, 

180 withBuchung=True, 

181 withHistorySearch=False, 

182 withHistoryDisplay=True, 

183 displayThemes=[ 

184 'lage', 

185 'gebaeude', 

186 'nutzung', 

187 'festlegung', 

188 'bewertung', 

189 'buchung', 

190 'eigentuemer', 

191 ], 

192 ) 

193 

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

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

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