4378726039

Build Type

github

Committed by GitHub

Commit Message

<a href="https://github.com/Ouranosinc/miranda/commit/<a class=hub.com/Ouranosinc/miranda/commit/84127db3b4a449e2bd6acbd8005a5dadda458da9">84127db3b<a href="https://github.com/Ouranosinc/miranda/commit/84127db3b4a449e2bd6acbd8005a5dadda458da9">">Merge </a><a class="double-link" href="https://github.com/Ouranosinc/miranda/commit/<a class="double-link" href="https://github.com/Ouranosinc/miranda/commit/28e8657b02e6d2b4b08a3201104ddccaf4b6e3fa">28e8657b0</a>">28e8657b0</a><a href="https://github.com/Ouranosinc/miranda/commit/84127db3b4a449e2bd6acbd8005a5dadda458da9"> into d176753fe">d176753fe</a>

Pull Request Pull Request #101: Support ETS-GRNCH dataset

Run Details

37 of 526 new or added lines in 11 files covered. (7.03%)

901 of 4548 relevant lines covered (19.81%)

0.59 hits per line

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

16.91

/miranda/storage.py

"""
=====================
Disk space management
=====================

Classes:

 * DiskSpaceError - the exception raised on failure.
 * :class:`FileMeta` - file and its size.
 * :class:`StorageState` - storage capacity and availability of a medium.

Functions:

 * :func:`total_size` - get total size of a list of files.
 * :func:`size_division` - divide files based on number and size restrictions.

"""
import logging
import logging.config
import os
import subprocess
from functools import reduce
from pathlib import Path
from types import GeneratorType
from typing import Dict, List, Union

from .scripting import LOGGING_CONFIG

logging.config.dictConfig(LOGGING_CONFIG)


__all__ = [
    "DiskSpaceError",
    "FileMeta",
    "StorageState",
    "file_size",
    "report_file_size",
    "size_division",
    "size_evaluation",
]


class DiskSpaceError(Exception):
    pass


class FileMeta:
    """File path and size."""

    django = {
        "path": ["CharField", "max_length=512"],
        "size": ["IntegerField", "null=True", "blank=True"],
    }

    def __init__(self, path: str, size: int = -1):
        """Initialize file meta.

        Parameters
        ----------
        path : str
            full path of the file.
        size : int
            size of file in bytes (default: will obtain from os.path.getsize
            if file exists, set to 0 otherwise).

        """

        # Make sure we have the full path of the file
        self.path = Path(path).absolute()

        # Get size of file if it is not specified
        if (-1 == size) and self.path.exists():
            try:
                self.size = self.path.stat().st_size
            except OSError:
                raise DiskSpaceError(f"Cannot get size of {self.path.name}.")
        elif -1 == size:
            self.size = 0
        else:
            self.size = size

    def __eq__(self, other):
        if self.path == other.path:
            return True
        else:
            return False


class StorageState:
    """Information regarding the storage capacity of a disk."""

    def __init__(self, base_path, capacity=-1, used_space=-1, free_space=-1):
        """Initialize storage state.

        Parameters
        ----------
        base_path : str
            base path of the storage medium.
        capacity : int
            capacity of medium in bytes (default: will obtain from system
            call to 'df').
        used_space : int
            space currently used on the medium (default: will obtain from
            system call to 'df').
        free_space : int
            space available on the medium (default: will obtain from system
            call to 'df').

        """

        # Make sure we have the full base path
        self.base_path = Path(base_path).absolute()

        # Get attributes from 'df' function if they are not specified
        if not self.base_path.is_dir():
            raise DiskSpaceError(f"Cannot analyze {self.base_path}.")
        if not Path("/bin/df").exists():
            raise DiskSpaceError("/bin/df does not exist.")
        df_output = subprocess.run(
            ["/bin/df", "-P", base_path, "|", "tail", "-1"], capture_output=True
        )
        if not df_output.stdout:
            raise DiskSpaceError("/bin/df call failed.")
        df_output_split = df_output.stdout.split()
        if -1 == capacity:
            try:
                self.capacity = int(df_output_split[1]) * 1000
            except Exception as e:
                raise DiskSpaceError("/bin/df output not as expected.") from e
        else:
            self.capacity = capacity
        if -1 == used_space:
            try:
                self.used_space = int(df_output_split[2]) * 1000
            except Exception as e:
                raise DiskSpaceError("/bin/df output not as expected.") from e
        else:
            self.used_space = used_space
        if -1 == free_space:
            try:
                self.free_space = int(df_output_split[3]) * 1000
            except Exception as e:
                raise DiskSpaceError("/bin/df output not as expected.") from e
        else:
            self.free_space = free_space


def size_evaluation(file_list: List[Union[str, FileMeta, Path]]) -> int:
    """Total size of files.

    Parameters
    ----------
    file_list : Union[str, Path, FileMeta]

    Returns
    -------
    int
      total size of files in bytes.

    """

    if file_list:
        size = 0
        for file_to_add in file_list:
            # If file paths are given, convert to FileMeta objects first
            if not isinstance(file_to_add, FileMeta):
                try:
                    file_to_add = FileMeta(file_to_add)
                except DiskSpaceError:
                    raise
            size += file_to_add.size
        return size
    else:
        return 0


def size_division(
    files_to_divide: Union[List, FileMeta, Path],
    size_limit: int = 0,
    file_limit: int = 0,
    check_name_repetition: bool = False,
    preserve_order: bool = False,
) -> List[list]:
    """Divide files according to size and number limits.

    Parameters
    ----------
    files_to_divide : Union[List, FileMeta, Path]
    size_limit : int
      Size limit of divisions in bytes. Default: 0 (no limit).
    file_limit : int
      Number of files limit of divisions. Default: 0 (no limit).
    check_name_repetition : bool
      Flag to prevent file name repetitions. Default: False.
    preserve_order : bool
      Flag to force files to be restored in the order they are given. Default: False.

    Returns
    -------
    List[list]
      list of divisions (each division is a list of FileMeta objects).

    """

    divisions = list()
    for file_divide in files_to_divide:
        # If file paths are given, convert to FileMeta objects first
        if not isinstance(file_divide, FileMeta):
            try:
                file_divide = FileMeta(file_divide)
            except DiskSpaceError:
                raise
        # Loop through divisions and try to add file according to limitations
        for i, division in enumerate(divisions):
            size = file_divide.size
            file_count = 1
            flag_skip = 0
            for file_divided in division:
                if check_name_repetition and (
                    os.path.basename(file_divided.path)
                    == os.path.basename(file_divide.path)
                ):
                    flag_skip = 1
                size = size + file_divided.size
                file_count = file_count + 1
            if (
                (size > size_limit != 0)
                or (file_count > file_limit != 0)
                or flag_skip == 1
            ):
                continue
            elif preserve_order and (i != len(divisions) - 1):
                continue
            else:
                divisions[i].append(file_divide)
                break
        else:
            divisions.append([file_divide])
    return divisions


def file_size(
    file_path_or_bytes_or_dict: Union[
        Path,
        str,
        int,
        List[Union[str, Path]],
        GeneratorType,
        Dict[str, Union[Path, List[Path]]],
    ]
) -> int:
    """

    Parameters
    ----------
    file_path_or_bytes_or_dict : Union[Path, str, int, List[Union[str, Path]], GeneratorType, Dict[str, Union[Path, List[Path]]]]

    Returns
    -------
    int
    """
    try:
        if isinstance(file_path_or_bytes_or_dict, int):
            total = file_path_or_bytes_or_dict
        elif isinstance(file_path_or_bytes_or_dict, (list, GeneratorType)):
            try:
                total = reduce(
                    (lambda x, y: x + y),
                    map(lambda f: Path(f).stat().st_size, file_path_or_bytes_or_dict),
                )
            except TypeError:
                total = 0
        elif isinstance(file_path_or_bytes_or_dict, dict):
            total: int = 0
            for key, val in file_path_or_bytes_or_dict.items():
                if isinstance(val, list):
                    try:
                        total += reduce(
                            (lambda x, y: x + y),
                            map(lambda f: Path(f).stat().st_size, val),
                        )
                    except TypeError:
                        continue
                elif Path(val).is_file():
                    total += Path(val).stat().st_size
        elif Path(file_path_or_bytes_or_dict).is_file():
            total = Path(file_path_or_bytes_or_dict).stat().st_size
        elif Path(file_path_or_bytes_or_dict).is_dir():
            total = reduce(
                (lambda x, y: x + y),
                [f.stat().st_size for f in Path(file_path_or_bytes_or_dict).rglob("*")],
            )
        else:
            raise FileNotFoundError
    except FileNotFoundError:
        logging.error(
            f"File Not Found: Unable to parse file size from {file_path_or_bytes_or_dict}"
        )
        raise

    return total


def report_file_size(
    file_path_or_bytes_or_dict: Union[
        Path, str, int, List[Union[str, Path]], Dict[str, Union[Path, List[Path]]]
    ],
    use_binary: bool = True,
    significant_digits: int = 2,
) -> str:
    """
    This function will parse the contents of a list or generator of files and return the
    size in bytes of a file or a list of files in pretty formatted text.
    """
    _CONVERSIONS = ["B", "k{}B", "M{}B", "G{}B", "T{}B", "P{}B", "E{}B", "Z{}B", "Y{}B"]

    def _size_formatter(i: int, binary: bool = True, precision: int = 2) -> str:
        """
        This function will format byte size into an appropriate nomenclature for prettier printing.
        """
        import math

        base = 1024 if binary else 1000
        if i == 0:
            return "0 B"
        multiple = math.trunc(math.log2(i) / math.log2(base))
        value = i / math.pow(base, multiple)
        suffix = _CONVERSIONS[multiple].format("i" if binary else "")
        return f"{value:.{precision}f} {suffix}"

    total = file_size(file_path_or_bytes_or_dict)
    return _size_formatter(total, binary=use_binary, precision=significant_digits)

1	"""
2	=====================
3	Disk space management
4	=====================
5
6	Classes:
7
8	* DiskSpaceError - the exception raised on failure.
9	* :class:`FileMeta` - file and its size.
10	* :class:`StorageState` - storage capacity and availability of a medium.
11
12	Functions:
13
14	* :func:`total_size` - get total size of a list of files.
15	* :func:`size_division` - divide files based on number and size restrictions.
16
17	"""
18	import logging
19	import logging.config
20	import os
21	import subprocess
22	from functools import reduce
23	from pathlib import Path
24	from types import GeneratorType
25	from typing import Dict, List, Union
26
27	from .scripting import LOGGING_CONFIG
28
29	logging.config.dictConfig(LOGGING_CONFIG)
30
31
32	__all__ = [
33	"DiskSpaceError",
34	"FileMeta",
35	"StorageState",
36	"file_size",
37	"report_file_size",
38	"size_division",
39	"size_evaluation",
40	]
41
42
43	class DiskSpaceError(Exception):
44	pass
45
46
47	class FileMeta:
48	"""File path and size."""
49
50	django = {
51	"path": ["CharField", "max_length=512"],
52	"size": ["IntegerField", "null=True", "blank=True"],
53	}
54
55	def __init__(self, path: str, size: int = -1):
56	"""Initialize file meta.
57
58	Parameters
59	----------
60	path : str
61	full path of the file.
62	size : int
63	size of file in bytes (default: will obtain from os.path.getsize
64	if file exists, set to 0 otherwise).
65
66	"""
67
68	# Make sure we have the full path of the file
69	self.path = Path(path).absolute()
70
71	# Get size of file if it is not specified
72	if (-1 == size) and self.path.exists():
73	try:
74	self.size = self.path.stat().st_size
75	except OSError:
76	raise DiskSpaceError(f"Cannot get size of {self.path.name}.")
77	elif -1 == size:
78	self.size = 0
79	else:
80	self.size = size
81
82	def __eq__(self, other):
83	if self.path == other.path:
84	return True
85	else:
86	return False
87
88
89	class StorageState:
90	"""Information regarding the storage capacity of a disk."""
91
92	def __init__(self, base_path, capacity=-1, used_space=-1, free_space=-1):
93	"""Initialize storage state.
94
95	Parameters
96	----------
97	base_path : str
98	base path of the storage medium.
99	capacity : int
100	capacity of medium in bytes (default: will obtain from system
101	call to 'df').
102	used_space : int
103	space currently used on the medium (default: will obtain from
104	system call to 'df').
105	free_space : int
106	space available on the medium (default: will obtain from system
107	call to 'df').
108
109	"""
110
111	# Make sure we have the full base path
112	self.base_path = Path(base_path).absolute()
113
114	# Get attributes from 'df' function if they are not specified
115	if not self.base_path.is_dir():
116	raise DiskSpaceError(f"Cannot analyze {self.base_path}.")
117	if not Path("/bin/df").exists():
118	raise DiskSpaceError("/bin/df does not exist.")
119	df_output = subprocess.run(
120	["/bin/df", "-P", base_path, "\|", "tail", "-1"], capture_output=True
121	)
122	if not df_output.stdout:
123	raise DiskSpaceError("/bin/df call failed.")
124	df_output_split = df_output.stdout.split()
125	if -1 == capacity:
126	try:
127	self.capacity = int(df_output_split[1]) * 1000
128	except Exception as e:
129	raise DiskSpaceError("/bin/df output not as expected.") from e
130	else:
131	self.capacity = capacity
132	if -1 == used_space:
133	try:
134	self.used_space = int(df_output_split[2]) * 1000
135	except Exception as e:
136	raise DiskSpaceError("/bin/df output not as expected.") from e
137	else:
138	self.used_space = used_space
139	if -1 == free_space:
140	try:
141	self.free_space = int(df_output_split[3]) * 1000
142	except Exception as e:
143	raise DiskSpaceError("/bin/df output not as expected.") from e
144	else:
145	self.free_space = free_space
146
147
148	def size_evaluation(file_list: List[Union[str, FileMeta, Path]]) -> int:
149	"""Total size of files.
150
151	Parameters
152	----------
153	file_list : Union[str, Path, FileMeta]
154
155	Returns
156	-------
157	int
158	total size of files in bytes.
159
160	"""
161
162	if file_list:
163	size = 0
164	for file_to_add in file_list:
165	# If file paths are given, convert to FileMeta objects first
166	if not isinstance(file_to_add, FileMeta):
167	try:
168	file_to_add = FileMeta(file_to_add)
169	except DiskSpaceError:
170	raise
171	size += file_to_add.size
172	return size
173	else:
174	return 0
175
176
177	def size_division(
178	files_to_divide: Union[List, FileMeta, Path],
179	size_limit: int = 0,
180	file_limit: int = 0,
181	check_name_repetition: bool = False,
182	preserve_order: bool = False,
183	) -> List[list]:
184	"""Divide files according to size and number limits.
185
186	Parameters
187	----------
188	files_to_divide : Union[List, FileMeta, Path]
189	size_limit : int
190	Size limit of divisions in bytes. Default: 0 (no limit).
191	file_limit : int
192	Number of files limit of divisions. Default: 0 (no limit).
193	check_name_repetition : bool
194	Flag to prevent file name repetitions. Default: False.
195	preserve_order : bool
196	Flag to force files to be restored in the order they are given. Default: False.
197
198	Returns
199	-------
200	List[list]
201	list of divisions (each division is a list of FileMeta objects).
202
203	"""
204
205	divisions = list()
206	for file_divide in files_to_divide:
207	# If file paths are given, convert to FileMeta objects first
208	if not isinstance(file_divide, FileMeta):
209	try:
210	file_divide = FileMeta(file_divide)
211	except DiskSpaceError:
212	raise
213	# Loop through divisions and try to add file according to limitations
214	for i, division in enumerate(divisions):
215	size = file_divide.size
216	file_count = 1
217	flag_skip = 0
218	for file_divided in division:
219	if check_name_repetition and (
220	os.path.basename(file_divided.path)
221	== os.path.basename(file_divide.path)
222	):
223	flag_skip = 1
224	size = size + file_divided.size
225	file_count = file_count + 1
226	if (
227	(size > size_limit != 0)
228	or (file_count > file_limit != 0)
229	or flag_skip == 1
230	):
231	continue
232	elif preserve_order and (i != len(divisions) - 1):
233	continue
234	else:
235	divisions[i].append(file_divide)
236	break
237	else:
238	divisions.append([file_divide])
239	return divisions
240
241
242	def file_size(
243	file_path_or_bytes_or_dict: Union[
244	Path,
245	str,
246	int,
247	List[Union[str, Path]],
248	GeneratorType,
249	Dict[str, Union[Path, List[Path]]],
250	]
251	) -> int:
252	"""
253
254	Parameters
255	----------
256	file_path_or_bytes_or_dict : Union[Path, str, int, List[Union[str, Path]], GeneratorType, Dict[str, Union[Path, List[Path]]]]
257
258	Returns
259	-------
260	int
261	"""
262	try:
NEW 263	if isinstance(file_path_or_bytes_or_dict, int):
NEW 264	total = file_path_or_bytes_or_dict
NEW 265	elif isinstance(file_path_or_bytes_or_dict, (list, GeneratorType)):
266	try:
267	total = reduce(
268	(lambda x, y: x + y),
269	map(lambda f: Path(f).stat().st_size, file_path_or_bytes_or_dict),
270	)
271	except TypeError:
272	total = 0
NEW 273	elif isinstance(file_path_or_bytes_or_dict, dict):
NEW 274	total: int = 0
NEW 275	for key, val in file_path_or_bytes_or_dict.items():
NEW 276	if isinstance(val, list):
NEW 277	try:
NEW 278	total += reduce(
279	(lambda x, y: x + y),
280	map(lambda f: Path(f).stat().st_size, val),
281	)
NEW 282	except TypeError:
NEW 283	continue
NEW 284	elif Path(val).is_file():
NEW 285	total += Path(val).stat().st_size
NEW 286	elif Path(file_path_or_bytes_or_dict).is_file():
NEW 287	total = Path(file_path_or_bytes_or_dict).stat().st_size
NEW 288	elif Path(file_path_or_bytes_or_dict).is_dir():
289	total = reduce(
290	(lambda x, y: x + y),
291	[f.stat().st_size for f in Path(file_path_or_bytes_or_dict).rglob("*")],
292	)
293	else:
294	raise FileNotFoundError
295	except FileNotFoundError:
296	logging.error(
297	f"File Not Found: Unable to parse file size from {file_path_or_bytes_or_dict}"
298	)
299	raise
300
301	return total
302
303
304	def report_file_size(
305	file_path_or_bytes_or_dict: Union[
306	Path, str, int, List[Union[str, Path]], Dict[str, Union[Path, List[Path]]]
307	],
308	use_binary: bool = True,
309	significant_digits: int = 2,
310	) -> str:
311	"""
312	This function will parse the contents of a list or generator of files and return the
313	size in bytes of a file or a list of files in pretty formatted text.
314	"""
315	_CONVERSIONS = ["B", "k{}B", "M{}B", "G{}B", "T{}B", "P{}B", "E{}B", "Z{}B", "Y{}B"]
316
317	def _size_formatter(i: int, binary: bool = True, precision: int = 2) -> str:
318	"""
319	This function will format byte size into an appropriate nomenclature for prettier printing.
320	"""
321	import math
322
323	base = 1024 if binary else 1000
324	if i == 0:
325	return "0 B"
326	multiple = math.trunc(math.log2(i) / math.log2(base))
327	value = i / math.pow(base, multiple)
328	suffix = _CONVERSIONS[multiple].format("i" if binary else "")
329	return f"{value:.{precision}f} {suffix}"
330
NEW 331	total = file_size(file_path_or_bytes_or_dict)
332	return _size_formatter(total, binary=use_binary, precision=significant_digits)

Ouranosinc / miranda / 4378726039

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