9517288918

Committed 14 Jun 2024 01:53PM UTC coverage: 88.671% (-0.1%) from 88.796%

Build # 9517288918

Build Type

push

github

Committed by

web-flow

Commit Message

Fix error logging when csv file is missing (#86)

* Fix error logging when csv file is missing

* Change BasicGrid to CellGrid

Fix #85

Run Details

691 of 821 branches covered (84.17%)

Branch coverage included in aggregate %.

7 of 12 new or added lines in 2 files covered. (58.33%)

1 existing line in 1 file now uncovered.

1430 of 1571 relevant lines covered (91.02%)

3.64 hits per line

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

86.86

/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

import os
from tempfile import gettempdir
from pathlib import Path, PurePosixPath
import numpy as np
from typing import Union
from operator import itemgetter
import time
from typing import List, Tuple
import pandas as pd
from collections import OrderedDict
from repurpose.process import parallel_process_async
import traceback

from ismn.base import IsmnRoot
import ismn.const as const
from ismn.const import ismnlog
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,
) -> Tuple[List, List]:
    """
    Parallelizable function to read metadata for files in station dir
    """
    logger = logging.getLogger('ismn_meta_collector')

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

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

    erroneous_files = []

    try:
        if len(csv) == 0:
            raise const.IsmnFileError(
                f"Expected 1 csv file but got 0 for station {stat_dir}. "
                "Continue with empty static metadata instead.")
        else:
            if len(csv) > 1:
                logger.warning(
                    f"Expected 1 csv file but got {len(csv)} for "
                    f"station {stat_dir}. "
                    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 const.IsmnFileError as e:
        _csv = "*missing*" if len(csv) == 0 else csv[0]
        logger.warning(f"Error loading static meta file for {stat_dir}/{_csv}. "
                       f"We will use the placeholder metadata here. "
                       f"Error traceback: {traceback.format_exc()}")
        station_meta = MetaData(
            [MetaVar(k, v) for k, v in const.CSV_META_TEMPLATE.items()]
        )
        erroneous_files.append(os.path.join(str(root.path), str(stat_dir), str(_csv)))

    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 Exception as e:
            logger.error(f"Error loading ismn file {file_path}. "
                         f"We will skip this file, it might be malformed. "
                         f"Error traceback: {traceback.format_exc()}")
            erroneous_files.append(file_path)

            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()])
                if cmeta is not None:
                    f.metadata.merge(cmeta, inplace=True)

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

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

        logger.info(f"Processed file {file_path}")

    if proc_root:
        root.close()

    return filelist, erroneous_files


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)

        log_filename = f"{root.name}.log"

        n_proc = 1 if not parallel else os.cpu_count()

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

        if not parallel:
            hint = 'Hint: Use `parallel=True` to speed up metadata ' \
                   'generation for large datasets'
        else:
            hint = ''

        print(
            f"Collecting metadata for all ismn stations in archive "
            f"{root.path}.\n"
            f"This may take a few minutes, but is only done once...\n{hint}"
        )

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

        STATIC_KWARGS = {
            'root': root.path if root.zip else root,
            'temp_root': temp_root,
            'custom_meta_reader': custom_meta_readers,
        }

        ITER_KWARGS = {
            'stat_dir': process_stat_dirs
        }

        res = parallel_process_async(
            _read_station_dir, ITER_KWARGS=ITER_KWARGS,
            STATIC_KWARGS=STATIC_KWARGS,
            n_proc=n_proc, show_progress_bars=True,
            ignore_errors=True, log_path=log_path,
            log_filename=log_filename, logger_name='ismn_meta_collector',
            loglevel='INFO', progress_bar_label="Stations Processed",
            backend='loky', verbose=False,
        )

        elements = []
        errors = []
        for r in res:
            elements += r[0]
            if len(r[1]) > 0:
                errors += r[1]

        elements.sort(key=itemgetter(0, 1))  # sort by net name, stat name

        filelist = OrderedDict([])
        for net, stat, fh in elements:
            if net not in filelist.keys():
                filelist[net] = []
            filelist[net].append(fh)

        t1 = time.time()
        info = f"Metadata collection finished after {int(t1-t0)} Seconds."
        if log_path is not None:
            info += (f"\nMetadata for this archive and "
                     f"Logfile stored in {log_path}")

        if len(errors) > 0:
            info += (f"\nNOTE: {len(errors)} potentially malformed file(s) "
                     f"found during metadata collection. These files will be "
                     f"ignored by the reader. "
                     f"Affected files and errors are listed in: "
                     f"{os.path.join(log_path, log_filename)}.")

            with open(os.path.join(log_path, log_filename), mode='a') as f:
                f.write("\n----- Summary: Erroneous Files -----\n")
                for e in errors:
                    f.write(f"{e}\n")
                f.write("\n-------------------------------------\n")

        ismnlog.info(info)
        print(info)

        return cls(root, filelist=filelist)

    @classmethod
    def from_metadata_df(cls, data_root, metadata_df, 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
        metadata_df : pd.DataFrame
            Metadata frame
        temp_root : str or Path, optional (default: gettempdir())
            Temporary folder where extracted data is copied during reading from
            zip archive.
        """
        if isinstance(data_root, IsmnRoot):
            root = data_root
        else:
            root = IsmnRoot(data_root)

        filelist = OrderedDict([])

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

        for i, row in enumerate(metadata_df.values):
            # this_nw = row.loc['network', 'val']
            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=Path(str(PurePosixPath(row[-2]))),
                load_metadata=False,
                temp_root=temp_root,
                verify_filepath=False,
                verify_temp_root=False,
            )

            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)

        cls.metadata_df = metadata_df

        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)

        print(f"Using the existing ismn metadata in {meta_csv_file} to set "
              f"up ISMN_Interface. \n"
              "If there are issues with the data reader, you can remove "
              "the metadata csv file to repeat metadata collection.")

        metadata_df = _load_metadata_df(meta_csv_file)

        if network is not None:
            metadata_df = metadata_df[np.isin(metadata_df["network"].values,
                                              network)]

        metadata_df.index = range(len(metadata_df.index))

        return cls.from_metadata_df(
            data_root, metadata_df, temp_root=temp_root)

    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.infer_objects(copy=False).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	import os	4✔
26	from tempfile import gettempdir	4✔
27	from pathlib import Path, PurePosixPath	4✔
28	import numpy as np	4✔
29	from typing import Union	4✔
30	from operator import itemgetter	4✔
31	import time	4✔
32	from typing import List, Tuple	4✔
33	import pandas as pd	4✔
34	from collections import OrderedDict	4✔
35	from repurpose.process import parallel_process_async	4✔
36	import traceback	4✔
37
38	from ismn.base import IsmnRoot	4✔
39	import ismn.const as const	4✔
40	from ismn.const import ismnlog	4✔
41	from ismn.filehandlers import DataFile, StaticMetaFile	4✔
42	from ismn.meta import MetaData, MetaVar, Depth	4✔
43
44
45	def _read_station_dir(	4✔
46	root: Union[IsmnRoot, Path, str],
47	stat_dir: Union[Path, str],
48	temp_root: Path,
49	custom_meta_reader: list,
50	) -> Tuple[List, List]:
51	"""
52	Parallelizable function to read metadata for files in station dir
53	"""
54	logger = logging.getLogger('ismn_meta_collector')	4✔
55
56	if not isinstance(root, IsmnRoot):	4✔
57	proc_root = True	4✔
58	root = IsmnRoot(root)	4✔
59	else:
60	proc_root = False	4✔
61
62	csv = root.find_files(stat_dir, "*.csv")	4✔
63
64	erroneous_files = []	4✔
65
66	try:	4✔
67	if len(csv) == 0:	4!
68	raise const.IsmnFileError(	×
69	f"Expected 1 csv file but got 0 for station {stat_dir}. "
70	"Continue with empty static metadata instead.")
71	else:
72	if len(csv) > 1:	4!
73	logger.warning(	×
74	f"Expected 1 csv file but got {len(csv)} for "
75	f"station {stat_dir}. "
76	f"Use first file in dir.")
77	static_meta_file = StaticMetaFile(	4✔
78	root, csv[0], load_metadata=True, temp_root=temp_root)
79	station_meta = static_meta_file.metadata	4✔
80	except const.IsmnFileError as e:	×
NEW 81	_csv = "missing" if len(csv) == 0 else csv[0]	×
NEW 82	logger.warning(f"Error loading static meta file for {stat_dir}/{_csv}. "	×
83	f"We will use the placeholder metadata here. "
84	f"Error traceback: {traceback.format_exc()}")
85	station_meta = MetaData(	×
86	[MetaVar(k, v) for k, v in const.CSV_META_TEMPLATE.items()]
87	)
NEW 88	erroneous_files.append(os.path.join(str(root.path), str(stat_dir), str(_csv)))	×
89
90	data_files = root.find_files(stat_dir, "*.stm")	4✔
91
92	filelist = []	4✔
93
94	for file_path in data_files:	4✔
95	try:	4✔
96	f = DataFile(root, file_path, temp_root=temp_root)	4✔
97	except Exception as e:	×
98	logger.error(f"Error loading ismn file {file_path}. "	×
99	f"We will skip this file, it might be malformed. "
100	f"Error traceback: {traceback.format_exc()}")
101	erroneous_files.append(file_path)	×
102
103	continue	×
104
105	f.metadata.merge(station_meta, inplace=True, exclude_empty=False)	4✔
106
107	f.metadata = f.metadata.best_meta_for_depth(	4✔
108	Depth(
109	f.metadata["instrument"].depth.start,
110	f.metadata["instrument"].depth.end,
111	))
112
113	# If custom metadata readers are available
114	if custom_meta_reader is not None:	4✔
115	for cmr in np.atleast_1d(custom_meta_reader):	4✔
116	cmeta = cmr.read_metadata(f.metadata)	4✔
117	if isinstance(cmeta, dict):	4!
118	cmeta = MetaData([MetaVar(k, v) for k, v in cmeta.items()])	×
119	if cmeta is not None:	4!
120	f.metadata.merge(cmeta, inplace=True)	4✔
121
122	network = f.metadata["network"].val	4✔
123	station = f.metadata["station"].val	4✔
124
125	filelist.append((network, station, f))	4✔
126
127	logger.info(f"Processed file {file_path}")	4✔
128
129	if proc_root:	4✔
130	root.close()	4✔
131
132	return filelist, erroneous_files	4✔
133
134
135	def _load_metadata_df(meta_csv_file: Union[str, Path]) -> pd.DataFrame:	4✔
136	"""
137	Load metadata data frame from csv file
138	"""
139
140	metadata_df = pd.read_csv(	4✔
141	meta_csv_file,
142	index_col=0,
143	header=[0, 1],
144	low_memory=False,
145	engine="c")
146
147	# parse date cols as datetime
148	for col in ["timerange_from", "timerange_to"]:	4✔
149	metadata_df[col, "val"] = pd.to_datetime(metadata_df[col, "val"])	4✔
150
151	lvars = []	4✔
152	for c in metadata_df.columns:	4✔
153	if c[0] not in lvars:	4✔
154	lvars.append(c[0])	4✔
155
156	# we assume triples for all vars except these, so they must be at the end
157	assert lvars[-2:] == [	4✔
158	"file_path",
159	"file_type",
160	], "file_type and file_path must be at the end."
161
162	metadata_df.index.name = "idx"	4✔
163
164	return metadata_df	4✔
165
166
167	class IsmnFileCollection(object):	4✔
168	"""
169	The IsmnFileCollection class contains a list of file handlers to access data
170	in the given data directory. The file list can be loaded from a previously
171	stored csv file, or built by iterating over all files in the data root.
172	This class also contains function to load filehandlers for certain networks
173	only.
174
175	Attributes
176	----------
177	root : IsmnRoot
178	Root object where data is stored.
179	filelist : collections.OrderedDict
180	A collection of filehandlers and network names
181	temp_root : Path
182	Temporary root dir.
183	"""
184
185	def __init__(self, root, filelist, temp_root=gettempdir()):	4✔
186	"""
187	Parameters
188	----------
189	root : IsmnRoot
190	Root object where data is stored.
191	filelist : collections.OrderedDict
192	A collection of filehandler stored in lists with network name as key.
193	temp_root : Path or str, optional (default : gettempdir())
194	Root directory where a separate subdir for temporary files
195	will be created (and deleted).
196	"""
197	self.root = root	4✔
198	self.filelist = filelist	4✔
199	self.temp_root = Path(temp_root)	4✔
200
201	os.makedirs(self.temp_root, exist_ok=True)	4✔
202
203	def __repr__(self):
204	return f"{self.__class__.__name__} for {len(self.filelist.keys())} Networks"
205
206	@classmethod	4✔
207	def build_from_scratch(	4✔
208	cls,
209	data_root,
210	parallel=True,
211	log_path=None,
212	temp_root=gettempdir(),
213	custom_meta_readers=None,
214	):
215	"""
216	Parameters
217	----------
218	data_root : IsmnRoot or str or Path
219	Root path of ISMN files or path to metadata pkl file.
220	i.e. path to the downloaded zip file or the extracted zip directory (faster)
221	or a file list that contains these infos already.
222	parallel : bool, optional (default: True)
223	Speed up metadata collecting with multiple processes.
224	log_path : str or Path, optional (default: None)
225	Path where the log file is created. If None is set, no log file
226	will be written.
227	temp_root : str or Path, (default: gettempdir())
228	Temporary folder where extracted data is copied during reading from
229	zip archive.
230	custom_meta_readers: tuple, optional (default: None)
231	Custom metadata readers
232	"""
233	t0 = time.time()	4✔
234	if isinstance(data_root, IsmnRoot):	4✔
235	root = data_root	4✔
236	else:
237	root = IsmnRoot(data_root)	4✔
238
239	os.makedirs(temp_root, exist_ok=True)	4✔
240
241	log_filename = f"{root.name}.log"	4✔
242
243	n_proc = 1 if not parallel else os.cpu_count()	4✔
244
245	ismnlog.info(f"Collecting metadata with {n_proc} processes.")	4✔
246
247	if not parallel:	4✔
248	hint = 'Hint: Use `parallel=True` to speed up metadata ' \	4✔
249	'generation for large datasets'
250	else:
251	hint = ''	4✔
252
253	print(	4✔
254	f"Collecting metadata for all ismn stations in archive "
255	f"{root.path}.\n"
256	f"This may take a few minutes, but is only done once...\n{hint}"
257	)
258
259	process_stat_dirs = []	4✔
260	for net_dir, stat_dirs in root.cont.items():	4✔
261	process_stat_dirs += list(stat_dirs)	4✔
262
263	STATIC_KWARGS = {	4✔
264	'root': root.path if root.zip else root,
265	'temp_root': temp_root,
266	'custom_meta_reader': custom_meta_readers,
267	}
268
269	ITER_KWARGS = {	4✔
270	'stat_dir': process_stat_dirs
271	}
272
273	res = parallel_process_async(	4✔
274	_read_station_dir, ITER_KWARGS=ITER_KWARGS,
275	STATIC_KWARGS=STATIC_KWARGS,
276	n_proc=n_proc, show_progress_bars=True,
277	ignore_errors=True, log_path=log_path,
278	log_filename=log_filename, logger_name='ismn_meta_collector',
279	loglevel='INFO', progress_bar_label="Stations Processed",
280	backend='loky', verbose=False,
281	)
282
283	elements = []	4✔
284	errors = []	4✔
285	for r in res:	4✔
286	elements += r[0]	4✔
287	if len(r[1]) > 0:	4!
288	errors += r[1]	×
289
290	elements.sort(key=itemgetter(0, 1)) # sort by net name, stat name	4✔
291
292	filelist = OrderedDict([])	4✔
293	for net, stat, fh in elements:	4✔
294	if net not in filelist.keys():	4✔
295	filelist[net] = []	4✔
296	filelist[net].append(fh)	4✔
297
298	t1 = time.time()	4✔
299	info = f"Metadata collection finished after {int(t1-t0)} Seconds."	4✔
300	if log_path is not None:	4✔
301	info += (f"\nMetadata for this archive and "	4✔
302	f"Logfile stored in {log_path}")
303
304	if len(errors) > 0:	4!
305	info += (f"\nNOTE: {len(errors)} potentially malformed file(s) "	×
306	f"found during metadata collection. These files will be "
307	f"ignored by the reader. "
308	f"Affected files and errors are listed in: "
309	f"{os.path.join(log_path, log_filename)}.")
310
311	with open(os.path.join(log_path, log_filename), mode='a') as f:	×
312	f.write("\n----- Summary: Erroneous Files -----\n")	×
NEW 313	for e in errors:	×
NEW 314	f.write(f"{e}\n")	×
UNCOV 315	f.write("\n-------------------------------------\n")	×
316
317	ismnlog.info(info)	4✔
318	print(info)	4✔
319
320	return cls(root, filelist=filelist)	4✔
321
322	@classmethod	4✔
323	def from_metadata_df(cls, data_root, metadata_df, temp_root=gettempdir()):	4✔
324	"""
325	Load a previously created and stored filelist from
326	:func:`ismn.filecollection.IsmnFileCollection.to_metadata_csv`
327	Parameters
328	----------
329	data_root : IsmnRoot or str or Path
330	Path where the ismn data is stored, can also be a zip file
331	metadata_df : pd.DataFrame
332	Metadata frame
333	temp_root : str or Path, optional (default: gettempdir())
334	Temporary folder where extracted data is copied during reading from
335	zip archive.
336	"""
337	if isinstance(data_root, IsmnRoot):	4✔
338	root = data_root	4✔
339	else:
340	root = IsmnRoot(data_root)	4✔
341
342	filelist = OrderedDict([])	4✔
343
344	columns = np.array(list(metadata_df.columns))	4✔
345
346	for i, row in enumerate(metadata_df.values):	4✔
347	# this_nw = row.loc['network', 'val']
348	vars = np.unique(columns[:-2][:, 0])	4✔
349	vals = row[:-2].reshape(-1, 3)	4✔
350
351	metadata = MetaData([	4✔
352	MetaVar.from_tuple(
353	(vars[i], vals[i][2], vals[i][0], vals[i][1]))
354	for i in range(len(vars))
355	])
356
357	f = DataFile(	4✔
358	root=root,
359	file_path=Path(str(PurePosixPath(row[-2]))),
360	load_metadata=False,
361	temp_root=temp_root,
362	verify_filepath=False,
363	verify_temp_root=False,
364	)
365
366	f.metadata = metadata	4✔
367	f.file_type = row[-1]	4✔
368
369	this_nw = f.metadata["network"].val	4✔
370
371	if this_nw not in filelist.keys():	4✔
372	filelist[this_nw] = []	4✔
373
374	filelist[this_nw].append(f)	4✔
375
376	cls.metadata_df = metadata_df	4✔
377
378	return cls(root, filelist=filelist)	4✔
379
380	@classmethod	4✔
381	def from_metadata_csv(cls,	4✔
382	data_root,
383	meta_csv_file,
384	network=None,
385	temp_root=gettempdir()):
386	"""
387	Load a previously created and stored filelist from
388	:func:`ismn.filecollection.IsmnFileCollection.to_metadata_csv`
389	Parameters
390	----------
391	data_root : IsmnRoot or str or Path
392	Path where the ismn data is stored, can also be a zip file
393	meta_csv_file : str or Path
394	Csv file where the metadata is stored.
395	network : list, optional (default: None)
396	List of networks that are considered.
397	Filehandlers for other networks are set to None.
398	temp_root : str or Path, optional (default: gettempdir())
399	Temporary folder where extracted data is copied during reading from
400	zip archive.
401	"""
402	if network is not None:	4✔
403	network = np.atleast_1d(network)	4✔
404
405	print(f"Using the existing ismn metadata in {meta_csv_file} to set "	4✔
406	f"up ISMN_Interface. \n"
407	"If there are issues with the data reader, you can remove "
408	"the metadata csv file to repeat metadata collection.")
409
410	metadata_df = _load_metadata_df(meta_csv_file)	4✔
411
412	if network is not None:	4✔
413	metadata_df = metadata_df[np.isin(metadata_df["network"].values,	4✔
414	network)]
415
416	metadata_df.index = range(len(metadata_df.index))	4✔
417
418	return cls.from_metadata_df(	4✔
419	data_root, metadata_df, temp_root=temp_root)
420
421	def to_metadata_csv(self, meta_csv_file):	4✔
422	"""
423	Write filehandle metadata from filelist to metdata csv that contains
424	ALL metadata / variables of the filehander.
425	Can be read back in as filelist with filehandlers using
426	:func:`ismn.filecollection.IsmnFileCollection.from_metadata_csv`.
427
428	Parameters
429	----------
430	meta_csv_file : Path or str, optional (default: None)
431	Directory where the csv file with the correct name is crated
432	"""
433
434	dfs = []	4✔
435
436	for i, filehandler in enumerate(self.iter_filehandlers()):	4✔
437	df = filehandler.metadata.to_pd(True, dropna=False)	4✔
438	df[("file_path",	4✔
439	"val")] = str(PurePosixPath(filehandler.file_path))
440	df[("file_type", "val")] = filehandler.file_type	4✔
441
442	df.index = [i]	4✔
443	dfs.append(df)	4✔
444	i += 1	4✔
445
446	dfs = pd.concat(dfs, axis=0, sort=True)	4✔
447	cols_end = ["file_path", "file_type"]	4✔
448
449	dfs = dfs[[c for c in dfs.columns if c[0] not in cols_end] +	4✔
450	[c for c in dfs.columns if c[0] in cols_end]]
451	dfs = dfs.infer_objects(copy=False).fillna(np.nan)	4✔
452
453	os.makedirs(Path(os.path.dirname(meta_csv_file)), exist_ok=True)	4✔
454	dfs.to_csv(meta_csv_file)	4✔
455
456	def get_filehandler(self, idx):	4✔
457	"""
458	Get the nth filehandler in a list of all filehandlers for all networks.
459	e.g. if there are 2 networks, with 3 filehandlers/sensors each, idx=4
460	will return the first filehandler of the second network.
461
462	Parameters
463	----------
464	idx: int
465	Index of filehandler to read.
466
467	Returns
468	-------
469	filehandler : DataFile
470	nth filehandler of all filehandlers in the sorted list.
471	"""
472	fs = 0	4✔
473	for net, files in self.filelist.items():	4!
474	l = len(files)	4✔
475	if fs + l > idx:	4!
476	return files[idx - fs]	4✔
477	else:
478	fs += l	×
479
480	def iter_filehandlers(self, networks=None):	4✔
481	"""
482	Iterator over files for networks
483
484	Parameters
485	----------
486	networks : list, optional (default: None)
487	Name of networks to get files for, or None to use all networks.
488
489	Yields
490	-------
491	file : DataFile
492	Filehandler with metadata
493	"""
494	for net, files in self.filelist.items():	4✔
495	if (networks is None) or (net in networks):	4✔
496	for f in files:	4✔
497	yield f	4✔
498	yield from () # in case networks is an empty list	4✔
499
500	def close(self):	4✔
501	# close root and all filehandlers
502	self.root.close()	4✔
503	for f in self.iter_filehandlers():	4✔
504	f.close()	4✔

TUW-GEO / ismn / 9517288918

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