Coverage for aixweather/imports/TRY.py: 81%

1""" 

2import DWD TRY data 

3""" 

4 

5import logging 

6import re 

7import random 

8import pandas as pd 

9 

10from aixweather.imports.utils_import import MetaData 

11 

12logger = logging.getLogger(__name__) 

13 

14def _handle_TRY_type(path: str) -> tuple: 

15 """ 

16 Determine the TRY format type based on the provided file path. 

17 

18 Args: 

19 path (str): The file path to the TRY dataset file. 

20 

21 Returns: 

22 tuple: A tuple containing the TRY year (int) and the header row number (int). 

23 Raises: 

24 ValueError: If the TRY format cannot be detected through the file name or is not supported. 

25 """ 

26 

27 ### get type of TRY, i.e. the year of the TRY 

28 TRY_year = None 

29 # Header_rows are the rows with general information of the dataset 

30 # Are skipped until variable declaration 

31 TRY_file_naming = { 

32 "TRY2004": {"year": 2004}, 

33 "TRY2010": {"year": 2010}, 

34 "TRY2015": {"year": 2015}, 

35 "TRY2045": {"year": 2045}, 

36 } 

37 

38 if path.endswith(".dat"): 

39 for key in TRY_file_naming.keys(): 

40 if key in path: 

41 TRY_year = TRY_file_naming[key]["year"] 

42 break 

43 if TRY_year is None: 

44 raise ValueError( 

45 f"TRY format could not be detected through file name," 

46 f" expected {[key for key in TRY_file_naming.keys()]} in the file name." 

47 ) 

48 if TRY_year == 2004 or TRY_year == 2010: 

49 raise ValueError(f"TRY format {TRY_year} is not supported.") 

50 

51 if TRY_year == 2015 or TRY_year == 2045: 

52 with open(path, "r") as file: 

53 for line_number, line in enumerate(file, start=1): 

54 if "***" in line: 

55 header_row = ( 

56 line_number - 1 - 1 

57 ) # -1 for header above *** and -1 for start to count at 0 

58 break 

59 

60 return TRY_year, header_row 

61 

62 

63def load_try_meta_from_file(path: str) -> MetaData: 

64 """ 

65 Load a TRY file from a specified path and parse the header for metadata. 

66 

67 Args: 

68 path (str): The file path to the TRY file to be loaded. 

69 

70 Returns: 

71 MetaData: An object containing the parsed metadata from the TRY file. 

72 """ 

73 

74 meta = MetaData() 

75 TRY_year, header_row = _handle_TRY_type(path) 

76 

77 ### load file to python 

78 header_lines = [] 

79 with open(path, "r") as file: 

80 for i, line in enumerate(file): 

81 if i >= header_row: 

82 break 

83 header_lines.append(line) 

84 

85 ### read raw meta data 

86 # Extract Rechtswert (Easting) 

87 rechtswert_line = next( 

88 line for line in header_lines if "Rechtswert" in line and ":" in line 

89 ) 

90 rechtswert = int(re.search(r":\s*(\d+) Meter", rechtswert_line).group(1)) 

91 

92 # Extract Hochwert (Northing) 

93 hochwert_line = next( 

94 line for line in header_lines if "Hochwert" in line and ":" in line 

95 ) 

96 hochwert = int(re.search(r":\s*(\d+) Meter", hochwert_line).group(1)) 

97 

98 # Extract Höhenlage (altitude) 

99 hoehenlage_line = next(line for line in header_lines if "Hoehenlage" in line) 

100 hoehenlage = int(re.search(r":\s*(\d+) Meter", hoehenlage_line).group(1)) 

101 

102 try: 

103 import geopandas as gpd 

104 from geopy.geocoders import Nominatim 

105 from shapely.geometry import Point 

106 except ImportError: 

107 raise ImportError("Optional dependency 'TRY' not installed. Conversion of longitude and " 

108 "latitude not possible and hence no radiation transformation.") 

109 

110 ### convert latitude and longitude 

111 # Create a GeoDataFrame with the provided coordinates 

112 # (using pyproj directly led to wrong calculation) 

113 gdf = gpd.GeoDataFrame( 

114 {"geometry": [Point(rechtswert, hochwert)]}, 

115 crs="EPSG:3034", # Original coordinate system 

116 ) 

117 

118 # Transform to WGS 84 

119 gdf_wgs84 = gdf.to_crs("EPSG:4326") 

120 

121 # get transformed coordinates, Get the longitude (x) and latitude (y) 

122 point_wgs84 = gdf_wgs84.geometry.iloc[0] 

123 longitude_wgs84 = point_wgs84.x 

124 latitude_wgs84 = point_wgs84.y 

125 

126 ### try to get city of location 

127 # Initialize Nominatim geolocator 

128 user_agent = f"aixweather_{str(random.randint(1, 1000))}" 

129 geolocator = Nominatim(user_agent=user_agent) 

130 # Perform reverse geocoding 

131 location = geolocator.reverse((latitude_wgs84, longitude_wgs84)) 

132 

133 # If you want specific components like city, state, etc. 

134 address = location.raw["address"] 

135 if "city" in address: 

136 city = address["city"] 

137 elif "town" in address: 

138 city = address["town"] 

139 elif "village" in address: 

140 city = address["village"] 

141 elif "hamlet" in address: 

142 city = address["hamlet"] 

143 elif "suburb" in address: 

144 city = address["suburb"] 

145 elif "locality" in address: 

146 city = address["locality"] 

147 else: 

148 city = meta.station_name 

149 

150 meta.station_name = city 

151 meta.input_source = f"TRY{TRY_year}" 

152 meta.try_year = TRY_year 

153 meta.altitude = hoehenlage 

154 meta.longitude = longitude_wgs84 

155 meta.latitude = latitude_wgs84 

156 

157 return meta 

158 

159 

160def load_try_from_file(path: str) -> pd.DataFrame: 

161 """ 

162 Import data from a TRY file and convert it into a DataFrame. 

163 

164 Args: 

165 path (str): The absolute path to the TRY file. 

166 

167 Returns: 

168 pd.DataFrame: A DataFrame containing the imported data from the TRY file. 

169 """ 

170 

171 

172 TRY_year, header_row = _handle_TRY_type(path) 

173 

174 ### load file to Dataframe 

175 weather_df = pd.read_table( 

176 filepath_or_buffer=path, 

177 header=header_row, 

178 sep='\s+', 

179 skip_blank_lines=False, 

180 encoding="latin", 

181 ) 

182 # drop first row cause empty 

183 weather_df = weather_df.iloc[1:] 

184 

185 return weather_df