12763622959

Committed 14 Jan 2025 08:17AM UTC coverage: 97.928%. Remained the same

Build # 12763622959

Build Type

push

github

Committed by

web-flow

Commit Message

Removing duplicate run from pull request creation.

Run Details

898 of 917 relevant lines covered (97.93%)

1.95 hits per line

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

100.0

/pytrnsys_process/headers.py

import pathlib as _pl
from abc import ABC
from collections import defaultdict as _defaultdict
from concurrent.futures import ProcessPoolExecutor

from pytrnsys_process import utils
from pytrnsys_process.readers import HeaderReader


def _process_sim_file(sim_file):
    try:
        headers = HeaderReader().read_headers(sim_file)
        return headers, sim_file.parents[1], sim_file
    except Exception as e:  # pylint: disable=broad-exception-caught
        print(f"Could not read {sim_file}: {e}")
        return None


class Headers:

    header_index: _defaultdict[str, list]

    def __init__(self, path_to_results: _pl.Path):
        self.path_to_results = path_to_results
        self.header_index = _defaultdict(list)

    def init_headers(self):
        sim_files = utils.get_files(
            utils.get_sim_folders(self.path_to_results)
        )
        for sim_file in sim_files:
            try:
                headers = HeaderReader().read_headers(sim_file)
                self._index_headers(headers, sim_file.parents[1], sim_file)
            except Exception as e:  # pylint: disable=broad-exception-caught

                print(f"Could not read {sim_file}: {e}")

    def init_headers_multi_process(self):
        sim_files = utils.get_files(
            utils.get_sim_folders(self.path_to_results)
        )

        with ProcessPoolExecutor() as executor:
            results = executor.map(_process_sim_file, sim_files)
            for result in results:
                if result:
                    headers, sim_folder, sim_file = result
                    self._index_headers(headers, sim_folder, sim_file)

    # TODO: Discuss if something like this is needed # pylint: disable=fixme
    def search_header(self, header_name: str):  # pragma: no cover
        if header_name in self.header_index:
            print(f"Header '{header_name}' found in:")
            for folder, file in self.header_index[header_name]:
                print(f"- Folder: {folder}, File: {file}")
        else:
            print(f"Header '{header_name}' not found in any files.")

    def _index_headers(
        self, headers: list[str], sim_folder: _pl.Path, sim_file: _pl.Path
    ):
        for header in headers:
            self.header_index[header].append((sim_folder.name, sim_file.name))

    # TODO: Add function to validate headers and log files with invalid headers  #pylint: disable=fixme


class HeaderValidationMixin(ABC):
    def validate_headers(
        self, headers: Headers, columns: list[str]
    ) -> tuple[bool, list[str]]:
        """Validates that all columns exist in the headers index.

        Args:
            headers: Headers instance containing the index of available headers
            columns: List of column names to validate

        Returns:
            Tuple of (is_valid, missing_columns)
            - is_valid: True if all columns exist
            - missing_columns: List of column names that are missing
        """
        missing_columns = []
        for column in columns:
            if column not in headers.header_index:
                missing_columns.append(column)

        return len(missing_columns) == 0, missing_columns

1	import pathlib as _pl	2✔
2	from abc import ABC	2✔
3	from collections import defaultdict as _defaultdict	2✔
4	from concurrent.futures import ProcessPoolExecutor	2✔
5
6	from pytrnsys_process import utils	2✔
7	from pytrnsys_process.readers import HeaderReader	2✔
8
9
10	def _process_sim_file(sim_file):	2✔
11	try:	1✔
12	headers = HeaderReader().read_headers(sim_file)	1✔
13	return headers, sim_file.parents[1], sim_file	1✔
14	except Exception as e: # pylint: disable=broad-exception-caught	1✔
15	print(f"Could not read {sim_file}: {e}")	1✔
16	return None	1✔
17
18
19	class Headers:	2✔
20
21	header_index: _defaultdict[str, list]	2✔
22
23	def __init__(self, path_to_results: _pl.Path):	2✔
24	self.path_to_results = path_to_results	2✔
25	self.header_index = _defaultdict(list)	2✔
26
27	def init_headers(self):	2✔
28	sim_files = utils.get_files(	2✔
29	utils.get_sim_folders(self.path_to_results)
30	)
31	for sim_file in sim_files:	2✔
32	try:	2✔
33	headers = HeaderReader().read_headers(sim_file)	2✔
34	self._index_headers(headers, sim_file.parents[1], sim_file)	2✔
35	except Exception as e: # pylint: disable=broad-exception-caught	2✔
36
37	print(f"Could not read {sim_file}: {e}")	2✔
38
39	def init_headers_multi_process(self):	2✔
40	sim_files = utils.get_files(	2✔
41	utils.get_sim_folders(self.path_to_results)
42	)
43
44	with ProcessPoolExecutor() as executor:	2✔
45	results = executor.map(_process_sim_file, sim_files)	2✔
46	for result in results:	2✔
47	if result:	2✔
48	headers, sim_folder, sim_file = result	2✔
49	self._index_headers(headers, sim_folder, sim_file)	2✔
50
51	# TODO: Discuss if something like this is needed # pylint: disable=fixme
52	def search_header(self, header_name: str): # pragma: no cover
53	if header_name in self.header_index:
54	print(f"Header '{header_name}' found in:")
55	for folder, file in self.header_index[header_name]:
56	print(f"- Folder: {folder}, File: {file}")
57	else:
58	print(f"Header '{header_name}' not found in any files.")
59
60	def _index_headers(	2✔
61	self, headers: list[str], sim_folder: _pl.Path, sim_file: _pl.Path
62	):
63	for header in headers:	2✔
64	self.header_index[header].append((sim_folder.name, sim_file.name))	2✔
65
66	# TODO: Add function to validate headers and log files with invalid headers #pylint: disable=fixme
67
68
69	class HeaderValidationMixin(ABC):	2✔
70	def validate_headers(	2✔
71	self, headers: Headers, columns: list[str]
72	) -> tuple[bool, list[str]]:
73	"""Validates that all columns exist in the headers index.
74
75	Args:
76	headers: Headers instance containing the index of available headers
77	columns: List of column names to validate
78
79	Returns:
80	Tuple of (is_valid, missing_columns)
81	- is_valid: True if all columns exist
82	- missing_columns: List of column names that are missing
83	"""
84	missing_columns = []	2✔
85	for column in columns:	2✔
86	if column not in headers.header_index:	2✔
87	missing_columns.append(column)	2✔
88
89	return len(missing_columns) == 0, missing_columns	2✔

SPF-OST / pytrnsys_process / 12763622959

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