18699607926

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

Build # 18699607926

Build Type

Pull #819

github

Committed by

web-flow

Commit Message

Merge 721e90ab6 into 412f40496

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

Run Details

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

idaholab / MontePy / 18699607926

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