18700174356

Committed 21 Oct 2025 11:08PM UTC coverage: 97.985% (+0.002%) from 97.983%

Build # 18700174356

Build Type

Pull #819

github

Committed by

web-flow

Commit Message

Merge 784589048 into 412f40496

Pull Request Pull Request #819: Feature/525 detect empty lines

Run Details

13 of 13 new or added lines in 3 files covered. (100.0%)

1 existing line in 1 file now uncovered.

8025 of 8190 relevant lines covered (97.99%)

0.98 hits per line

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

99.14

/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
        # keep parsing while there is input or termination has not been triggered
        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
        # IF 3  BLOCKS are parsed, the rest should be ignored with a warning and print 3 lines
        if block_counter >= 3:
            joined_lines = "\n".join(input_raw_lines[0:3])
            msg = f"Unexpected input after line {current_file.lineno - 1}\n line content: {joined_lines}\n"
            warnings.warn(
                msg,
                UndefinedBlock,
                stacklevel=6,
            )
            return

        print(f"Still parsing lines...", 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✔
48	context = input_file	1✔
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	# keep parsing while there is input or termination has not been triggered
150	if len(input_raw_lines) > 0:	1✔
151	yield from flush_input()	1✔
152	block_counter += 1	1✔
153	if block_counter < 3:	1✔
154	block_type = BlockType(block_counter)	1✔
155
156	def flush_input():	1✔
157	nonlocal input_raw_lines
158	# IF 3 BLOCKS are parsed, the rest should be ignored with a warning and print 3 lines
159	if block_counter >= 3:	1✔
160	joined_lines = "\n".join(input_raw_lines[0:3])	1✔
161	msg = f"Unexpected input after line {current_file.lineno - 1}\n line content: {joined_lines}\n"	1✔
162	warnings.warn(	1✔
163	msg,
164	UndefinedBlock,
165	stacklevel=6,
166	)
167	return	1✔
168
169	print(f"Still parsing lines...", input_raw_lines)	1✔
170	start_line = current_file.lineno + 1 - len(input_raw_lines)	1✔
171	input = Input(	1✔
172	input_raw_lines,
173	block_type,
174	current_file,
175	start_line,
176	)
177	try:	1✔
178	read_input = ReadInput(	1✔
179	input_raw_lines, block_type, current_file, start_line
180	)
181	reading_queue.append((block_type, read_input.file_name, current_file.path))	1✔
182	yield None	1✔
183	except ValueError as e:	1✔
184	if isinstance(e, ParsingError):	1✔
UNCOV 185	raise e	×
186	yield input	1✔
187	continue_input = False	1✔
188	input_raw_lines = []	1✔
189
190	for line in fh:	1✔
191	line = line.expandtabs(TABSIZE)	1✔
192	line_is_comment = is_comment(line)	1✔
193	# transition to next block with blank line
194	if not line.strip():	1✔
195	yield from flush_block()	1✔
196	has_non_comments = False	1✔
197	continue	1✔
198	# if a new input
199	if (	1✔
200	line[0:BLANK_SPACE_CONTINUE].strip()
201	and not continue_input
202	and not line_is_comment
203	and has_non_comments
204	and input_raw_lines
205	):
206	yield from flush_input()	1✔
207
208	# die if it is a vertical syntax format
209	start_o_line = line[0:BLANK_SPACE_CONTINUE]	1✔
210	# eliminate comments, and inputs that use # for other syntax
211	if (	1✔
212	"#" in start_o_line
213	and not line_is_comment
214	and start_o_line.strip().startswith("#")
215	):
216	input_raw_lines.append(line.rstrip())	1✔
217	input = next(flush_input())	1✔
218	lineno = 1	1✔
219	token = sly.lex.Token()	1✔
220	token.value = "#"	1✔
221	index = line[0:BLANK_SPACE_CONTINUE].index("#")	1✔
222	err = {"message": "", "token": token, "line": lineno, "index": index}	1✔
223	raise UnsupportedFeature(	1✔
224	"Vertical Input encountered, which is not supported by Montepy",
225	input,
226	[err],
227	)
228	# cut line down to allowed length
229	old_line = line	1✔
230	line = line[:line_length]	1✔
231	if len(old_line) != len(line):	1✔
232	comment_free = old_line.split("$")[0]	1✔
233	if len(comment_free.rstrip()) > line_length and not COMMENT_FINDER.match(	1✔
234	line
235	):
236	warnings.warn(	1✔
237	f"The line number {fh.lineno} exceeded the allowed line length of: {line_length} for MCNP{mcnp_version} "
238	f'and "{comment_free[line_length -1:].rstrip()}" was removed.',
239	LineOverRunWarning,
240	)
241	# if extra length is a comment keep it long
242	else:
243	line = old_line	1✔
244	if line.endswith(" &\n"):	1✔
245	continue_input = True	1✔
246	else:
247	continue_input = False	1✔
248	has_non_comments = has_non_comments or not line_is_comment	1✔
249	input_raw_lines.append(line.rstrip())	1✔
250	yield from flush_block()	1✔
251
252	if not recursion:	1✔
253	path = os.path.dirname(fh.name)	1✔
254	while reading_queue:	1✔
255	block_type, file_name, parent = reading_queue.popleft()	1✔
256	new_wrapper = MCNP_InputFile(os.path.join(path, file_name), parent)	1✔
257	with new_wrapper.open("r") as sub_fh:	1✔
258	new_wrapper = MCNP_InputFile(file_name, parent)	1✔
259	for input in read_data(sub_fh, mcnp_version, block_type, True):	1✔
260	yield input	1✔

idaholab / MontePy / 18700174356

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