2272299416

Build Type

github

Committed by GitHub

Commit Message

Merge 648cff5be into dad775e9d

Pull Request Pull Request #33: Support CORDEX and CMIP5/6

Run Details

34 of 260 new or added lines in 16 files covered. (13.08%)

10 existing lines in 7 files now uncovered.

661 of 3276 relevant lines covered (20.18%)

0.61 hits per line

Source File
Press 'n' to go to next uncovered line, 'b' for previous

0.0

/miranda/remote/rsdtmf.py

"""
=================
rstdmf Operations
=================

Classes:

 * rstdmfError - the exception raised on failure.

Objects:

 * DiskSpaceEvent - threading event for when there is not enough space on disk.

Functions:

 * :func:`rstdmf_rename` - single rstdmf call with file renaming at the end.
 * :func:`local_storage_for_rstdmf` - storage object for rstdmf_divisions.
 * :func:`rstdmf_divisions` - rstdmf calls based on file numbers and size.

Notes
-----
The restored files are always renamed to their full path with '/' replaced
by '.' (leading '/' removed) to allow for restoring files with the same name
in different locations.
e.g. /server/path/sample.tar becomes server.path.sample.tar
The python trick to convert is::

   >>> path_string = Path(dmf1_file).absolute() # to get full path if necessary.
   >>> str(path_string).replace('/','.').lstrip('.')

"""
import logging.config
import os
import subprocess
import threading
import time
from pathlib import Path
from typing import List, Union

from miranda.scripting import LOGGING_CONFIG
from miranda.storage import (
    DiskSpaceError,
    FileMeta,
    StorageState,
    size_division,
    size_evaluation,
)
from miranda.units import GiB
from miranda.utils import find_filepaths, yesno_prompt

from .ops import transfer_file

DiskSpaceEvent = threading.Event()

logging.config.dictConfig(LOGGING_CONFIG)

__all__ = [
    "local_storage_for_rstdmf",
    "rstdmf_divisions",
    "rstdmf_rename",
    "RstdmfError",
]


def rstdmf_rename(file_list: Union[str, List[str]], restore_path: str):
    """Single rstdmf call with file renaming at the end.

    Parameters
    ----------
    file_list : List[str]
      List of files to restore.
    restore_path : str
      Path to restore files and folders to.

    Notes
    -----
    The restored files are renamed to their full path with '/' replaced by '.'.
    e.g. /dmf1/scenario/sample.tar becomes dmf1.scenario.sample.tar

    """
    # Make sure we have the full path of each argument
    if not isinstance(file_list, list):
        file_list = [file_list]
    file_list = [Path(x).absolute().as_posix() for x in file_list]
    restore_path = Path(restore_path).absolute()

    # Create output directory, if necessary
    try:
        Path(restore_path).mkdir(parents=True, exist_ok=True)
    except OSError:
        raise RstdmfError(f"Cannot create restore path: {restore_path}.")

    # Create string with list of files
    file_list_string = "' '".join(file_list)
    file_list_string = f"'{file_list_string}'"
    file_list_string = file_list_string.replace("(", r"\(")
    file_list_string = file_list_string.replace(")", r"\)")

    # rstdmf call
    flag_system = subprocess.call(
        ["rstdmf", "-m", "-b", file_list_string, restore_path]
    )
    if flag_system:
        raise RstdmfError("rstdmf call failed.")

    # rename files
    for file_to_restore in file_list:
        file_name = file_to_restore.replace("/", ".").lstrip(".")
        restored_file = Path(restore_path).joinpath(Path(file_to_restore).name)
        try:
            transfer_file(restored_file, Path(restore_path).joinpath(file_name))
        except OSError:
            raise RstdmfError(f"Could not move {restored_file}.")


def local_storage_for_rstdmf(
    files_for_rstdmf: List[str], restore_path: str, max_size_on_disk: int
) -> StorageState:
    """Create a local storage object for rstdmf_divisions function.

    Parameters
    ----------
    files_for_rstdmf : List[str]
      files that will be restored.
    restore_path : str
      Path to restore files and folders to.
    max_size_on_disk : int
      maximum amount of space to be used on disk.

    Returns
    -------
    StorageState
    """

    # Get files already restored on disk and their size
    files_on_disk = list()
    for file_on_disk in files_for_rstdmf:
        renamed_file = file_on_disk.replace("/", ".").lstrip(".")
        renamed_path = Path(restore_path).joinpath(renamed_file)
        if os.path.isfile(renamed_path):
            files_on_disk.append(renamed_path)
        elif os.path.isdir(renamed_path):
            files_within = find_filepaths(renamed_path)
            files_on_disk.append(
                [FileMeta(renamed_path.as_posix()), size_evaluation(files_within)]
            )
    size_on_disk = size_evaluation(files_on_disk)
    # create a local storage object
    return StorageState(
        restore_path, max_size_on_disk, size_on_disk, max_size_on_disk - size_on_disk
    )


#
def rstdmf_divisions(
    paths_to_restore,
    restore_path,
    disk_space_margin: int = 100 * GiB,
    disk_refresh_time: int = 60,
    rstdmf_size_limit: int = 10 * GiB,
    rstdmf_file_limit: int = 24,
    max_size_on_disk: int = 0,
    verbose: bool = False,
    preserve_order: bool = True,
):
    """Securely make call(s) to rstdmf.

    Parameters
    ----------
    paths_to_restore : list of strings
        list of files to restore (path or FileMeta objects), giving
        directories also works, but they are considered like a single file
        (i.e. be careful of their size!).
    restore_path : str
        Path where files will be restored.
    disk_space_margin : int
        minimum space required on disk to proceed, set to 0 for no minimum
        requirement (default: 100 GiB).
    disk_refresh_time : int
        time to wait to lookup disk space again when full (default: 60 s).
    rstdmf_size_limit : int
        size limit of a single rstdmf call, set to 0 for no size limit
        (default: 10 GiB).
    rstdmf_file_limit : int
        limit in number of files of a single rstdmf call, set to 0 for no
        limit (default: 24).
    max_size_on_disk : int
        maximum size of restored files on disk, set to 0 for no maximum size
        (default: 0).
    verbose : bool
        verbose mode (default: off).
    preserve_order : bool
        flag to force files to be restored in the order they are given
        (default: True).

    Notes
    -----
    The restored files are renamed to their full path with '/' replaced by '.':
    for example, ``/server/path/sample.tar`` becomes ``server.path.sample.tar``
    This allows getting files with the same name in different locations.
    If this renamed file is already in the target location, it is not restored.

    """

    # Make a list of files that should be restored
    files_for_rstdmf = []
    for file_to_restore in paths_to_restore:
        # If a directory is given, create its associated FileMeta object
        if os.path.isdir(file_to_restore):
            files_within = find_filepaths(file_to_restore)
            file_to_restore = FileMeta(file_to_restore, size_evaluation(files_within))
        # If just a file name is given, convert it to a FileMeta object
        elif not isinstance(file_to_restore, FileMeta):
            file_to_restore = FileMeta(file_to_restore)
        # Skip files that do not exist or are already in the restoration path
        file_name = file_to_restore.path.replace("/", ".").lstrip(".")
        if (
            Path(restore_path).joinpath(file_name).exists()
            or not Path(file_to_restore).exists()
        ):
            continue
        # Make sure no files are repeated
        for file_for_rstdmf in files_for_rstdmf:
            if file_for_rstdmf.path == file_to_restore.path:
                break
        else:
            files_for_rstdmf.append(file_to_restore)
    if not files_for_rstdmf:
        logging.warning("All files are already on disk or nonexistent.")
        return

    # Divide files to satisfy rstdmf constraints
    if (max_size_on_disk != 0) and (max_size_on_disk < rstdmf_size_limit):
        division_size_limit = max_size_on_disk
    else:
        division_size_limit = rstdmf_size_limit
    divisions = size_division(
        files_for_rstdmf,
        division_size_limit,
        rstdmf_file_limit,
        check_name_repetition=True,
        preserve_order=preserve_order,
    )

    try:
        storage = StorageState(restore_path)
    except DiskSpaceError:
        storage = StorageState(restore_path, 0, 0, 0)
    if verbose:
        size_of_files = size_evaluation(files_for_rstdmf)
        logging.warning(
            f"Disk space: {float(storage.free_space) / GiB} GiB, "
            f"Restoration size: {size_of_files / GiB} GiB, "
            f"Subdivisions: {str(len(divisions))}"
        )
        if not yesno_prompt("Proceed with restoration?"):
            return

    for division in divisions:
        size_of_files = size_evaluation(division)
        list_of_path = [file_meta.path for file_meta in division]
        # If we can't get disk space, disregard it in the process
        try:
            storage = StorageState(restore_path)
        except DiskSpaceError:
            storage = StorageState(restore_path, 0, 0, 0)
            disk_space_margin = 0
        local_storage = local_storage_for_rstdmf(
            files_for_rstdmf, restore_path, max_size_on_disk
        )
        if (
            (disk_space_margin != 0)
            and (storage.free_space - disk_space_margin < size_of_files)
        ) or ((max_size_on_disk != 0) and (local_storage.free_space < size_of_files)):
            DiskSpaceEvent.set()
            if storage.free_space - disk_space_margin < size_of_files:
                logging.warning(
                    f"Disk space running low! Starting wait loop ({disk_refresh_time} seconds)."
                )
                logging.warning(
                    f"Disk space: {float(storage.free_space) / GiB} GB, "
                    f"Disk space to maintain: {disk_space_margin / GiB} GiB, "
                    f"Restore size: {str(size_of_files / GiB)} GiB."
                )
            elif local_storage.free_space < size_of_files:
                logging.warning(
                    f"Reached size limit of files on disk ({float(max_size_on_disk) / GiB} GiB). "
                    f"Starting wait loop ({disk_refresh_time} seconds)."
                )
            while (storage.free_space - disk_space_margin < size_of_files) or (
                local_storage.free_space < size_of_files
            ):
                time.sleep(disk_refresh_time)
                # If we can't get disk space, disregard it in the process
                try:
                    storage = StorageState(restore_path)
                except DiskSpaceError:
                    storage = StorageState(restore_path, 0, 0, 0)
                    disk_space_margin = 0
                local_storage = local_storage_for_rstdmf(
                    files_for_rstdmf, restore_path, max_size_on_disk
                )
        DiskSpaceEvent.clear()
        try:
            rstdmf_rename(list_of_path, restore_path)
        except RstdmfError:
            raise


class RstdmfError(Exception):
    pass

1	"""
2	=================
3	rstdmf Operations
4	=================
5
6	Classes:
7
8	* rstdmfError - the exception raised on failure.
9
10	Objects:
11
12	* DiskSpaceEvent - threading event for when there is not enough space on disk.
13
14	Functions:
15
16	* :func:`rstdmf_rename` - single rstdmf call with file renaming at the end.
17	* :func:`local_storage_for_rstdmf` - storage object for rstdmf_divisions.
18	* :func:`rstdmf_divisions` - rstdmf calls based on file numbers and size.
19
20	Notes
21	-----
22	The restored files are always renamed to their full path with '/' replaced
23	by '.' (leading '/' removed) to allow for restoring files with the same name
24	in different locations.
25	e.g. /server/path/sample.tar becomes server.path.sample.tar
26	The python trick to convert is::
27
28	>>> path_string = Path(dmf1_file).absolute() # to get full path if necessary.
29	>>> str(path_string).replace('/','.').lstrip('.')
30
31	"""
32	import logging.config	×
33	import os	×
34	import subprocess	×
35	import threading	×
36	import time	×
37	from pathlib import Path	×
38	from typing import List, Union	×
39
40	from miranda.scripting import LOGGING_CONFIG	×
41	from miranda.storage import (	×
42	DiskSpaceError,
43	FileMeta,
44	StorageState,
45	size_division,
46	size_evaluation,
47	)
48	from miranda.units import GiB	×
49	from miranda.utils import find_filepaths, yesno_prompt	×
50
51	from .ops import transfer_file	×
52
53	DiskSpaceEvent = threading.Event()	×
54
55	logging.config.dictConfig(LOGGING_CONFIG)	×
56
57	__all__ = [	×
58	"local_storage_for_rstdmf",
59	"rstdmf_divisions",
60	"rstdmf_rename",
61	"RstdmfError",
62	]
63
64
65	def rstdmf_rename(file_list: Union[str, List[str]], restore_path: str):	×
66	"""Single rstdmf call with file renaming at the end.
67
68	Parameters
69	----------
70	file_list : List[str]
71	List of files to restore.
72	restore_path : str
73	Path to restore files and folders to.
74
75	Notes
76	-----
77	The restored files are renamed to their full path with '/' replaced by '.'.
78	e.g. /dmf1/scenario/sample.tar becomes dmf1.scenario.sample.tar
79
80	"""
81	# Make sure we have the full path of each argument
82	if not isinstance(file_list, list):	×
83	file_list = [file_list]	×
84	file_list = [Path(x).absolute().as_posix() for x in file_list]	×
85	restore_path = Path(restore_path).absolute()	×
86
87	# Create output directory, if necessary
88	try:	×
89	Path(restore_path).mkdir(parents=True, exist_ok=True)	×
90	except OSError:	×
NEW 91	raise RstdmfError(f"Cannot create restore path: {restore_path}.")	×
92
93	# Create string with list of files
94	file_list_string = "' '".join(file_list)	×
NEW 95	file_list_string = f"'{file_list_string}'"	×
96	file_list_string = file_list_string.replace("(", r"\(")	×
97	file_list_string = file_list_string.replace(")", r"\)")	×
98
99	# rstdmf call
100	flag_system = subprocess.call(	×
101	["rstdmf", "-m", "-b", file_list_string, restore_path]
102	)
103	if flag_system:	×
104	raise RstdmfError("rstdmf call failed.")	×
105
106	# rename files
107	for file_to_restore in file_list:	×
108	file_name = file_to_restore.replace("/", ".").lstrip(".")	×
109	restored_file = Path(restore_path).joinpath(Path(file_to_restore).name)	×
110	try:	×
111	transfer_file(restored_file, Path(restore_path).joinpath(file_name))	×
112	except OSError:	×
113	raise RstdmfError(f"Could not move {restored_file}.")	×
114
115
116	def local_storage_for_rstdmf(	×
117	files_for_rstdmf: List[str], restore_path: str, max_size_on_disk: int
118	) -> StorageState:
119	"""Create a local storage object for rstdmf_divisions function.
120
121	Parameters
122	----------
123	files_for_rstdmf : List[str]
124	files that will be restored.
125	restore_path : str
126	Path to restore files and folders to.
127	max_size_on_disk : int
128	maximum amount of space to be used on disk.
129
130	Returns
131	-------
132	StorageState
133	"""
134
135	# Get files already restored on disk and their size
136	files_on_disk = list()	×
137	for file_on_disk in files_for_rstdmf:	×
138	renamed_file = file_on_disk.replace("/", ".").lstrip(".")	×
139	renamed_path = Path(restore_path).joinpath(renamed_file)	×
140	if os.path.isfile(renamed_path):	×
141	files_on_disk.append(renamed_path)	×
142	elif os.path.isdir(renamed_path):	×
143	files_within = find_filepaths(renamed_path)	×
144	files_on_disk.append(	×
145	[FileMeta(renamed_path.as_posix()), size_evaluation(files_within)]
146	)
147	size_on_disk = size_evaluation(files_on_disk)	×
148	# create a local storage object
149	return StorageState(	×
150	restore_path, max_size_on_disk, size_on_disk, max_size_on_disk - size_on_disk
151	)
152
153
154	#
155	def rstdmf_divisions(	×
156	paths_to_restore,
157	restore_path,
158	disk_space_margin: int = 100 * GiB,
159	disk_refresh_time: int = 60,
160	rstdmf_size_limit: int = 10 * GiB,
161	rstdmf_file_limit: int = 24,
162	max_size_on_disk: int = 0,
163	verbose: bool = False,
164	preserve_order: bool = True,
165	):
166	"""Securely make call(s) to rstdmf.
167
168	Parameters
169	----------
170	paths_to_restore : list of strings
171	list of files to restore (path or FileMeta objects), giving
172	directories also works, but they are considered like a single file
173	(i.e. be careful of their size!).
174	restore_path : str
175	Path where files will be restored.
176	disk_space_margin : int
177	minimum space required on disk to proceed, set to 0 for no minimum
178	requirement (default: 100 GiB).
179	disk_refresh_time : int
180	time to wait to lookup disk space again when full (default: 60 s).
181	rstdmf_size_limit : int
182	size limit of a single rstdmf call, set to 0 for no size limit
183	(default: 10 GiB).
184	rstdmf_file_limit : int
185	limit in number of files of a single rstdmf call, set to 0 for no
186	limit (default: 24).
187	max_size_on_disk : int
188	maximum size of restored files on disk, set to 0 for no maximum size
189	(default: 0).
190	verbose : bool
191	verbose mode (default: off).
192	preserve_order : bool
193	flag to force files to be restored in the order they are given
194	(default: True).
195
196	Notes
197	-----
198	The restored files are renamed to their full path with '/' replaced by '.':
199	for example, ``/server/path/sample.tar`` becomes ``server.path.sample.tar``
200	This allows getting files with the same name in different locations.
201	If this renamed file is already in the target location, it is not restored.
202
203	"""
204
205	# Make a list of files that should be restored
206	files_for_rstdmf = []	×
207	for file_to_restore in paths_to_restore:	×
208	# If a directory is given, create its associated FileMeta object
209	if os.path.isdir(file_to_restore):	×
210	files_within = find_filepaths(file_to_restore)	×
211	file_to_restore = FileMeta(file_to_restore, size_evaluation(files_within))	×
212	# If just a file name is given, convert it to a FileMeta object
213	elif not isinstance(file_to_restore, FileMeta):	×
214	file_to_restore = FileMeta(file_to_restore)	×
215	# Skip files that do not exist or are already in the restoration path
216	file_name = file_to_restore.path.replace("/", ".").lstrip(".")	×
217	if (	×
218	Path(restore_path).joinpath(file_name).exists()
219	or not Path(file_to_restore).exists()
220	):
221	continue	×
222	# Make sure no files are repeated
223	for file_for_rstdmf in files_for_rstdmf:	×
224	if file_for_rstdmf.path == file_to_restore.path:	×
225	break	×
226	else:
227	files_for_rstdmf.append(file_to_restore)	×
228	if not files_for_rstdmf:	×
229	logging.warning("All files are already on disk or nonexistent.")	×
230	return	×
231
232	# Divide files to satisfy rstdmf constraints
233	if (max_size_on_disk != 0) and (max_size_on_disk < rstdmf_size_limit):	×
234	division_size_limit = max_size_on_disk	×
235	else:
236	division_size_limit = rstdmf_size_limit	×
237	divisions = size_division(	×
238	files_for_rstdmf,
239	division_size_limit,
240	rstdmf_file_limit,
241	check_name_repetition=True,
242	preserve_order=preserve_order,
243	)
244
245	try:	×
246	storage = StorageState(restore_path)	×
247	except DiskSpaceError:	×
248	storage = StorageState(restore_path, 0, 0, 0)	×
249	if verbose:	×
250	size_of_files = size_evaluation(files_for_rstdmf)	×
251	logging.warning(	×
252	f"Disk space: {float(storage.free_space) / GiB} GiB, "
253	f"Restoration size: {size_of_files / GiB} GiB, "
254	f"Subdivisions: {str(len(divisions))}"
255	)
256	if not yesno_prompt("Proceed with restoration?"):	×
257	return	×
258
259	for division in divisions:	×
260	size_of_files = size_evaluation(division)	×
261	list_of_path = [file_meta.path for file_meta in division]	×
262	# If we can't get disk space, disregard it in the process
263	try:	×
264	storage = StorageState(restore_path)	×
265	except DiskSpaceError:	×
266	storage = StorageState(restore_path, 0, 0, 0)	×
267	disk_space_margin = 0	×
268	local_storage = local_storage_for_rstdmf(	×
269	files_for_rstdmf, restore_path, max_size_on_disk
270	)
271	if (	×
272	(disk_space_margin != 0)
273	and (storage.free_space - disk_space_margin < size_of_files)
274	) or ((max_size_on_disk != 0) and (local_storage.free_space < size_of_files)):
275	DiskSpaceEvent.set()	×
276	if storage.free_space - disk_space_margin < size_of_files:	×
277	logging.warning(	×
278	f"Disk space running low! Starting wait loop ({disk_refresh_time} seconds)."
279	)
280	logging.warning(	×
281	f"Disk space: {float(storage.free_space) / GiB} GB, "
282	f"Disk space to maintain: {disk_space_margin / GiB} GiB, "
283	f"Restore size: {str(size_of_files / GiB)} GiB."
284	)
285	elif local_storage.free_space < size_of_files:	×
286	logging.warning(	×
287	f"Reached size limit of files on disk ({float(max_size_on_disk) / GiB} GiB). "
288	f"Starting wait loop ({disk_refresh_time} seconds)."
289	)
290	while (storage.free_space - disk_space_margin < size_of_files) or (	×
291	local_storage.free_space < size_of_files
292	):
293	time.sleep(disk_refresh_time)	×
294	# If we can't get disk space, disregard it in the process
295	try:	×
296	storage = StorageState(restore_path)	×
297	except DiskSpaceError:	×
298	storage = StorageState(restore_path, 0, 0, 0)	×
299	disk_space_margin = 0	×
300	local_storage = local_storage_for_rstdmf(	×
301	files_for_rstdmf, restore_path, max_size_on_disk
302	)
303	DiskSpaceEvent.clear()	×
304	try:	×
305	rstdmf_rename(list_of_path, restore_path)	×
306	except RstdmfError:	×
307	raise	×
308
309
310	class RstdmfError(Exception):	×
311	pass	×

Ouranosinc / miranda / 2272299416

Source File Press 'n' to go to next uncovered line, 'b' for previous

Source File
Press 'n' to go to next uncovered line, 'b' for previous