3578611372

Build Type

github

Committed by GitHub

Commit Message

Merge d758304a2 into 7b00b1797

Pull Request Pull Request #54: Custom meta

Run Details

529 of 640 branches covered (82.66%)

Branch coverage included in aggregate %.

59 of 64 new or added lines in 4 files covered. (92.19%)

31 existing lines in 3 files now uncovered.

1291 of 1421 relevant lines covered (90.85%)

3.63 hits per line

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

91.67

/src/ismn/filecollection.py

# The MIT License (MIT)
#
# Copyright (c) 2021 TU Wien
#
# Permission is hereby granted, free of charge, to any person obtaining a copy
# of this software and associated documentation files (the "Software"), to deal
# in the Software without restriction, including without limitation the rights
# to use, copy, modify, merge, publish, distribute, sublicense, and/or sell
# copies of the Software, and to permit persons to whom the Software is
# furnished to do so, subject to the following conditions:
#
# The above copyright notice and this permission notice shall be included in all
# copies or substantial portions of the Software.
#
# THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
# IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
# FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE
# AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
# LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,
# OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE
# SOFTWARE.

import logging

from tempfile import gettempdir
from pathlib import Path, PurePosixPath

import numpy as np
from tqdm import tqdm
from typing import Union
from multiprocessing import Pool, cpu_count
from operator import itemgetter
import time

from ismn.base import IsmnRoot
from ismn.const import *
from ismn.filehandlers import DataFile, StaticMetaFile
from ismn.meta import MetaData, MetaVar, Depth


def _read_station_dir(
    root: Union[IsmnRoot, Path, str],
    stat_dir: Union[Path, str],
    temp_root: Path,
    custom_meta_reader: list,
) -> (dict, list):
    """
    Parallelizable function to read metadata for files in station dir
    """
    infos = []

    if not isinstance(root, IsmnRoot):
        proc_root = True
        root = IsmnRoot(root)
    else:
        proc_root = False

    csv = root.find_files(stat_dir, "*.csv")

    try:
        if len(csv) == 0:
            raise IsmnFileError(
                "Expected 1 csv file for station, found 0. "
                "Use empty static metadata."
            )
        else:
            if len(csv) > 1:
                infos.append(
                    f"Expected 1 csv file for station, found {len(csv)}. "
                    f"Use first file in dir."
                )
            static_meta_file = StaticMetaFile(
                root, csv[0], load_metadata=True, temp_root=temp_root
            )
            station_meta = static_meta_file.metadata
    except IsmnFileError as e:
        infos.append(f"Error loading static meta for station: {e}")
        station_meta = MetaData([MetaVar(k, v) for k, v in CSV_META_TEMPLATE.items()])

    data_files = root.find_files(stat_dir, "*.stm")

    filelist = []

    for file_path in data_files:
        try:
            f = DataFile(root, file_path, temp_root=temp_root)
        except IOError as e:
            infos.append(f"Error loading ismn file: {e}")
            continue

        f.metadata.merge(station_meta, inplace=True, exclude_empty=False)

        f.metadata = f.metadata.best_meta_for_depth(
            Depth(
                f.metadata["instrument"].depth.start,
                f.metadata["instrument"].depth.end,
            )
        )

        # If custom metadata readers are available
        if custom_meta_reader is not None:
            for cmr in np.atleast_1d(custom_meta_reader):
                cmeta = cmr.read_metadata(f.metadata)
                if isinstance(cmeta, dict):
                    cmeta = MetaData([MetaVar(k, v) for k, v in cmeta.items()])
                f.metadata.merge(cmeta, inplace=True)


        network = f.metadata["network"].val
        station = f.metadata["station"].val

        filelist.append((network, station, f))

        infos.append(f"Processed file {file_path}")

    if proc_root:
        root.close()

    return filelist, infos


def _load_metadata_df(meta_csv_file: Union[str, Path]) -> pd.DataFrame:
    """
    Load metadata data frame from csv file
    """

    metadata_df = pd.read_csv(
        meta_csv_file,
        index_col=0,
        header=[0, 1],
        low_memory=False,
        engine="c",
    )

    # parse date cols as datetime
    for col in ["timerange_from", "timerange_to"]:
        metadata_df[col, "val"] = pd.to_datetime(metadata_df[col, "val"])

    lvars = []
    for c in metadata_df.columns:
        if c[0] not in lvars:
            lvars.append(c[0])

    # we assume triples for all vars except these, so they must be at the end
    assert lvars[-2:] == [
        "file_path",
        "file_type",
    ], "file_type and file_path must be at the end."

    metadata_df.index.name = "idx"

    return metadata_df


class IsmnFileCollection(object):
    """
    The IsmnFileCollection class contains a list of file handlers to access data
    in the given data directory. The file list can be loaded from a previously
    stored csv file, or built by iterating over all files in the data root.
    This class also contains function to load filehandlers for certain networks
    only.

    Attributes
    ----------
    root : IsmnRoot
        Root object where data is stored.
    filelist : collections.OrderedDict
        A collection of filehandlers and network names
    temp_root : Path
        Temporary root dir.
    """

    def __init__(self, root, filelist, temp_root=gettempdir()):
        """
        Parameters
        ----------
        root : IsmnRoot
            Root object where data is stored.
        filelist : collections.OrderedDict
            A collection of filehandler stored in lists with network name as key.
        temp_root : Path or str, optional (default : gettempdir())
            Root directory where a separate subdir for temporary files
            will be created (and deleted).
        """
        self.root = root
        self.filelist = filelist
        self.temp_root = Path(temp_root)

        os.makedirs(self.temp_root, exist_ok=True)

    def __repr__(self):
        return f"{self.__class__.__name__} for {len(self.filelist.keys())} Networks"

    @classmethod
    def build_from_scratch(
        cls, data_root, parallel=True, log_path=None, temp_root=gettempdir(),
        custom_meta_readers=None,
    ):
        """
        Parameters
        ----------
        data_root : IsmnRoot or str or Path
            Root path of ISMN files or path to metadata pkl file.
            i.e. path to the downloaded zip file or the extracted zip directory (faster)
            or a file list that contains these infos already.
        parallel : bool, optional (default: True)
            Speed up metadata collecting with multiple processes.
        log_path : str or Path, optional (default: None)
            Path where the log file is created. If None is set, no log file
            will be written.
        temp_root : str or Path, (default: gettempdir())
            Temporary folder where extracted data is copied during reading from
            zip archive.
        custom_meta_readers: tuple, optional (default: None)
            Custom metadata readers
        """
        t0 = time.time()
        if isinstance(data_root, IsmnRoot):
            root = data_root
        else:
            root = IsmnRoot(data_root)

        os.makedirs(temp_root, exist_ok=True)

        if log_path is not None:
            log_file = os.path.join(log_path, f"{root.name}.log")
        else:
            log_file = None

        if log_file:
            os.makedirs(os.path.dirname(log_file), exist_ok=True)
            logging.basicConfig(
                filename=log_file,
                level=logging.INFO,
                format="%(levelname)s %(asctime)s %(message)s",
                datefmt="%Y-%m-%d %H:%M:%S",
            )

        n_proc = 1 if not parallel else cpu_count()

        logging.info(f"Collecting metadata with {n_proc} processes.")

        print(
            f"Processing metadata for all ismn stations into folder {root.path}.\n"
            f"This may take a few minutes, but is only done once..."
            f"\n{'Hint: Use `parallel=True` to speed up metadata generation for large datasets' if not parallel else ''}"
        )

        process_stat_dirs = []
        for net_dir, stat_dirs in root.cont.items():
            process_stat_dirs += list(stat_dirs)

        args = [
            (root.path if root.zip else root, d, temp_root, custom_meta_readers)
            for d in process_stat_dirs
        ]

        pbar = tqdm(total=len(args), desc="Files Processed")

        fl_elements = []

        def update(r):
            net_stat_fh, infos = r
            for i in infos:
                logging.info(i)
            for elements in net_stat_fh:
                fl_elements.append(elements)
            pbar.update()

        def error(e):
            logging.error(e)
            pbar.update()

        if n_proc == 1:
            for arg in args:
                try:
                    r = _read_station_dir(*arg)
                    update(r)
                except Exception as e:
                    error(e)
        else:
            with Pool(n_proc) as pool:
                for arg in args:
                    pool.apply_async(
                        _read_station_dir,
                        arg,
                        callback=update,
                        error_callback=error,
                    )
                pool.close()
                pool.join()

        pbar.close()

        fl_elements.sort(key=itemgetter(0, 1))  # sort by net name, stat name
        # to ensure alphabetical order... not sure if necessary, slow?
        filelist = OrderedDict([])
        for net, stat, fh in fl_elements:
            if net not in filelist.keys():
                filelist[net] = []
            filelist[net].append(fh)

        t1 = time.time()
        info = f"Metadata generation finished after {int(t1-t0)} Seconds."
        if log_file is not None:
            info += f"\nMetadata and Log stored in {os.path.dirname(log_file)}"

        logging.info(info)
        print(info)

        return cls(root, filelist=filelist)

    @classmethod
    def from_metadata_csv(
        cls, data_root, meta_csv_file, network=None, temp_root=gettempdir()
    ):
        """
        Load a previously created and stored filelist from
        :func:`ismn.filecollection.IsmnFileCollection.to_metadata_csv`

        Parameters
        ----------
        data_root : IsmnRoot or str or Path
            Path where the ismn data is stored, can also be a zip file
        meta_csv_file : str or Path
            Csv file where the metadata is stored.
        network : list, optional (default: None)
            List of networks that are considered.
            Filehandlers for other networks are set to None.
        temp_root : str or Path, optional (default: gettempdir())
            Temporary folder where extracted data is copied during reading from
            zip archive.
        """
        if network is not None:
            network = np.atleast_1d(network)

        if isinstance(data_root, IsmnRoot):
            root = data_root
        else:
            root = IsmnRoot(data_root)

        print(f"Found existing ismn metadata in {meta_csv_file}.")

        metadata_df = _load_metadata_df(meta_csv_file)

        filelist = OrderedDict([])

        all_networks = metadata_df["network"]["val"].values

        columns = np.array(list(metadata_df.columns))

        for i, row in enumerate(metadata_df.values):  # todo: slow!?? parallelise?
            this_nw = all_networks[i]
            if (network is not None) and not np.isin([this_nw], network)[0]:
                f = None
                continue
            else:
                vars = np.unique(columns[:-2][:, 0])
                vals = row[:-2].reshape(-1, 3)

                metadata = MetaData(
                    [
                        MetaVar.from_tuple(
                            (vars[i], vals[i][2], vals[i][0], vals[i][1])
                        )
                        for i in range(len(vars))
                    ]
                )

                f = DataFile(
                    root=root,
                    file_path=str(PurePosixPath(row[-2])),
                    load_metadata=False,
                    temp_root=temp_root,
                )

                f.metadata = metadata
                f.file_type = row[-1]

                this_nw = f.metadata["network"].val

            if this_nw not in filelist.keys():
                filelist[this_nw] = []

            filelist[this_nw].append(f)

        if network is None:
            cls.metadata_df = metadata_df
        else:
            flags = np.isin(metadata_df["network"]["val"].values, network)
            cls.metadata_df = metadata_df.loc[flags]

        return cls(root, filelist=filelist)

    def to_metadata_csv(self, meta_csv_file):
        """
        Write filehandle metadata from filelist to metdata csv that contains
        ALL metadata / variables of the filehander.
        Can be read back in as filelist with filehandlers using
        :func:`ismn.filecollection.IsmnFileCollection.from_metadata_csv`.

        Parameters
        ----------
        meta_csv_file : Path or str, optional (default: None)
            Directory where the csv file with the correct name is crated
        """

        dfs = []

        for i, filehandler in enumerate(self.iter_filehandlers()):
            df = filehandler.metadata.to_pd(True, dropna=False)
            df[("file_path", "val")] = str(PurePosixPath(filehandler.file_path))
            df[("file_type", "val")] = filehandler.file_type

            df.index = [i]
            dfs.append(df)
            i += 1

        dfs = pd.concat(dfs, axis=0, sort=True)
        cols_end = ["file_path", "file_type"]

        dfs = dfs[
            [c for c in dfs.columns if c[0] not in cols_end]
            + [c for c in dfs.columns if c[0] in cols_end]
        ]
        dfs = dfs.fillna(np.nan)

        os.makedirs(Path(os.path.dirname(meta_csv_file)), exist_ok=True)
        dfs.to_csv(meta_csv_file)

    def get_filehandler(self, idx):
        """
        Get the nth filehandler in a list of all filehandlers for all networks.
        e.g. if there are 2 networks, with 3 filehandlers/sensors each, idx=4
        will return the first filehandler of the second network.

        Parameters
        ----------
        idx: int
            Index of filehandler to read.

        Returns
        -------
        filehandler : DataFile
            nth filehandler of all filehandlers in the sorted list.
        """
        fs = 0
        for net, files in self.filelist.items():
            l = len(files)
            if fs + l > idx:
                return files[idx - fs]
            else:
                fs += l

    def iter_filehandlers(self, networks=None):
        """
        Iterator over files for networks

        Parameters
        ----------
        networks : list, optional (default: None)
            Name of networks to get files for, or None to use all networks.

        Yields
        -------
        file : DataFile
            Filehandler with metadata
        """
        for net, files in self.filelist.items():
            if (networks is None) or (net in networks):
                for f in files:
                    yield f
        yield from ()  # in case networks is an empty list

    def close(self):
        # close root and all filehandlers
        self.root.close()
        for f in self.iter_filehandlers():
            f.close()

1	# The MIT License (MIT)
2	#
3	# Copyright (c) 2021 TU Wien
4	#
5	# Permission is hereby granted, free of charge, to any person obtaining a copy
6	# of this software and associated documentation files (the "Software"), to deal
7	# in the Software without restriction, including without limitation the rights
8	# to use, copy, modify, merge, publish, distribute, sublicense, and/or sell
9	# copies of the Software, and to permit persons to whom the Software is
10	# furnished to do so, subject to the following conditions:
11	#
12	# The above copyright notice and this permission notice shall be included in all
13	# copies or substantial portions of the Software.
14	#
15	# THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
16	# IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
17	# FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE
18	# AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
19	# LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,
20	# OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE
21	# SOFTWARE.
22
23	import logging	4✔
24
25	from tempfile import gettempdir	4✔
26	from pathlib import Path, PurePosixPath	4✔
27
28	import numpy as np	4✔
29	from tqdm import tqdm	4✔
30	from typing import Union	4✔
31	from multiprocessing import Pool, cpu_count	4✔
32	from operator import itemgetter	4✔
33	import time	4✔
34
35	from ismn.base import IsmnRoot	4✔
36	from ismn.const import *	4✔
37	from ismn.filehandlers import DataFile, StaticMetaFile	4✔
38	from ismn.meta import MetaData, MetaVar, Depth	4✔
39
40
41	def _read_station_dir(	4✔
42	root: Union[IsmnRoot, Path, str],
43	stat_dir: Union[Path, str],
44	temp_root: Path,
45	custom_meta_reader: list,
46	) -> (dict, list):
47	"""
48	Parallelizable function to read metadata for files in station dir
49	"""
50	infos = []	4✔
51
52	if not isinstance(root, IsmnRoot):	4✔
53	proc_root = True	4✔
54	root = IsmnRoot(root)	4✔
55	else:
56	proc_root = False	4✔
57
58	csv = root.find_files(stat_dir, "*.csv")	4✔
59
60	try:	4✔
61	if len(csv) == 0:	4!
62	raise IsmnFileError(	×
63	"Expected 1 csv file for station, found 0. "
64	"Use empty static metadata."
65	)
66	else:
67	if len(csv) > 1:	4!
68	infos.append(	×
69	f"Expected 1 csv file for station, found {len(csv)}. "
70	f"Use first file in dir."
71	)
72	static_meta_file = StaticMetaFile(	4✔
73	root, csv[0], load_metadata=True, temp_root=temp_root
74	)
75	station_meta = static_meta_file.metadata	4✔
76	except IsmnFileError as e:	×
77	infos.append(f"Error loading static meta for station: {e}")	×
78	station_meta = MetaData([MetaVar(k, v) for k, v in CSV_META_TEMPLATE.items()])	×
79
80	data_files = root.find_files(stat_dir, "*.stm")	4✔
81
82	filelist = []	4✔
83
84	for file_path in data_files:	4✔
85	try:	4✔
86	f = DataFile(root, file_path, temp_root=temp_root)	4✔
87	except IOError as e:	×
88	infos.append(f"Error loading ismn file: {e}")	×
89	continue	×
90
91	f.metadata.merge(station_meta, inplace=True, exclude_empty=False)	4✔
92
93	f.metadata = f.metadata.best_meta_for_depth(	4✔
94	Depth(
95	f.metadata["instrument"].depth.start,
96	f.metadata["instrument"].depth.end,
97	)
98	)
99
100	# If custom metadata readers are available
101	if custom_meta_reader is not None:	4✔
102	for cmr in np.atleast_1d(custom_meta_reader):	4✔
103	cmeta = cmr.read_metadata(f.metadata)	4✔
104	if isinstance(cmeta, dict):	4!
NEW 105	cmeta = MetaData([MetaVar(k, v) for k, v in cmeta.items()])	×
106	f.metadata.merge(cmeta, inplace=True)	4✔
107
108
109	network = f.metadata["network"].val	4✔
110	station = f.metadata["station"].val	4✔
111
112	filelist.append((network, station, f))	4✔
113
114	infos.append(f"Processed file {file_path}")	4✔
115
116	if proc_root:	4✔
117	root.close()	4✔
118
119	return filelist, infos	4✔
120
121
122	def _load_metadata_df(meta_csv_file: Union[str, Path]) -> pd.DataFrame:	4✔
123	"""
124	Load metadata data frame from csv file
125	"""
126
127	metadata_df = pd.read_csv(	4✔
128	meta_csv_file,
129	index_col=0,
130	header=[0, 1],
131	low_memory=False,
132	engine="c",
133	)
134
135	# parse date cols as datetime
136	for col in ["timerange_from", "timerange_to"]:	4✔
137	metadata_df[col, "val"] = pd.to_datetime(metadata_df[col, "val"])	4✔
138
139	lvars = []	4✔
140	for c in metadata_df.columns:	4✔
141	if c[0] not in lvars:	4✔
142	lvars.append(c[0])	4✔
143
144	# we assume triples for all vars except these, so they must be at the end
145	assert lvars[-2:] == [	4✔
146	"file_path",
147	"file_type",
148	], "file_type and file_path must be at the end."
149
150	metadata_df.index.name = "idx"	4✔
151
152	return metadata_df	4✔
153
154
155	class IsmnFileCollection(object):	4✔
156	"""
157	The IsmnFileCollection class contains a list of file handlers to access data
158	in the given data directory. The file list can be loaded from a previously
159	stored csv file, or built by iterating over all files in the data root.
160	This class also contains function to load filehandlers for certain networks
161	only.
162
163	Attributes
164	----------
165	root : IsmnRoot
166	Root object where data is stored.
167	filelist : collections.OrderedDict
168	A collection of filehandlers and network names
169	temp_root : Path
170	Temporary root dir.
171	"""
172
173	def __init__(self, root, filelist, temp_root=gettempdir()):	4✔
174	"""
175	Parameters
176	----------
177	root : IsmnRoot
178	Root object where data is stored.
179	filelist : collections.OrderedDict
180	A collection of filehandler stored in lists with network name as key.
181	temp_root : Path or str, optional (default : gettempdir())
182	Root directory where a separate subdir for temporary files
183	will be created (and deleted).
184	"""
185	self.root = root	4✔
186	self.filelist = filelist	4✔
187	self.temp_root = Path(temp_root)	4✔
188
189	os.makedirs(self.temp_root, exist_ok=True)	4✔
190
191	def __repr__(self):
192	return f"{self.__class__.__name__} for {len(self.filelist.keys())} Networks"
193
194	@classmethod	4✔
195	def build_from_scratch(	4✔
196	cls, data_root, parallel=True, log_path=None, temp_root=gettempdir(),
197	custom_meta_readers=None,
198	):
199	"""
200	Parameters
201	----------
202	data_root : IsmnRoot or str or Path
203	Root path of ISMN files or path to metadata pkl file.
204	i.e. path to the downloaded zip file or the extracted zip directory (faster)
205	or a file list that contains these infos already.
206	parallel : bool, optional (default: True)
207	Speed up metadata collecting with multiple processes.
208	log_path : str or Path, optional (default: None)
209	Path where the log file is created. If None is set, no log file
210	will be written.
211	temp_root : str or Path, (default: gettempdir())
212	Temporary folder where extracted data is copied during reading from
213	zip archive.
214	custom_meta_readers: tuple, optional (default: None)
215	Custom metadata readers
216	"""
217	t0 = time.time()	4✔
218	if isinstance(data_root, IsmnRoot):	4✔
219	root = data_root	4✔
220	else:
221	root = IsmnRoot(data_root)	4✔
222
223	os.makedirs(temp_root, exist_ok=True)	4✔
224
225	if log_path is not None:	4✔
226	log_file = os.path.join(log_path, f"{root.name}.log")	4✔
227	else:
228	log_file = None	4✔
229
230	if log_file:	4✔
231	os.makedirs(os.path.dirname(log_file), exist_ok=True)	4✔
232	logging.basicConfig(	4✔
233	filename=log_file,
234	level=logging.INFO,
235	format="%(levelname)s %(asctime)s %(message)s",
236	datefmt="%Y-%m-%d %H:%M:%S",
237	)
238
239	n_proc = 1 if not parallel else cpu_count()	4✔
240
241	logging.info(f"Collecting metadata with {n_proc} processes.")	4✔
242
243	print(	4✔
244	f"Processing metadata for all ismn stations into folder {root.path}.\n"
245	f"This may take a few minutes, but is only done once..."
246	f"\n{'Hint: Use `parallel=True` to speed up metadata generation for large datasets' if not parallel else ''}"
247	)
248
249	process_stat_dirs = []	4✔
250	for net_dir, stat_dirs in root.cont.items():	4✔
251	process_stat_dirs += list(stat_dirs)	4✔
252
253	args = [	4✔
254	(root.path if root.zip else root, d, temp_root, custom_meta_readers)
255	for d in process_stat_dirs
256	]
257
258	pbar = tqdm(total=len(args), desc="Files Processed")	4✔
259
260	fl_elements = []	4✔
261
262	def update(r):	4✔
263	net_stat_fh, infos = r	4✔
264	for i in infos:	4✔
265	logging.info(i)	4✔
266	for elements in net_stat_fh:	4✔
267	fl_elements.append(elements)	4✔
268	pbar.update()	4✔
269
270	def error(e):	4✔
UNCOV 271	logging.error(e)	×
272	pbar.update()	×
273
274	if n_proc == 1:	4✔
275	for arg in args:	4✔
276	try:	4✔
277	r = _read_station_dir(*arg)	4✔
278	update(r)	4✔
UNCOV 279	except Exception as e:	×
280	error(e)	×
281	else:
282	with Pool(n_proc) as pool:	4✔
283	for arg in args:	4✔
284	pool.apply_async(	4✔
285	_read_station_dir,
286	arg,
287	callback=update,
288	error_callback=error,
289	)
290	pool.close()	4✔
291	pool.join()	4✔
292
293	pbar.close()	4✔
294
295	fl_elements.sort(key=itemgetter(0, 1)) # sort by net name, stat name	4✔
296	# to ensure alphabetical order... not sure if necessary, slow?
297	filelist = OrderedDict([])	4✔
298	for net, stat, fh in fl_elements:	4✔
299	if net not in filelist.keys():	4✔
300	filelist[net] = []	4✔
301	filelist[net].append(fh)	4✔
302
303	t1 = time.time()	4✔
304	info = f"Metadata generation finished after {int(t1-t0)} Seconds."	4✔
305	if log_file is not None:	4✔
306	info += f"\nMetadata and Log stored in {os.path.dirname(log_file)}"	4✔
307
308	logging.info(info)	4✔
309	print(info)	4✔
310
311	return cls(root, filelist=filelist)	4✔
312
313	@classmethod	4✔
314	def from_metadata_csv(	4✔
315	cls, data_root, meta_csv_file, network=None, temp_root=gettempdir()
316	):
317	"""
318	Load a previously created and stored filelist from
319	:func:`ismn.filecollection.IsmnFileCollection.to_metadata_csv`
320
321	Parameters
322	----------
323	data_root : IsmnRoot or str or Path
324	Path where the ismn data is stored, can also be a zip file
325	meta_csv_file : str or Path
326	Csv file where the metadata is stored.
327	network : list, optional (default: None)
328	List of networks that are considered.
329	Filehandlers for other networks are set to None.
330	temp_root : str or Path, optional (default: gettempdir())
331	Temporary folder where extracted data is copied during reading from
332	zip archive.
333	"""
334	if network is not None:	4✔
335	network = np.atleast_1d(network)	4✔
336
337	if isinstance(data_root, IsmnRoot):	4✔
338	root = data_root	4✔
339	else:
340	root = IsmnRoot(data_root)	4✔
341
342	print(f"Found existing ismn metadata in {meta_csv_file}.")	4✔
343
344	metadata_df = _load_metadata_df(meta_csv_file)	4✔
345
346	filelist = OrderedDict([])	4✔
347
348	all_networks = metadata_df["network"]["val"].values	4✔
349
350	columns = np.array(list(metadata_df.columns))	4✔
351
352	for i, row in enumerate(metadata_df.values): # todo: slow!?? parallelise?	4✔
353	this_nw = all_networks[i]	4✔
354	if (network is not None) and not np.isin([this_nw], network)[0]:	4✔
355	f = None	4✔
356	continue	4✔
357	else:
358	vars = np.unique(columns[:-2][:, 0])	4✔
359	vals = row[:-2].reshape(-1, 3)	4✔
360
361	metadata = MetaData(	4✔
362	[
363	MetaVar.from_tuple(
364	(vars[i], vals[i][2], vals[i][0], vals[i][1])
365	)
366	for i in range(len(vars))
367	]
368	)
369
370	f = DataFile(	4✔
371	root=root,
372	file_path=str(PurePosixPath(row[-2])),
373	load_metadata=False,
374	temp_root=temp_root,
375	)
376
377	f.metadata = metadata	4✔
378	f.file_type = row[-1]	4✔
379
380	this_nw = f.metadata["network"].val	4✔
381
382	if this_nw not in filelist.keys():	4✔
383	filelist[this_nw] = []	4✔
384
385	filelist[this_nw].append(f)	4✔
386
387	if network is None:	4✔
388	cls.metadata_df = metadata_df	4✔
389	else:
390	flags = np.isin(metadata_df["network"]["val"].values, network)	4✔
391	cls.metadata_df = metadata_df.loc[flags]	4✔
392
393	return cls(root, filelist=filelist)	4✔
394
395	def to_metadata_csv(self, meta_csv_file):	4✔
396	"""
397	Write filehandle metadata from filelist to metdata csv that contains
398	ALL metadata / variables of the filehander.
399	Can be read back in as filelist with filehandlers using
400	:func:`ismn.filecollection.IsmnFileCollection.from_metadata_csv`.
401
402	Parameters
403	----------
404	meta_csv_file : Path or str, optional (default: None)
405	Directory where the csv file with the correct name is crated
406	"""
407
408	dfs = []	4✔
409
410	for i, filehandler in enumerate(self.iter_filehandlers()):	4✔
411	df = filehandler.metadata.to_pd(True, dropna=False)	4✔
412	df[("file_path", "val")] = str(PurePosixPath(filehandler.file_path))	4✔
413	df[("file_type", "val")] = filehandler.file_type	4✔
414
415	df.index = [i]	4✔
416	dfs.append(df)	4✔
417	i += 1	4✔
418
419	dfs = pd.concat(dfs, axis=0, sort=True)	4✔
420	cols_end = ["file_path", "file_type"]	4✔
421
422	dfs = dfs[	4✔
423	[c for c in dfs.columns if c[0] not in cols_end]
424	+ [c for c in dfs.columns if c[0] in cols_end]
425	]
426	dfs = dfs.fillna(np.nan)	4✔
427
428	os.makedirs(Path(os.path.dirname(meta_csv_file)), exist_ok=True)	4✔
429	dfs.to_csv(meta_csv_file)	4✔
430
431	def get_filehandler(self, idx):	4✔
432	"""
433	Get the nth filehandler in a list of all filehandlers for all networks.
434	e.g. if there are 2 networks, with 3 filehandlers/sensors each, idx=4
435	will return the first filehandler of the second network.
436
437	Parameters
438	----------
439	idx: int
440	Index of filehandler to read.
441
442	Returns
443	-------
444	filehandler : DataFile
445	nth filehandler of all filehandlers in the sorted list.
446	"""
447	fs = 0	4✔
448	for net, files in self.filelist.items():	4!
449	l = len(files)	4✔
450	if fs + l > idx:	4!
451	return files[idx - fs]	4✔
452	else:
UNCOV 453	fs += l	×
454
455	def iter_filehandlers(self, networks=None):	4✔
456	"""
457	Iterator over files for networks
458
459	Parameters
460	----------
461	networks : list, optional (default: None)
462	Name of networks to get files for, or None to use all networks.
463
464	Yields
465	-------
466	file : DataFile
467	Filehandler with metadata
468	"""
469	for net, files in self.filelist.items():	4✔
470	if (networks is None) or (net in networks):	4✔
471	for f in files:	4✔
472	yield f	4✔
473	yield from () # in case networks is an empty list	4✔
474
475	def close(self):	4✔
476	# close root and all filehandlers
477	self.root.close()	4✔
478	for f in self.iter_filehandlers():	4✔
479	f.close()	4✔

TUW-GEO / ismn / 3578611372

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