diff --git a/tsdb/utils/file.py b/tsdb/utils/file.py index a188edf..7c03e43 100644 --- a/tsdb/utils/file.py +++ b/tsdb/utils/file.py @@ -15,95 +15,6 @@ from .logging import logger -def migrate(old_path: str, new_path: str) -> None: - """Migrate files in a directory from old_path to new_path. - - Parameters - ---------- - old_path: - The old path of the dataset. - - new_path: - The new path of the dataset. - - """ - if not os.path.exists(old_path): - raise FileNotFoundError(f"Given old_path {old_path} does not exist.") - - if os.path.exists(new_path): - 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) - else: - # 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.rename(old_path, new_path) - - logger.info( - f"Successfully migrated {old_path} to {new_path}, and deleted {old_path}" - ) - - -def determine_data_home(): - # read data_home from the config file - 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("~")) - old_cached_dataset_dir = os.path.join( - os.path.expanduser("~"), ".tsdb_cached_datasets" - ) - - if os.path.exists(old_cached_dataset_dir): - # use the old path and warn the user - logger.warning( - "‼️ Detected the home dir of the old version TSDB. " - "Since v0.3, TSDB has changed the default cache dir to '~/.tsdb'. " - "Auto migrating downloaded datasets to the new path. " - ) - cached_dataset_dir = data_home_path - migrate(old_cached_dataset_dir, cached_dataset_dir) - elif os.path.exists(data_home_path): - # use the path directly, may be in a portable disk - cached_dataset_dir = data_home_path - else: - # use the default path - default_path = os.path.join(os.path.expanduser("~"), ".tsdb") - cached_dataset_dir = default_path - if os.path.abspath(data_home_path) != os.path.abspath(default_path): - logger.warning( - f"‼️ The preset data_home path '{data_home_path}' doesn't exist. " - f"Using the default path '{default_path}'." - ) - return cached_dataset_dir - - -def migrate_cache(target_path: str) -> None: - """Migrate datasets from old_path to new_path. - - Parameters - ---------- - target_path: - The new path for TSDB to store cached datasets. - - """ - cached_dataset_dir = determine_data_home() - migrate(cached_dataset_dir, target_path) - config_parser = read_configs() - write_configs(config_parser, {"path": {"data_home": target_path}}) - logger.info(f"Have set {target_path} as the default cache dir.") - - def pickle_dump(data: object, path: str) -> Optional[str]: """Pickle the given object. @@ -184,3 +95,91 @@ def purge_path(path: str, ignore_errors: bool = True) -> None: ) except shutil.Error: raise shutil.Error("Operation failed.") + + +def determine_data_home(): + # read data_home from the config file + 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("~")) + old_cached_dataset_dir = os.path.join( + os.path.expanduser("~"), ".tsdb_cached_datasets" + ) + + if os.path.exists(old_cached_dataset_dir): + # use the old path and warn the user + logger.warning( + "‼️ Detected the home dir of the old version TSDB. " + "Since v0.3, TSDB has changed the default cache dir to '~/.tsdb'. " + "Auto migrating downloaded datasets to the new path. " + ) + cached_dataset_dir = data_home_path + migrate(old_cached_dataset_dir, cached_dataset_dir) + elif os.path.exists(data_home_path): + # use the path directly, may be in a portable disk + cached_dataset_dir = data_home_path + else: + # use the default path + default_path = os.path.join(os.path.expanduser("~"), ".tsdb") + cached_dataset_dir = default_path + if os.path.abspath(data_home_path) != os.path.abspath(default_path): + logger.warning( + f"‼️ The preset data_home path '{data_home_path}' doesn't exist. " + f"Using the default path '{default_path}'." + ) + return cached_dataset_dir + + +def migrate(old_path: str, new_path: str) -> None: + """Migrate files in a directory from old_path to new_path. + + Parameters + ---------- + old_path: + The old path of the dataset. + + new_path: + The new path of the dataset. + + """ + if not os.path.exists(old_path): + raise FileNotFoundError(f"Given old_path {old_path} does not exist.") + + 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) + + 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}" + ) + + +def migrate_cache(target_path: str) -> None: + """Migrate datasets from old_path to new_path. + + Parameters + ---------- + target_path: + The new path for TSDB to store cached datasets. + + """ + cached_dataset_dir = determine_data_home() + migrate(cached_dataset_dir, target_path) + config_parser = read_configs() + write_configs(config_parser, {"path": {"data_home": target_path}}) + logger.info(f"Have set {target_path} as the default cache dir.")