Coverage for src/geodense/main.py: 96%

93 statements  

« prev     ^ index     » next       coverage.py v7.6.1, created at 2024-10-11 14:11 +0000

1import argparse 

2import logging 

3import os 

4import sys 

5from collections.abc import Callable 

6from enum import Enum 

7from functools import wraps 

8from typing import Any, Literal 

9 

10from rich_argparse import RichHelpFormatter 

11 

12from geodense import __version__, add_stderr_logger 

13from geodense.lib import ( 

14 SUPPORTED_FILE_FORMATS, 

15 check_density_file, 

16 densify_file, 

17) 

18from geodense.models import DEFAULT_MAX_SEGMENT_LENGTH, GeodenseError 

19 

20logger = logging.getLogger("geodense") 

21 

22 

23def cli_exception_handler(f: Callable) -> Callable: 

24 @wraps(f) 

25 def decorated(*args, **kwargs) -> Any: # noqa: ANN002, ANN003, ANN401 

26 try: 

27 return f(*args, **kwargs) 

28 except GeodenseError as e: 

29 logger.error(e) 

30 sys.exit(1) 

31 except Exception as e: # unexpected exception, show stacktrace by calling logger.exception 

32 logger.exception(e) 

33 sys.exit(1) 

34 

35 return decorated 

36 

37 

38@cli_exception_handler 

39def densify_cmd( # noqa: PLR0913 

40 input_file: str, 

41 output_file: str, 

42 overwrite: bool = False, 

43 max_segment_length: float | None = None, 

44 in_projection: bool = False, 

45 src_crs: str | None = None, 

46) -> None: 

47 densify_file( 

48 input_file, 

49 output_file, 

50 overwrite, 

51 max_segment_length, 

52 in_projection, 

53 src_crs, 

54 ) 

55 

56 

57@cli_exception_handler 

58def check_density_cmd( # noqa: PLR0913 

59 input_file: str, 

60 max_segment_length: float, 

61 overwrite: bool = False, 

62 in_projection: bool = False, 

63 src_crs: str | None = None, 

64 density_check_report_path: str | None = None, 

65) -> None: 

66 print(overwrite) 

67 

68 check_status, density_check_report_path, nr_line_segments = check_density_file( 

69 input_file, 

70 max_segment_length, 

71 density_check_report_path, 

72 src_crs, 

73 in_projection=in_projection, 

74 overwrite=overwrite, 

75 ) 

76 

77 status = "OK" if check_status else "FAILED" 

78 status_message = f"density-check {status} for file {input_file} with max-segment-length: {max_segment_length}" 

79 

80 print(status_message) # print status message for both OK and FAILED status 

81 

82 if check_status: 

83 sys.exit(0) 

84 else: 

85 print( 

86 f"{nr_line_segments} line segments in data exceed max-segment-length {max_segment_length}, line segment geometries written to GeoJSON FeatureCollection: {density_check_report_path}" 

87 ) 

88 sys.exit(1) 

89 

90 

91def main() -> None: 

92 input_file_help = ( 

93 "any valid GeoJSON file, accepted GeoJSON objects: FeatureCollection, Feature, Geometry and GeometryCollection " 

94 ) 

95 source_crs_help = "override source CRS, if not specified then the CRS found in the GeoJSON input file will be used; format: $AUTH:$CODE; for example: EPSG:4326" 

96 verbose_help = "verbose output" 

97 max_segment_length_help = f"max allowed segment length in meters; default: {DEFAULT_MAX_SEGMENT_LENGTH}" 

98 

99 parser = argparse.ArgumentParser( 

100 prog="geodense", 

101 description="Check density and densify geometries using the geodesic (ellipsoidal great-circle) calculation for accurate CRS transformations", 

102 epilog="Created by https://www.nsgi.nl/", 

103 formatter_class=RichHelpFormatter, 

104 ) 

105 parser.add_argument("-v", "--version", action="version", version=__version__) 

106 subparsers = parser.add_subparsers() 

107 

108 densify_parser = subparsers.add_parser( 

109 "densify", 

110 formatter_class=parser.formatter_class, 

111 description="Densify (multi)polygon and (multi)linestring geometries along the geodesic (ellipsoidal great-circle), in base CRS (geographic) in case of projected source CRS. Supports GeoJSON as input file format. When supplying 3D coordinates, the height is linear interpolated for both geographic CRSs with ellipsoidal height and for compound CRSs with physical height.", 

112 ) 

113 densify_parser.add_argument( 

114 "input_file", 

115 type=lambda x: is_json_file_arg(parser, x, "input_file", exist_required=FileRequired.exist), 

116 help=input_file_help, 

117 ) 

118 densify_parser.add_argument( 

119 "output_file", 

120 type=lambda x: is_json_file_arg(parser, x, "output_file", exist_required=FileRequired.either), 

121 help="output file path", 

122 ) 

123 

124 densify_parser.add_argument( 

125 "--max-segment-length", 

126 "-m", 

127 type=float, 

128 default=DEFAULT_MAX_SEGMENT_LENGTH, 

129 help=max_segment_length_help, 

130 ) 

131 

132 densify_parser.add_argument( 

133 "--in-projection", 

134 "-p", 

135 action="store_true", 

136 default=False, 

137 help="densify using linear interpolation in source projection instead of the geodesic, not applicable when source CRS is geographic", 

138 ) 

139 densify_parser.add_argument( 

140 "--overwrite", 

141 "-o", 

142 action="store_true", 

143 default=False, 

144 help="overwrite output file if exists", 

145 ) 

146 

147 densify_parser.add_argument("-v", "--verbose", action="store_true", default=False, help=verbose_help) 

148 

149 densify_parser.add_argument( 

150 "--src-crs", 

151 "-s", 

152 type=str, 

153 help=source_crs_help, 

154 default=None, 

155 ) 

156 

157 densify_parser.set_defaults(func=densify_cmd) 

158 

159 check_density_parser = subparsers.add_parser( 

160 "check-density", 

161 formatter_class=parser.formatter_class, 

162 description="Check density of (multi)polygon and (multi)linestring geometries based on geodesic (ellipsoidal great-circle) distance, in base CRS (geographic) in case of projected source CRS. \ 

163 When result of check is OK the program will return with exit code 0, when result \ 

164 is FAILED the program will return with exit code 1. The density-check report is a GeoJSON FeatureCollection containing line segments exceeding the max-segment-length treshold.", 

165 ) 

166 check_density_parser.add_argument( 

167 "input_file", 

168 type=lambda x: is_json_file_arg(parser, x, "input_file", exist_required=FileRequired.exist), 

169 help=input_file_help, 

170 ) 

171 check_density_parser.add_argument( 

172 "--max-segment-length", 

173 "-m", 

174 type=float, 

175 default=DEFAULT_MAX_SEGMENT_LENGTH, 

176 help=max_segment_length_help, 

177 ) 

178 check_density_parser.add_argument( 

179 "--src-crs", 

180 "-s", 

181 type=str, 

182 help=source_crs_help, 

183 default=None, 

184 ) 

185 check_density_parser.add_argument( 

186 "--in-projection", 

187 "-p", 

188 action="store_true", 

189 default=False, 

190 help="check density using linear interpolation in source projection instead of the geodesic, not applicable when source CRS is geographic", 

191 ) 

192 check_density_parser.add_argument( 

193 "--density-check-report-path", 

194 "-r", 

195 dest="density_check_report_path", 

196 required=False, 

197 help="density-check report path, when omitted a temp file will be used. Report is only generated when density-check fails.", 

198 metavar="FILE_PATH", 

199 type=lambda x: is_json_file_arg(parser, x, "density-check-report-path", FileRequired.either), 

200 ) 

201 check_density_parser.add_argument( 

202 "--overwrite", 

203 "-o", 

204 action="store_true", 

205 default=False, 

206 help="overwrite density-check report if exists", 

207 ) 

208 check_density_parser.add_argument("-v", "--verbose", action="store_true", default=False, help=verbose_help) 

209 check_density_parser.set_defaults(func=check_density_cmd) 

210 

211 parser._positionals.title = "commands" 

212 args = parser.parse_args() 

213 

214 try: 

215 add_stderr_logger(args.verbose) 

216 del args.verbose 

217 func = args.func 

218 del args.func 

219 func(**vars(args)) 

220 except AttributeError as _: 

221 parser.print_help(file=sys.stderr) 

222 sys.exit(1) 

223 

224 

225class FileRequired(Enum): 

226 exist: Literal["exist"] = "exist" 

227 not_exist: Literal["not_exist"] = "not_exist" 

228 either: Literal["either"] = "either" 

229 

230 

231def is_json_file_arg( 

232 parser: argparse.ArgumentParser, 

233 arg: str, 

234 arg_name: str, 

235 exist_required: FileRequired, 

236) -> str: 

237 _, file_ext = os.path.splitext(arg) 

238 unsupported_file_extension_msg = ( 

239 "unsupported file extension of {input_file}, received: {ext}, expected one of: {supported_ext}" 

240 ) 

241 if arg != "-" and file_ext not in SUPPORTED_FILE_FORMATS["GeoJSON"]: 

242 parser.error( 

243 unsupported_file_extension_msg.format( 

244 input_file=arg_name, 

245 ext=file_ext, 

246 supported_ext=", ".join(SUPPORTED_FILE_FORMATS["GeoJSON"]), 

247 ) 

248 ) 

249 if ( 

250 exist_required != FileRequired.either 

251 and arg != "-" 

252 and (not os.path.isfile(arg) if exist_required == FileRequired.exist else os.path.isfile(arg)) 

253 ): 

254 if exist_required: 

255 parser.error(f"{arg_name} {arg} does not exist") 

256 else: 

257 parser.error(f"{arg_name} {arg} exists") 

258 

259 if ( 

260 arg not in ["", "-"] 

261 and exist_required in [FileRequired.not_exist, FileRequired.either] 

262 and not os.path.exists(os.path.realpath(os.path.dirname(arg))) 

263 ): 

264 os.mkdir(os.path.realpath(os.path.dirname(arg))) 

265 return arg 

266 

267 

268if __name__ == "__main__": 

269 main() # pragma: no cover