1883563084

Build Type

github

Committed by GitHub

Commit Message

Merge 4555deb62 into 4a2a34e15

Pull Request Pull Request #24: Add CMIP file structure - WIP

Run Details

41 of 361 new or added lines in 19 files covered. (11.36%)

3 existing lines in 2 files now uncovered.

627 of 2829 relevant lines covered (22.16%)

0.66 hits per line

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

71.05

/miranda/data.py

"""
Copyright 2019 Trevor James Smith

Licensed under the Apache License, Version 2.0 (the "License");
you may not use this file except in compliance with the License.
You may obtain a copy of the License at

   http://www.apache.org/licenses/LICENSE-2.0

Unless required by applicable law or agreed to in writing, software
distributed under the License is distributed on an "AS IS" BASIS,
WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
See the License for the specific language governing permissions and
limitations under the License.
"""
import logging.config
import os
from pathlib import Path
from types import GeneratorType
from typing import List, Optional, Tuple, Union

from .ops import url_validate
from .scripting import LOGGING_CONFIG
from .utils import GiB, find_filepaths

logging.config.dictConfig(LOGGING_CONFIG)

__all__ = ["DataBase"]


class DataBase:
    """ """

    def __init__(
        self,
        source,
        *,
        destination: Optional[Union[Path, str]] = None,
        common_path: Optional[Union[Path, str]] = None,
        file_pattern: Union[str, List[str]] = "*.nc",
        project_name: str = None,
        recursive: bool = True,
    ):
        self._source = Path(source)

        if destination is not None:
            self._destination = Path(destination)
        else:
            self._destination = Path().cwd()

        self.project_name = str(project_name)
        if not self.project_name:
            self.project_name = self._destination.stem

        if not file_pattern:
            self.file_suffixes = ["*"]

        elif isinstance(file_pattern, str):
            self.file_suffixes = [file_pattern]
        elif isinstance(file_pattern, (GeneratorType, List)):
            self.file_suffixes = file_pattern

        if not recursive:
            self.recursive = False
        else:
            self.recursive = True

        # if common_path is None:
        #     self._common_path = Path(source)

        self._files = self._scrape(source)
        self._is_server = False

        self.successful_transfers = int(0)

    def __repr__(self):
        return "<{}.{} object at {}>".format(
            self.__class__.__module__, self.__class__.__name__, hex(id(self))
        )

    def __str__(self):
        prepr = "[%s]" % ", ".join([f'{k}: "{v}"' for k, v in self.__dict__.items()])
        return f"{self.__class__.__name__}({prepr})"

    def __getitem__(self, key):
        return self.__dict__[key]

    def __setitem__(self, key, value):
        self.__dict__[key] = value

    def __delitem__(self, key):
        del self.__dict__[key]

    def __contains__(self, key):
        return key in self.__dict__

    def __len__(self):
        return len(self._files)

    def _scrape(self, source) -> List[Path]:
        if source is None:
            raise ValueError("No source provided.")
        if isinstance(source, (GeneratorType, List, Tuple, str, Path)):
            files = find_filepaths(source, **self._as_dict())
            common_path = os.path.commonpath(files)
            self._files = files
            self._common_path = common_path
            return files
        raise ValueError("Source must be an iterable of strings or Paths.")

    def _as_dict(self):
        return {
            key: value
            for key, value in self.__dict__.items()
            if not key.startswith("_") and not callable(key)
        }

    def items(self):
        return self._as_dict().items()

    def keys(self):
        return self._as_dict().keys()

    def values(self):
        return self._as_dict().values()

    def group_by(
        self,
        *,
        common_path: Union[Path, str] = None,
        subdirectories: bool = True,
        dates: bool = True,
        size: int = 10 * GiB,
    ):
        # use_grouping = True
        #
        # if subdirectories:
        #     file_groups = group_by_subdirectories(self._files, within=common_path)
        #
        # else:
        #     file_groups = defaultdict(lambda: list())
        #     for f in self._files:
        #         file_groups["."].append(f)
        pass

    def target(self, target: Union[Path, str]):
        self._destination = target
        self._is_server = self._url_validate(target=target)

    @staticmethod
    def _url_validate(target):
        return url_validate(target=target)

    def archive(self):
        pass

    def transfer(self):
        pass

1	"""
2	Copyright 2019 Trevor James Smith
3
4	Licensed under the Apache License, Version 2.0 (the "License");
5	you may not use this file except in compliance with the License.
6	You may obtain a copy of the License at
7
8	http://www.apache.org/licenses/LICENSE-2.0
9
10	Unless required by applicable law or agreed to in writing, software
11	distributed under the License is distributed on an "AS IS" BASIS,
12	WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
13	See the License for the specific language governing permissions and
14	limitations under the License.
15	"""
16	import logging.config	3✔
17	import os	3✔
18	from pathlib import Path	3✔
19	from types import GeneratorType	3✔
20	from typing import List, Optional, Tuple, Union	3✔
21
22	from .ops import url_validate	3✔
23	from .scripting import LOGGING_CONFIG	3✔
24	from .utils import GiB, find_filepaths	3✔
25
26	logging.config.dictConfig(LOGGING_CONFIG)	3✔
27
28	__all__ = ["DataBase"]	3✔
29
30
31	class DataBase:	3✔
32	""" """
33
34	def __init__(	3✔
35	self,
36	source,
37	*,
38	destination: Optional[Union[Path, str]] = None,
39	common_path: Optional[Union[Path, str]] = None,
40	file_pattern: Union[str, List[str]] = "*.nc",
41	project_name: str = None,
42	recursive: bool = True,
43	):
44	self._source = Path(source)	3✔
45
46	if destination is not None:	3✔
47	self._destination = Path(destination)	×
48	else:
49	self._destination = Path().cwd()	3✔
50
51	self.project_name = str(project_name)	3✔
52	if not self.project_name:	3✔
53	self.project_name = self._destination.stem	×
54
55	if not file_pattern:	3✔
56	self.file_suffixes = ["*"]	×
57
58	elif isinstance(file_pattern, str):	3✔
59	self.file_suffixes = [file_pattern]	3✔
60	elif isinstance(file_pattern, (GeneratorType, List)):	×
61	self.file_suffixes = file_pattern	×
62
63	if not recursive:	3✔
64	self.recursive = False	×
65	else:
66	self.recursive = True	3✔
67
68	# if common_path is None:
69	# self._common_path = Path(source)
70
71	self._files = self._scrape(source)	3✔
72	self._is_server = False	3✔
73
74	self.successful_transfers = int(0)	3✔
75
76	def __repr__(self):	3✔
77	return "<{}.{} object at {}>".format(	×
78	self.__class__.__module__, self.__class__.__name__, hex(id(self))
79	)
80
81	def __str__(self):	3✔
NEW 82	prepr = "[%s]" % ", ".join([f'{k}: "{v}"' for k, v in self.__dict__.items()])	×
NEW 83	return f"{self.__class__.__name__}({prepr})"	×
84
85	def __getitem__(self, key):	3✔
86	return self.__dict__[key]	×
87
88	def __setitem__(self, key, value):	3✔
89	self.__dict__[key] = value	×
90
91	def __delitem__(self, key):	3✔
92	del self.__dict__[key]	×
93
94	def __contains__(self, key):	3✔
95	return key in self.__dict__	×
96
97	def __len__(self):	3✔
98	return len(self._files)	3✔
99
100	def _scrape(self, source) -> List[Path]:	3✔
101	if source is None:	3✔
102	raise ValueError("No source provided.")	×
103	if isinstance(source, (GeneratorType, List, Tuple, str, Path)):	3✔
104	files = find_filepaths(source, **self._as_dict())	3✔
105	common_path = os.path.commonpath(files)	3✔
106	self._files = files	3✔
107	self._common_path = common_path	3✔
108	return files	3✔
109	raise ValueError("Source must be an iterable of strings or Paths.")	×
110
111	def _as_dict(self):	3✔
112	return {	3✔
113	key: value
114	for key, value in self.__dict__.items()
115	if not key.startswith("_") and not callable(key)
116	}
117
118	def items(self):	3✔
119	return self._as_dict().items()	×
120
121	def keys(self):	3✔
122	return self._as_dict().keys()	3✔
123
124	def values(self):	3✔
125	return self._as_dict().values()	×
126
127	def group_by(	3✔
128	self,
129	*,
130	common_path: Union[Path, str] = None,
131	subdirectories: bool = True,
132	dates: bool = True,
133	size: int = 10 * GiB,
134	):
135	# use_grouping = True
136	#
137	# if subdirectories:
138	# file_groups = group_by_subdirectories(self._files, within=common_path)
139	#
140	# else:
141	# file_groups = defaultdict(lambda: list())
142	# for f in self._files:
143	# file_groups["."].append(f)
144	pass	×
145
146	def target(self, target: Union[Path, str]):	3✔
147	self._destination = target	×
148	self._is_server = self._url_validate(target=target)	×
149
150	@staticmethod	3✔
151	def _url_validate(target):	2✔
152	return url_validate(target=target)	3✔
153
154	def archive(self):	3✔
155	pass	×
156
157	def transfer(self):	3✔
158	pass	×

Ouranosinc / miranda / 1883563084

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