Coverage for csvforwkt/csvforwkt.py: 83%

1# -*- coding: utf-8 -*- 

2"""This module contains the library to convert a body description in CSV to 

3WKT-CRS.""" 

4import collections 

5import logging 

6import os 

7from typing import cast 

8from typing import Dict 

9from typing import Tuple 

10 

11import pandas as pd # pylint: disable=import-error 

12 

13from ._version import __name_soft__ 

14from .body import IAU_REPORT 

15from .body import ReferenceShape 

16from .crs import BodyCrs 

17from .crs import ICrs 

18from .crs import Planetocentric 

19from .crs import Planetographic 

20from .crs import ProjectionBody 

21 

22logger = logging.getLogger(__name__) 

23 

24 

25class CsvforwktLib: 

26 """The library""" 

27 

28 def __init__( 

29 self, 

30 iau_report: str, 

31 iau_version: int, 

32 iau_doi: str, 

33 directory: str, 

34 *args, 

35 **kwargs, 

36 ): 

37 # pylint: disable=unused-argument 

38 if "level" in kwargs: 

39 CsvforwktLib._parse_level(kwargs["level"]) 

40 

41 self.__directory = directory 

42 # self.__data_file: str = "/home/malapert/Dev/test/csvForWKT/data/naifcodes_radii_m_wAsteroids_IAU2015.csv" 

43 self.__iau_report: str = iau_report 

44 self.__iau_version: int = iau_version 

45 self.__iau_doi: str = iau_doi 

46 self.__df_bodies: pd.DataFrame = self._init_iau_report() 

47 

48 @staticmethod 

49 def _parse_level(level: str): 

50 """Parse level name and set the rigt level for the logger. 

51 If the level is not known, the INFO level is set 

52 

53 Args: 

54 level (str): level name 

55 """ 

56 logger_main = logging.getLogger(__name_soft__) 

57 if level == "INFO": 

58 logger_main.setLevel(logging.INFO) 

59 elif level == "DEBUG": 

60 logger_main.setLevel(logging.DEBUG) 

61 elif level == "WARNING": 

62 logger_main.setLevel(logging.WARNING) 

63 elif level == "ERROR": 

64 logger_main.setLevel(logging.ERROR) 

65 elif level == "CRITICAL": 

66 logger_main.setLevel(logging.CRITICAL) 

67 elif level == "TRACE": 

68 logger_main.setLevel(logging.TRACE) # type: ignore # pylint: disable=no-member 

69 else: 

70 logger_main.warning( 

71 "Unknown level name : %s - setting level to INFO", level 

72 ) 

73 logger_main.setLevel(logging.INFO) 

74 

75 @property 

76 def iau_report(self) -> str: 

77 """The IAU report as CSV file. 

78 

79 :getter: Returns the path of the CSV file 

80 :type: str 

81 """ 

82 return self.__iau_report 

83 

84 @property 

85 def iau_version(self) -> int: 

86 """The IAU version. 

87 

88 :getter: Returns the IAU version 

89 :type: int 

90 """ 

91 return self.__iau_version 

92 

93 @property 

94 def iau_doi(self) -> str: 

95 """The IAU DOI. 

96 

97 :getter: Returns the IAU Doi 

98 :type: str 

99 """ 

100 return self.__iau_doi 

101 

102 @property 

103 def directory(self) -> str: 

104 """The output directory. 

105 

106 :getter: Returns the output directory 

107 :type: str 

108 """ 

109 return self.__directory 

110 

111 def _init_iau_report(self) -> pd.DataFrame: 

112 """Init the IAU_REPORT class.""" 

113 logger.info( 

114 f"Creating WKT-CRS for {self.iau_version} - {self.iau_doi} ..." 

115 ) 

116 IAU_REPORT.DOI_IAU = self.iau_doi 

117 IAU_REPORT.VERSION = str(self.iau_version) 

118 return pd.read_csv(self.iau_report) 

119 

120 def _skip_records(self): 

121 """Skip records when not (IAU2015_Semimajor == -1 and IAU2015_Axisb == -1 and IAU2015_Semiminor == -1)""" 

122 nb_records: int = self.__df_bodies.shape[0] 

123 query_result = self.__df_bodies.query( 

124 "IAU2015_Semimajor == -1 and IAU2015_Axisb == -1 and IAU2015_Semiminor == -1" 

125 ) 

126 idx2 = self.__df_bodies.index.difference(query_result.index) 

127 self.__df_bodies = self.__df_bodies.iloc[idx2] 

128 nb_records_for_processing: int = self.__df_bodies.shape[0] 

129 nb_records_skip: int = nb_records - nb_records_for_processing 

130 logger.info(f"\t\t{nb_records_skip} records have been skipped") 

131 

132 def _split_body(self) -> Tuple[pd.DataFrame, pd.DataFrame]: 

133 """Split the bodies in two parts : biaxial and triaxial 

134 

135 Triaxial bodies is defined when IAU2015_Semimajor, IAU2015_Semiminor, 

136 IAU2015_Axisb are different 

137 

138 Returns: 

139 Tuple[pd.DataFrame, pd.DataFrame]: biaxial and triaxial bodies 

140 """ 

141 triaxial: pd.DataFrame = self.__df_bodies.query( 

142 "IAU2015_Semimajor != IAU2015_Axisb \ 

143 and IAU2015_Semiminor != IAU2015_Axisb \ 

144 and IAU2015_Semiminor != IAU2015_Semimajor" 

145 ).copy() 

146 biaxial: pd.DataFrame = self.__df_bodies.drop(triaxial.index).copy() 

147 logger.info( 

148 f"\tSplit biaxial ({biaxial.shape[0]} records) and triaxial ({triaxial.shape[0]} records) ... OK" 

149 ) 

150 return biaxial, triaxial 

151 

152 def _is_sphere( # pylint: disable=no-self-use 

153 self, row: pd.Series 

154 ) -> bool: 

155 """Check if the current body is a sphere. 

156 

157 The verification is done by comparing : 

158 * IAU2015_Semimajor, IAU2015_Semiminor and IAU2015_Axisb 

159 

160 Args: 

161 row (pd.Series): current body description 

162 

163 Returns: 

164 bool: True when the shape off the body is a sphere otherwise False 

165 """ 

166 return ( 

167 row["IAU2015_Semimajor"] == row["IAU2015_Semiminor"] 

168 and row["IAU2015_Semimajor"] == row["IAU2015_Axisb"] 

169 ) 

170 

171 def _is_valid_flatenning(self, row: pd.Series): 

172 """Check if the flatenning is valid. 

173 

174 The verification is done by checking IAU2015_Semimajor with IAU2015_Semiminor: 

175 

176 Args: 

177 row (pd.Series): current body description 

178 

179 Returns: 

180 bool: True when the IAU2015_Semimajor >= IAU2015_Semiminor 

181 """ 

182 return row["IAU2015_Semimajor"] >= row["IAU2015_Semiminor"] 

183 

184 def _is_retrograde( # pylint: disable=no-self-use 

185 self, row: pd.Series 

186 ) -> bool: 

187 """Check if the rotation of the body is retrograde. 

188 

189 Args: 

190 row (pd.Series): current body description 

191 

192 Returns: 

193 bool: True when the rotation of the body is retrograde otherwise False 

194 """ 

195 return row["rotation"] == "Retrograde" 

196 

197 def _is_historic( # pylint: disable=no-self-use 

198 self, row: pd.Series 

199 ) -> bool: 

200 """Check if the body is part of a historical Coordinate Reference System. 

201 

202 Args: 

203 row (pd.Series): current body description 

204 

205 Returns: 

206 bool: True when the body is part of a historical CRS otherwise False 

207 """ 

208 return row["Body"] in ["Sun", "Earth", "Moon"] 

209 

210 def _process_body_crs_biaxial(self, body: pd.DataFrame) -> Dict[int, ICrs]: 

211 """Process biaxial bodies and avoid duplicate desriptions. 

212 

213 The Rules are the following to avoid duplication: 

214 * For each shape, we approximate the shape as a sphere and we create 

215 a planetocentric (or planetographic) reference frame with east direction. 

216 * if the shape is a sphere => planetocentric based on an ellipsoid is 

217 not needed else planetocentric description is created 

218 * if the shape is a sphere and (retrograde movement or historical) => 

219 planetographic is not needed because the planetographic = planetocentric 

220 else plantetographic description is created 

221 

222 Specific rule when semi_major_axis < semi_minor_axis: 

223 * Only sphere is used as mean_radius as radius 

224 

225 Args: 

226 body (pd.DataFrame): bodies 

227 ref_shape (ReferenceShape): Type of the shape 

228 

229 Returns: 

230 Dict[int, ICrs]: IAU code and CRS description 

231 """ 

232 crs: Dict[int, ICrs] = dict() 

233 for _, row in body.iterrows(): 

234 sphere_crs = Planetocentric(row, ReferenceShape.SPHERE) 

235 crs[sphere_crs.crs.iau_code] = sphere_crs.crs 

236 if not self._is_sphere(row) and self._is_valid_flatenning(row): 

237 ocentric_crs = Planetocentric(row, ReferenceShape.ELLIPSE) 

238 crs[ocentric_crs.crs.iau_code] = ocentric_crs.crs 

239 if not ( 

240 self._is_sphere(row) 

241 and (self._is_retrograde(row) or self._is_historic(row)) 

242 ) and self._is_valid_flatenning(row): 

243 ographic = Planetographic(row, ReferenceShape.ELLIPSE) 

244 crs[ographic.crs.iau_code] = ographic.crs 

245 logger.info(f"\t\tNumber of processed bodies: {len(crs.keys())}") 

246 return crs 

247 

248 def _process_body_crs_triaxial( # pylint: disable=no-self-use 

249 self, body: pd.DataFrame 

250 ) -> Dict[int, ICrs]: 

251 """Process triaxial bodies. 

252 

253 Args: 

254 body (pd.DataFrame): bodies 

255 ref_shape (ReferenceShape): Type of the shape 

256 

257 Returns: 

258 Dict[int, ICrs]: IAU code and CRS description 

259 """ 

260 crs: Dict[int, ICrs] = dict() 

261 for _, row in body.iterrows(): 

262 sphere_crs = Planetocentric(row, ReferenceShape.SPHERE) 

263 crs[sphere_crs.crs.iau_code] = sphere_crs.crs 

264 ocentric_crs = Planetocentric(row, ReferenceShape.TRIAXIAL) 

265 crs[ocentric_crs.crs.iau_code] = ocentric_crs.crs 

266 ographic = Planetographic(row, ReferenceShape.TRIAXIAL) 

267 crs[ographic.crs.iau_code] = ographic.crs 

268 logger.info(f"\t\tNumber of processed bodies: {len(crs.keys())}") 

269 return crs 

270 

271 def _process_body_projection_crs( # pylint: disable=no-self-use 

272 self, crs: Dict[int, ICrs] 

273 ) -> Dict[int, ICrs]: 

274 """Process the projection description based on a body CRS. 

275 

276 Args: 

277 crs (Dict[int, ICrs]): bodies CRS 

278 

279 Returns: 

280 Dict[int, ICrs]: projections CRS 

281 """ 

282 crs_projection: Dict[int, ICrs] = dict() 

283 for _, value in crs.items(): 

284 body_crs: BodyCrs = cast(BodyCrs, value) 

285 for projection in ProjectionBody.iter_projection(body_crs): 

286 crs_projection[projection.iau_code] = projection 

287 return crs_projection 

288 

289 def process(self) -> Dict[int, ICrs]: 

290 """Process the bodies. 

291 

292 Returns: 

293 Dict[int, ICrs]: CRS 

294 """ 

295 crs: Dict[int, ICrs] = {} 

296 nb_records: int = self.__df_bodies.shape[0] 

297 logger.info(f"\tNumber of bodies in IAU report {nb_records}") 

298 self._skip_records() 

299 nb_records = self.__df_bodies.shape[0] 

300 logger.info(f"\t\t{nb_records} records for processing") 

301 

302 biaxial: pd.DataFrame 

303 triaxial: pd.DataFrame 

304 biaxial, triaxial = self._split_body() 

305 

306 logger.info("\n\tProcessing of biaxial body") 

307 biaxial_crs: Dict[int, ICrs] = self._process_body_crs_biaxial(biaxial) 

308 crs.update(biaxial_crs) 

309 logger.info("\t\tprocess WKT for biaxial bodies ... OK") 

310 

311 logger.info("\n\tProcessing of triaxial body") 

312 triaxial_crs: Dict[int, ICrs] = self._process_body_crs_triaxial( 

313 triaxial 

314 ) 

315 crs.update(triaxial_crs) 

316 logger.info("\t\tprocess WKT for triaxial bodies ... OK") 

317 

318 logger.info( 

319 "\n\tProcessing of projected CRS for both baxial and triaxial" 

320 ) 

321 crs.update(self._process_body_projection_crs(crs)) 

322 logger.info("\t\tprocess WKT for projected CRS ... OK") 

323 

324 return collections.OrderedDict(sorted(crs.items())) 

325 

326 def save(self, crs: Dict[int, ICrs]): 

327 """Save the result as file 

328 

329 Args: 

330 crs (Dict[int, ICrs]): CRS 

331 """ 

332 with open( 

333 os.path.join(self.directory, "iau.wkt"), "w", encoding="utf-8" 

334 ) as file: 

335 for _, wkt in crs.items(): 

336 file.write(wkt.wkt()) 

337 file.write("\n\n") 

338 logger.info( 

339 f"\n\tSave the WKTs in {os.path.join(self.directory, 'iau.wkt')} ... OK" 

340 ) 

341 logger.info("Finished.")