refactor: update functions in tsdb.utils.file with check_path;

tsdb/utils/file.py

@@ -121,18 +121,19 @@ def purge_path(path: str, ignore_errors: bool = True) -> None:
 
 
 def determine_data_home():
+    # default path
+    default_path = check_path("~/.pypots/tsdb")
+
     # read data_home from the config file
+    # data_home may be changed by users, hence not necessarily equal to the default path
     config = read_configs()
     data_home_path = config.get("path", "data_home")
-    # replace '~' with the absolute path if existing in the path
-    data_home_path = data_home_path.replace("~", os.path.expanduser("~"))
+    data_home_path = check_path(data_home_path)
 
     # old cached dataset dir path used in TSDB v0.2
-    old_cached_dataset_dir_02 = os.path.join(
-        os.path.expanduser("~"), ".tsdb_cached_datasets"
-    )
+    old_cached_dataset_dir_02 = check_path("~/.tsdb_cached_datasets")
     # old cached dataset dir path used in TSDB v0.4
-    old_cached_dataset_dir_04 = os.path.join(os.path.expanduser("~"), ".tsdb")
+    old_cached_dataset_dir_04 = check_path("~/.tsdb")
 
     if os.path.exists(old_cached_dataset_dir_02) or os.path.exists(
         old_cached_dataset_dir_04
@@ -150,15 +151,18 @@ def determine_data_home():
         # use the path directly, may be in a portable disk
         cached_dataset_dir = data_home_path
     else:
-        # use the default path for initialization,
-        # e.g. `data_home_path` in a portable disk but the disk is not connected
-        default_path = os.path.join(os.path.expanduser("~"), ".pypots", "tsdb")
-        cached_dataset_dir = default_path
-        if os.path.abspath(data_home_path) != os.path.abspath(default_path):
+        # if the preset data_home path does not exist,
+        # e.g. `data_home_path` is in a portable disk that is not connected
+        # then use the default path
+        if check_path(data_home_path) != check_path(default_path):
             logger.warning(
-                f"‼️ The preset data_home path '{data_home_path}' doesn't exist. "
-                f"Using the default path '{default_path}'"
+                f"❗️ The preset data_home {data_home_path} doesn't exist. "
+                f"This may be caused by the portable disk not connected."
             )
+            logger.warning(f"‼️ Using the default path {default_path} for now")
+
+        cached_dataset_dir = default_path
+
     return cached_dataset_dir
 
 
@@ -174,30 +178,33 @@ def migrate(old_path: str, new_path: str) -> None:
         The new path of the dataset.
 
     """
+    # check both old_path and new_path
+    old_path = check_path(old_path)
+    new_path = check_path(new_path)
+
+    # check if old_path exists
     if not os.path.exists(old_path):
         raise FileNotFoundError(f"Given old_path {old_path} does not exist.")
 
+    # create new_path if not exists
     if not os.path.exists(new_path):
-        # if new_path does not exist, just rename the old_path into it
-        new_parent_dir = os.path.abspath(os.path.join(new_path, ".."))
-        if not os.path.exists(new_parent_dir):
-            os.makedirs(new_parent_dir, exist_ok=True)
+        os.makedirs(new_path, exist_ok=True)
+    else:
+        logger.warning(f"‼️ Note that new_path {new_path} already exists.")
 
-    logger.warning(f"‼️ Please note that new_path {new_path} already exists.")
-    # if new_path exists, we have to move everything from old_path into it
     all_old_files = os.listdir(old_path)
     for f in all_old_files:
         old_f_path = os.path.join(old_path, f)
+
         if os.path.isdir(old_f_path):
             new_f_path = os.path.join(new_path, f)
             shutil.copytree(old_f_path, new_f_path)
         else:
             shutil.move(old_f_path, new_path)
     shutil.rmtree(old_path, ignore_errors=True)
 
-    logger.info(
-        f"Successfully migrated {old_path} to {new_path}, and deleted {old_path}"
-    )
+    logger.info(f"Successfully migrated {old_path} to {new_path}")
+    logger.info(f"Purged the old path {old_path}")
 
 
 def migrate_cache(target_path: str) -> None:
@@ -209,6 +216,7 @@ def migrate_cache(target_path: str) -> None:
         The new path for TSDB to store cached datasets.
 
     """
+    target_path = check_path(target_path)
     cached_dataset_dir = determine_data_home()
     migrate(cached_dataset_dir, target_path)
     config_parser = read_configs()