18820526907

Committed 26 Oct 2025 04:16PM UTC coverage: 89.323% (-7.4%) from 96.71%

Build # 18820526907

Build Type

push

github

Committed by

MicahGale

Commit Message

Fixed how class info is collected for iterables.

Run Details

2 of 2 new or added lines in 1 file covered. (100.0%)

637 existing lines in 30 files now uncovered.

7479 of 8373 relevant lines covered (89.32%)

0.89 hits per line

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

98.18

/montepy/input_parser/input_syntax_reader.py

# Copyright 2024-2025, Battelle Energy Alliance, LLC All Rights Reserved.
from collections import deque
import itertools
import io
import re
import os
import sly
import warnings

from montepy.constants import *
from montepy.exceptions import *
from montepy.input_parser.block_type import BlockType
from montepy.input_parser.input_file import MCNP_InputFile
from montepy.input_parser.mcnp_input import Input, Message, ReadInput, Title
from montepy.input_parser.read_parser import ReadParser
from montepy.utilities import is_comment


reading_queue = deque()


def read_input_syntax(input_file, mcnp_version=DEFAULT_VERSION, replace=True):
    """Creates a generator function to return a new MCNP input for
    every new one that is encountered.

    This is meant to just handle the MCNP input syntax, it does not
    semantically parse the inputs.

    The version must be a three component tuple e.g., (6, 2, 0) and (5, 1, 60).

    Parameters
    ----------
    input_file : MCNP_InputFile
        the path to the input file to be read
    mcnp_version : tuple
        The version of MCNP that the input is intended for.
    replace : bool
        replace all non-ASCII characters with a space (0x20)

    Returns
    -------
    generator
        a generator of MCNP_Object objects
    """
    global reading_queue
    reading_queue = deque()
    if input_file.is_stream:
        context = input_file
    else:
        context = input_file.open("r", replace=replace)
    with context as fh:
        yield from read_front_matters(fh, mcnp_version)
        yield from read_data(fh, mcnp_version)


def read_front_matters(fh, mcnp_version):
    """Reads the beginning of an MCNP file for all of the unusual data there.

    This is a generator function that will yield multiple :class:`MCNP_Input` instances.

    Warnings
    --------
    This function will move the file handle forward in state.

    Warnings
    --------
    This function will not close the file handle.

    Parameters
    ----------
    fh : MCNP_InputFile
        The file handle of the input file.
    mcnp_version : tuple
        The version of MCNP that the input is intended for.

    Returns
    -------
    MCNP_Object
        an instance of the Title class, and possible an instance of a
        Message class
    """
    is_in_message_block = False
    found_title = False
    lines = []
    raw_lines = []
    for i, line in enumerate(fh):
        if i == 0 and line.upper().startswith("MESSAGE:"):
            is_in_message_block = True
            raw_lines.append(line.rstrip())
            lines.append(line[9:])  # removes "MESSAGE: "
        elif is_in_message_block:
            if line.strip():
                raw_lines.append(line.rstrip())
                lines.append(line)
            # message block is terminated by a blank line
            else:
                yield Message(raw_lines, lines)
                is_in_message_block = False
        # title always follows complete message, or is first
        else:
            yield Title([line], line)
            break


def read_data(fh, mcnp_version, block_type=None, recursion=False):
    """Reads the bulk of an MCNP file for all of the MCNP data.

    This is a generator function that will yield multiple :class:`MCNP_Input` instances.

    Warnings
    --------
    This function will move the file handle forward in state.

    Warnings
    --------
    This function will not close the file handle.

    Parameters
    ----------
    fh : MCNP_InputFile
        The file handle of the input file.
    mcnp_version : tuple
        The version of MCNP that the input is intended for.
    block_type : BlockType
        The type of block this file is in. This is only used with
        partial files read using the ReadInput.
    recursion : bool
        Whether or not this is being called recursively. If True this
        has been called from read_data. This prevents the reading queue
        causing infinite recursion.

    Returns
    -------
    MCNP_Input
        MCNP_Input instances: Inputs that represent the data in the MCNP
        input.
    """
    current_file = fh
    line_length = get_max_line_length(mcnp_version)
    block_counter = 0
    if block_type is None:
        block_type = BlockType.CELL
    continue_input = False
    has_non_comments = False
    input_raw_lines = []

    def flush_block():
        nonlocal block_counter, block_type
        if len(input_raw_lines) > 0:
            yield from flush_input()
        block_counter += 1
        if block_counter < 3:
            block_type = BlockType(block_counter)

    def flush_input():
        nonlocal input_raw_lines
        start_line = current_file.lineno + 1 - len(input_raw_lines)
        input = Input(
            input_raw_lines,
            block_type,
            current_file,
            start_line,
        )
        try:
            read_input = ReadInput(
                input_raw_lines, block_type, current_file, start_line
            )
            reading_queue.append((block_type, read_input.file_name, current_file.path))
            yield None
        except ValueError as e:
            if isinstance(e, ParsingError):
                raise e
            yield input
        continue_input = False
        input_raw_lines = []

    for line in fh:
        line = line.expandtabs(TABSIZE)
        line_is_comment = is_comment(line)
        # transition to next block with blank line
        if not line.strip():
            yield from flush_block()
            has_non_comments = False
            continue
        # if a new input
        if (
            line[0:BLANK_SPACE_CONTINUE].strip()
            and not continue_input
            and not line_is_comment
            and has_non_comments
            and input_raw_lines
        ):
            yield from flush_input()
        # die if it is a vertical syntax format
        start_o_line = line[0:BLANK_SPACE_CONTINUE]
        # eliminate comments, and inputs that use # for other syntax
        if (
            "#" in start_o_line
            and not line_is_comment
            and start_o_line.strip().startswith("#")
        ):
            input_raw_lines.append(line.rstrip())
            input = next(flush_input())
            lineno = 1
            token = sly.lex.Token()
            token.value = "#"
            index = line[0:BLANK_SPACE_CONTINUE].index("#")
            err = {"message": "", "token": token, "line": lineno, "index": index}
            raise UnsupportedFeature(
                "Vertical Input encountered, which is not supported by Montepy",
                input,
                [err],
            )
        # cut line down to allowed length
        old_line = line
        line = line[:line_length]
        if len(old_line) != len(line):
            comment_free = old_line.split("$")[0]
            if len(comment_free.rstrip()) > line_length and not COMMENT_FINDER.match(
                line
            ):
                warnings.warn(
                    f"The line number {fh.lineno} exceeded the allowed line length of: {line_length} for MCNP{mcnp_version} "
                    f'and "{comment_free[line_length -1:].rstrip()}" was removed.',
                    LineOverRunWarning,
                )
            # if extra length is a comment keep it long
            else:
                line = old_line
        if line.endswith(" &\n"):
            continue_input = True
        else:
            continue_input = False
        has_non_comments = has_non_comments or not line_is_comment
        input_raw_lines.append(line.rstrip())
    yield from flush_block()

    if not recursion:
        path = os.path.dirname(fh.name)
        while reading_queue:
            block_type, file_name, parent = reading_queue.popleft()
            new_wrapper = MCNP_InputFile(os.path.join(path, file_name), parent)
            with new_wrapper.open("r") as sub_fh:
                new_wrapper = MCNP_InputFile(file_name, parent)
                for input in read_data(sub_fh, mcnp_version, block_type, True):
                    yield input

1	# Copyright 2024-2025, Battelle Energy Alliance, LLC All Rights Reserved.
2	from collections import deque	1✔
3	import itertools	1✔
4	import io	1✔
5	import re	1✔
6	import os	1✔
7	import sly	1✔
8	import warnings	1✔
9
10	from montepy.constants import *	1✔
11	from montepy.exceptions import *	1✔
12	from montepy.input_parser.block_type import BlockType	1✔
13	from montepy.input_parser.input_file import MCNP_InputFile	1✔
14	from montepy.input_parser.mcnp_input import Input, Message, ReadInput, Title	1✔
15	from montepy.input_parser.read_parser import ReadParser	1✔
16	from montepy.utilities import is_comment	1✔
17
18
19	reading_queue = deque()	1✔
20
21
22	def read_input_syntax(input_file, mcnp_version=DEFAULT_VERSION, replace=True):	1✔
23	"""Creates a generator function to return a new MCNP input for
24	every new one that is encountered.
25
26	This is meant to just handle the MCNP input syntax, it does not
27	semantically parse the inputs.
28
29	The version must be a three component tuple e.g., (6, 2, 0) and (5, 1, 60).
30
31	Parameters
32	----------
33	input_file : MCNP_InputFile
34	the path to the input file to be read
35	mcnp_version : tuple
36	The version of MCNP that the input is intended for.
37	replace : bool
38	replace all non-ASCII characters with a space (0x20)
39
40	Returns
41	-------
42	generator
43	a generator of MCNP_Object objects
44	"""
45	global reading_queue
46	reading_queue = deque()	1✔
47	if input_file.is_stream:	1✔
UNCOV 48	context = input_file	×
49	else:
50	context = input_file.open("r", replace=replace)	1✔
51	with context as fh:	1✔
52	yield from read_front_matters(fh, mcnp_version)	1✔
53	yield from read_data(fh, mcnp_version)	1✔
54
55
56	def read_front_matters(fh, mcnp_version):	1✔
57	"""Reads the beginning of an MCNP file for all of the unusual data there.
58
59	This is a generator function that will yield multiple :class:`MCNP_Input` instances.
60
61	Warnings
62	--------
63	This function will move the file handle forward in state.
64
65	Warnings
66	--------
67	This function will not close the file handle.
68
69	Parameters
70	----------
71	fh : MCNP_InputFile
72	The file handle of the input file.
73	mcnp_version : tuple
74	The version of MCNP that the input is intended for.
75
76	Returns
77	-------
78	MCNP_Object
79	an instance of the Title class, and possible an instance of a
80	Message class
81	"""
82	is_in_message_block = False	1✔
83	found_title = False	1✔
84	lines = []	1✔
85	raw_lines = []	1✔
86	for i, line in enumerate(fh):	1✔
87	if i == 0 and line.upper().startswith("MESSAGE:"):	1✔
88	is_in_message_block = True	1✔
89	raw_lines.append(line.rstrip())	1✔
90	lines.append(line[9:]) # removes "MESSAGE: "	1✔
91	elif is_in_message_block:	1✔
92	if line.strip():	1✔
93	raw_lines.append(line.rstrip())	1✔
94	lines.append(line)	1✔
95	# message block is terminated by a blank line
96	else:
97	yield Message(raw_lines, lines)	1✔
98	is_in_message_block = False	1✔
99	# title always follows complete message, or is first
100	else:
101	yield Title([line], line)	1✔
102	break	1✔
103
104
105	def read_data(fh, mcnp_version, block_type=None, recursion=False):	1✔
106	"""Reads the bulk of an MCNP file for all of the MCNP data.
107
108	This is a generator function that will yield multiple :class:`MCNP_Input` instances.
109
110	Warnings
111	--------
112	This function will move the file handle forward in state.
113
114	Warnings
115	--------
116	This function will not close the file handle.
117
118	Parameters
119	----------
120	fh : MCNP_InputFile
121	The file handle of the input file.
122	mcnp_version : tuple
123	The version of MCNP that the input is intended for.
124	block_type : BlockType
125	The type of block this file is in. This is only used with
126	partial files read using the ReadInput.
127	recursion : bool
128	Whether or not this is being called recursively. If True this
129	has been called from read_data. This prevents the reading queue
130	causing infinite recursion.
131
132	Returns
133	-------
134	MCNP_Input
135	MCNP_Input instances: Inputs that represent the data in the MCNP
136	input.
137	"""
138	current_file = fh	1✔
139	line_length = get_max_line_length(mcnp_version)	1✔
140	block_counter = 0	1✔
141	if block_type is None:	1✔
142	block_type = BlockType.CELL	1✔
143	continue_input = False	1✔
144	has_non_comments = False	1✔
145	input_raw_lines = []	1✔
146
147	def flush_block():	1✔
148	nonlocal block_counter, block_type
149	if len(input_raw_lines) > 0:	1✔
150	yield from flush_input()	1✔
151	block_counter += 1	1✔
152	if block_counter < 3:	1✔
153	block_type = BlockType(block_counter)	1✔
154
155	def flush_input():	1✔
156	nonlocal input_raw_lines
157	start_line = current_file.lineno + 1 - len(input_raw_lines)	1✔
158	input = Input(	1✔
159	input_raw_lines,
160	block_type,
161	current_file,
162	start_line,
163	)
164	try:	1✔
165	read_input = ReadInput(	1✔
166	input_raw_lines, block_type, current_file, start_line
167	)
168	reading_queue.append((block_type, read_input.file_name, current_file.path))	1✔
169	yield None	1✔
170	except ValueError as e:	1✔
171	if isinstance(e, ParsingError):	1✔
172	raise e	×
173	yield input	1✔
174	continue_input = False	1✔
175	input_raw_lines = []	1✔
176
177	for line in fh:	1✔
178	line = line.expandtabs(TABSIZE)	1✔
179	line_is_comment = is_comment(line)	1✔
180	# transition to next block with blank line
181	if not line.strip():	1✔
182	yield from flush_block()	1✔
183	has_non_comments = False	1✔
184	continue	1✔
185	# if a new input
186	if (	1✔
187	line[0:BLANK_SPACE_CONTINUE].strip()
188	and not continue_input
189	and not line_is_comment
190	and has_non_comments
191	and input_raw_lines
192	):
193	yield from flush_input()	1✔
194	# die if it is a vertical syntax format
195	start_o_line = line[0:BLANK_SPACE_CONTINUE]	1✔
196	# eliminate comments, and inputs that use # for other syntax
197	if (	1✔
198	"#" in start_o_line
199	and not line_is_comment
200	and start_o_line.strip().startswith("#")
201	):
202	input_raw_lines.append(line.rstrip())	1✔
203	input = next(flush_input())	1✔
204	lineno = 1	1✔
205	token = sly.lex.Token()	1✔
206	token.value = "#"	1✔
207	index = line[0:BLANK_SPACE_CONTINUE].index("#")	1✔
208	err = {"message": "", "token": token, "line": lineno, "index": index}	1✔
209	raise UnsupportedFeature(	1✔
210	"Vertical Input encountered, which is not supported by Montepy",
211	input,
212	[err],
213	)
214	# cut line down to allowed length
215	old_line = line	1✔
216	line = line[:line_length]	1✔
217	if len(old_line) != len(line):	1✔
218	comment_free = old_line.split("$")[0]	1✔
219	if len(comment_free.rstrip()) > line_length and not COMMENT_FINDER.match(	1✔
220	line
221	):
222	warnings.warn(	1✔
223	f"The line number {fh.lineno} exceeded the allowed line length of: {line_length} for MCNP{mcnp_version} "
224	f'and "{comment_free[line_length -1:].rstrip()}" was removed.',
225	LineOverRunWarning,
226	)
227	# if extra length is a comment keep it long
228	else:
229	line = old_line	1✔
230	if line.endswith(" &\n"):	1✔
231	continue_input = True	1✔
232	else:
233	continue_input = False	1✔
234	has_non_comments = has_non_comments or not line_is_comment	1✔
235	input_raw_lines.append(line.rstrip())	1✔
236	yield from flush_block()	1✔
237
238	if not recursion:	1✔
239	path = os.path.dirname(fh.name)	1✔
240	while reading_queue:	1✔
241	block_type, file_name, parent = reading_queue.popleft()	1✔
242	new_wrapper = MCNP_InputFile(os.path.join(path, file_name), parent)	1✔
243	with new_wrapper.open("r") as sub_fh:	1✔
244	new_wrapper = MCNP_InputFile(file_name, parent)	1✔
245	for input in read_data(sub_fh, mcnp_version, block_type, True):	1✔
246	yield input	1✔

idaholab / MontePy / 18820526907

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