Coverage for pds_crawler/models/pdssp_models.py: 93%

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

2# pds-crawler - ETL to index PDS data to pdssp 

3# Copyright (C) 2023 - CNES (Jean-Christophe Malapert for Pôle Surfaces Planétaires) 

4# This file is part of pds-crawler <https://github.com/pdssp/pds_crawler> 

5# SPDX-License-Identifier: LGPL-3.0-or-later 

6""" 

7Module Name: 

8 pdssp_models 

9 

10Description: 

11 Provides methods to make consistent the IDs and the used extensions in the PDSSP project. 

12 

13Classes: 

14 PdsspModel 

15""" 

16import logging 

17from datetime import datetime 

18from typing import Any 

19from typing import Dict 

20from typing import List 

21from typing import Optional 

22from typing import Union 

23 

24from astropy.time import Time 

25from pymarsseason import Hemisphere 

26from pymarsseason import PyMarsSeason 

27from pymarsseason import Season 

28 

29logger = logging.getLogger(__name__) 

30 

31 

32class PdsspModel: 

33 PREFIX_PDSSP: str = "urn:pdssp" 

34 

35 @staticmethod 

36 def create_lab_id(lab: str) -> str: 

37 id = lab.replace("/", "_") # STAC compatible 

38 return f"{PdsspModel.PREFIX_PDSSP}:{id.lower()}" 

39 

40 @staticmethod 

41 def create_body_id(lab: str, solar_body_id: str) -> str: 

42 id = solar_body_id.replace("/", "_") # STAC compatible 

43 return f"{PdsspModel.PREFIX_PDSSP}:{lab.lower()}:body:{id.lower()}" 

44 

45 @staticmethod 

46 def create_mission_id(lab: str, mission_id: str) -> str: 

47 id = mission_id.replace("/", "_") # STAC compatible 

48 return f"{PdsspModel.PREFIX_PDSSP}:{lab.lower()}:mission:{id.lower()}" 

49 

50 @staticmethod 

51 def create_platform_id(lab: str, plateform_id: str) -> str: 

52 id = plateform_id.replace("/", "_") # STAC compatible 

53 return ( 

54 f"{PdsspModel.PREFIX_PDSSP}:{lab.lower()}:plateform:{id.lower()}" 

55 ) 

56 

57 @staticmethod 

58 def create_instru_id(lab: str, instrument_id: str) -> str: 

59 id = instrument_id.replace("/", "_") # STAC compatible 

60 return f"{PdsspModel.PREFIX_PDSSP}:{lab.lower()}:instru:{id.lower()}" 

61 

62 @staticmethod 

63 def create_collection_id(lab: str, collection_id: str) -> str: 

64 id = collection_id.replace("/", "_") # STAC compatible 

65 return ( 

66 f"{PdsspModel.PREFIX_PDSSP}:{lab.lower()}:collection:{id.lower()}" 

67 ) 

68 

69 @staticmethod 

70 def create_ssys_extension(body: str) -> Dict[str, Union[List, Dict]]: 

71 return { 

72 "stac_extensions": [ 

73 "https://raw.githubusercontent.com/thareUSGS/ssys/main/json-schema/schema.json" 

74 ], 

75 "extra_fields": {"ssys:targets": [body]}, 

76 } 

77 

78 @staticmethod 

79 def add_mars_keywords_if_mars( 

80 body_id: str, 

81 ode_id: str, 

82 lat: Optional[float], 

83 slong: Optional[float], 

84 date: datetime, 

85 ) -> Dict[str, Any]: 

86 """Add specific keywords for Mars 

87 

88 If Mars, add solar longitude and season otherwise 

89 returns an empty dictionary 

90 

91 Args: 

92 body_id (str): solar body 

93 ode_id (str): ODE ID 

94 lat (Optional[float]): latitude of the footprint 

95 slong (Optional[float]): soalr longitude 

96 datetime (datetime): observation date 

97 

98 Returns: 

99 Dict[str, Any]: _description_ 

100 """ 

101 mars: Dict[str, Any] = dict() 

102 if body_id.upper() != "MARS": 

103 return mars 

104 if lat is None: 

105 logger.warning(f"No latitude for Mars : {ode_id}") 

106 return mars 

107 py_mars_season: Dict[ 

108 Hemisphere | str, Season | float 

109 ] = PyMarsSeason().compute_season_from_time( 

110 Time(date.isoformat(), format="isot", scale="utc") 

111 ) 

112 if slong is None: 

113 mars["Solar_longitude"] = py_mars_season["ls"] 

114 

115 hemisphere: Hemisphere = ( 

116 Hemisphere.NORTH if float(lat) > 0 else Hemisphere.SOUTH 

117 ) 

118 mars["season"] = hemisphere.value 

119 return mars