44and storing outputs.
55"""
66import logging
7- import os
87import shutil
98import time
109from pathlib import Path
10+ from typing import Dict
11+ from typing import Tuple
1112from typing import Union
1213from . import utils
1314
1415
1516logger = logging .getLogger (__name__ )
1617
17- def read_config (path_to_config_file ):
18+
19+ def read_config (config_file : Union [str , Path ]) -> Dict [str , str ]:
1820 """Read config from given config file.
1921
2022 Load paths from config file and save them into dict.
2123
2224 Args:
23- path_to_config_file : Path to the config file.
25+ config_file : Path to the config file.
2426
2527 Returns:
2628 Dictionary containing paths to work directory and all sub-directories.
2729 """
2830 config = {}
29- with open (path_to_config_file , "r" , encoding = "utf8" ) as f :
31+ with Path ( config_file ). open (encoding = "utf8" ) as f :
3032 for line in f :
3133 (key , val ) = line .split ("=" )
32- config [key ] = val .rstrip (' \n ' )
34+ config [key ] = val .rstrip (" \n " )
3335
3436 validate_config (config )
3537
3638 return config
3739
3840
3941def validate_config (config : Union [Path , dict ]):
40- if isinstance (config , Path ):
41- config = read_config (config )
42- elif not isinstance (config , dict ):
43- raise ValueError ("The input to validate_config should be either a Path or dict"
44- f" object, but a { type (config )} object was passed." )
42+ """Validate the config file."""
43+ if isinstance (config , dict ):
44+ cfg = config # For proper type narrowing understood by Mypy.
45+ elif isinstance (config , Path ):
46+ cfg = read_config (config )
47+ else :
48+ raise ValueError (
49+ "The input to validate_config should be either a Path or dict"
50+ f" object, but a { type (config )} object was passed."
51+ )
4552
4653 # TODO: add check if the input data directories/file exist, and return clear error to user.
47- _ = utils .check_location_fmt (config ["Location" ])
48- utils .check_time_fmt (config ["StartTime" ], config ["EndTime" ])
54+ _ = utils .check_location_fmt (cfg ["Location" ])
55+ utils .check_time_fmt (cfg ["StartTime" ], cfg ["EndTime" ])
4956
5057
51- def create_io_dir (config ) :
58+ def create_io_dir (config : Dict ) -> Tuple [ Path , Path , Path ] :
5259 """Create input directory and copy required files.
5360
5461 Work flow executor to create work directory and all sub-directories.
5562
5663 Returns:
57- Path (string) to input, output directory and config file for every station/forcing.
64+ Path to input, output directory and config file for every station/forcing.
5865 """
5966 # get start time with the format Y-M-D-HM
60- timestamp = time .strftime (' %Y-%m-%d-%H%M' )
67+ timestamp = time .strftime (" %Y-%m-%d-%H%M" )
6168
6269 loc , fmt = utils .check_location_fmt (config ["Location" ])
6370 if fmt == "site" :
@@ -67,14 +74,14 @@ def create_io_dir(config):
6774 site_name = "global"
6875 latstr = f"{ loc [0 ]:.3f} " .replace ("." , "-" )
6976 lonstr = f"{ loc [1 ]:.3f} " .replace ("." , "-" )
70- latstr = f"N{ latstr } " if loc [0 ] >= 0 else f"S{ latstr [1 :]} "
71- lonstr = f"E{ lonstr } " if loc [1 ] >= 0 else f"W{ lonstr [1 :]} "
77+ latstr = f"N{ latstr } " if loc [0 ] >= 0 else f"S{ latstr [1 :]} " # type: ignore
78+ lonstr = f"E{ lonstr } " if loc [1 ] >= 0 else f"W{ lonstr [1 :]} " # type: ignore
7279 input_dir_name = f"global_{ latstr } _{ lonstr } _{ timestamp } "
7380 else :
7481 raise NotImplementedError ()
7582
7683 # create input directory
77- work_dir = utils .to_absolute_path (config [' WorkDir' ])
84+ work_dir = utils .to_absolute_path (config [" WorkDir" ])
7885 input_dir = work_dir / "input" / input_dir_name
7986 input_dir .mkdir (parents = True , exist_ok = True )
8087 message = f"Prepare work directory { input_dir } for the location: { loc } "
@@ -90,13 +97,14 @@ def create_io_dir(config):
9097 logger .info ("%s" , message )
9198
9299 # update config file for ForcingFileName and InputPath
93- config_file_path = _update_config_file (input_dir , output_dir ,
94- config , site_name , timestamp )
100+ config_file_path = _update_config_file (
101+ input_dir , output_dir , config , site_name , timestamp # type: ignore
102+ )
95103
96- return str ( input_dir ), str ( output_dir ) , config_file_path
104+ return input_dir , output_dir , config_file_path
97105
98106
99- def _copy_data (input_dir , config ) :
107+ def _copy_data (input_dir : Path , config : dict ) -> None :
100108 """Copy required data to the work directory.
101109
102110 Create sub-directories inside the work directory and copy data.
@@ -105,34 +113,46 @@ def _copy_data(input_dir, config):
105113 input_dir: Path to the input directory.
106114 config: Dictionary containing all the paths.
107115 """
108- folder_list_vegetation = ["directional" , "fluspect_parameters" , "leafangles" ,
109- "radiationdata" , "soil_spectrum" ]
116+ folder_list_vegetation = [
117+ "directional" ,
118+ "fluspect_parameters" ,
119+ "leafangles" ,
120+ "radiationdata" ,
121+ "soil_spectrum" ,
122+ ]
110123 for folder in folder_list_vegetation :
111- os .makedirs (input_dir / folder , exist_ok = True )
112- shutil .copytree (str (config [folder ]), str (input_dir / folder ), dirs_exist_ok = True )
124+ (input_dir / folder ).mkdir (parents = True , exist_ok = True )
125+ shutil .copytree (
126+ str (config [folder ]), str (input_dir / folder ), dirs_exist_ok = True
127+ )
113128
114129 # copy input_data.xlsx
115130 shutil .copy (str (config ["input_data" ]), str (input_dir ))
116131
117132
118- def _update_config_file (input_dir , output_dir , config , site_name , timestamp ):
133+ def _update_config_file (
134+ input_dir : Path ,
135+ output_dir : Path ,
136+ config : Dict ,
137+ site_name : str ,
138+ timestamp : str ,
139+ ) -> Path :
119140 """Update config file for each station.
120141
121142 Create config file for each forcing/station under the work directory.
122143
123144 Args:
124- ncfile: Name of forcing file.
125145 input_dir: Path to the input directory.
126146 output_dir: Path to the output directory.
127147 config: Dictionary containing all the paths.
128- site_name: Either inferred from forcing file , or ' latlon' for global data.
148+ site_name: The site name (eg. "FI-Hyy") , or " latlon" for global data.
129149 timestamp: Timestamp when creating the config file.
130150
131151 Returns:
132152 Path to updated config file.
133153 """
134154 config_file_path = input_dir / f"{ site_name } _{ timestamp } _config.txt"
135- with open (config_file_path , 'w' , encoding = "utf8" ) as f :
155+ with config_file_path . open (mode = "w" , encoding = "utf8" ) as f :
136156 for key , value in config .items ():
137157 if key == "InputPath" :
138158 update_entry = f"{ key } ={ str (input_dir )} /\n "
@@ -143,4 +163,4 @@ def _update_config_file(input_dir, output_dir, config, site_name, timestamp):
143163
144164 f .write (update_entry )
145165
146- return str ( config_file_path )
166+ return config_file_path
0 commit comments