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

1"""Command-line ALKIS commands.""" 

2 

3from typing import Optional, cast 

4 

5import gws 

6import gws.config 

7import gws.base.action 

8import gws.lib.jsonx 

9import gws.lib.osx 

10from gws.lib.cli import ProgressIndicator 

11 

12from .data import types as dt 

13from .data import exporter, index, indexer 

14from . import action 

15 

16gws.ext.new.cli('alkis') 

17 

18 

19class CreateIndexParams(gws.CliParams): 

20 projectUid: Optional[str] 

21 """Project uid.""" 

22 force: bool = False 

23 """Force indexing.""" 

24 cache: bool = False 

25 """Use object cache.""" 

26 

27 

28class StatusParams(gws.CliParams): 

29 projectUid: Optional[str] 

30 """Project uid.""" 

31 

32 

33class DumpParams(gws.CliParams): 

34 projectUid: Optional[str] 

35 """Project uid.""" 

36 fs: str 

37 """Flurstueck UIDs""" 

38 path: str 

39 """Path to save the dump.""" 

40 

41 

42# @TODO options to filter data and configure groups 

43 

44 

45class ExportParams(gws.CliParams): 

46 projectUid: Optional[str] 

47 """Project uid.""" 

48 exporterUid: Optional[str] 

49 """Export uid.""" 

50 modelUids: Optional[str] 

51 """List of model uids.""" 

52 path: str 

53 """Path to save the export.""" 

54 

55 

56class Object(gws.Node): 

57 act: action.Object 

58 

59 def _prepare(self, project_uid): 

60 root = gws.config.load() 

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

62 if not self.act: 

63 exit(1) 

64 

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

66 def do_index(self, p: CreateIndexParams): 

67 """Create the ALKIS index.""" 

68 

69 self._prepare(p.projectUid) 

70 

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

72 if s.complete and not p.force: 

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

74 return 

75 

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

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

78 

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

80 def do_status(self, p: StatusParams): 

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

82 

83 self._prepare(p.projectUid) 

84 

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

86 if s.complete: 

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

88 return 

89 if s.missing: 

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

91 return 

92 

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

94 

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

96 def do_export(self, p: ExportParams): 

97 """Export ALKIS data.""" 

98 

99 self._prepare(p.projectUid) 

100 

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

102 if s.missing: 

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

104 exit(1) 

105 

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

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

108 if not exp: 

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

110 exit(3) 

111 

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

113 if not models: 

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

115 exit(4) 

116 

117 qo = dt.FlurstueckQueryOptions( 

118 withEigentuemer=True, 

119 withBuchung=True, 

120 withHistorySearch=False, 

121 withHistoryDisplay=False, 

122 displayThemes=[ 

123 'lage', 

124 'gebaeude', 

125 'nutzung', 

126 'festlegung', 

127 'bewertung', 

128 'buchung', 

129 'eigentuemer', 

130 ], 

131 pageSize=1000, 

132 ) 

133 

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

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

136 

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

138 exp.run( 

139 exporter.Args( 

140 fsList=fs, 

141 user=sys_user, 

142 models=models, 

143 path=p.path, 

144 progress=progress, 

145 ) 

146 ) 

147 

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

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

150 """Print ALKIS export keys.""" 

151 

152 d = index.all_flat_keys() 

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

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

155 

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

157 def do_dump(self, p: DumpParams): 

158 """Dump internal representations of ALKIS objects.""" 

159 

160 self._prepare(p.projectUid) 

161 

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

163 if s.missing: 

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

165 exit(1) 

166 

167 qo = dt.FlurstueckQueryOptions( 

168 withEigentuemer=True, 

169 withBuchung=True, 

170 withHistorySearch=False, 

171 withHistoryDisplay=True, 

172 displayThemes=[ 

173 'lage', 

174 'gebaeude', 

175 'nutzung', 

176 'festlegung', 

177 'bewertung', 

178 'buchung', 

179 'eigentuemer', 

180 ], 

181 ) 

182 

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

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

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