Implementation of a Parser for Analysis Description Language

Question

I have created a script for parsing ADL and the code looks like below. Example ADL file

I am looking for suggestions to divide the code in multiple methods and make it more readable. Thanks

block_names = ("object", "region", "info", "table ")
keywords = ("define", "select", "reject", "take", "using", "sort", "weight", "histo")
table_k_v = "# val    err-     err+     xmin     xmax"


def adl_parser(adl_file='adl.adl'):
    """Returns the ADL parsed file as JSON object

    :param type: File
    :param adl_file: Input .adl from the user.
    :return : JSON object
    """
    block_data = {}
    parsing_block = False
    parsing_table = False
    json_parsed = {
        "object": [],
        "region": [],
        "info": [],
        "table": [],
        "define": [],
    }

    # Read the ADL file
    f = open(adl_file, 'r')

    for line in f:
        # Skip comment lines outside block
        if line.startswith("#"):
            continue

        # Skip empty lines w/o whitespaces or tabs
        if line in ['\n', '\r\n']:
            # Pop out the previous block when new line appears
            if parsing_block:
                json_parsed[block_type].append(block_data)
                block_data = {}
                parsing_block = False
                parsing_table = False
            else:
                continue

        line = line.strip()
        if not line:
            continue

        # Check for block types and parse the block data
        if line.startswith(block_names) and not block_data:
            block = line.split(" ", 1)
            block_type = block[0]
            block_data["name"] = block[-1]
            parsing_block = True
            continue

        # Pop out the block when new block starts without empty line
        elif line.startswith(block_names) and block_data:
            json_parsed[block_type].append(block_data)
            block_data = {}
            block_data["name"] = line.split(" ", 1)[-1]
            parsing_block = True
            parsing_table = False
            continue

        elif line.startswith("define"):
            line_data = line.split("=")
            variable_name = line_data[0].strip().split(" ")[-1]
            variable_value = line_data[1].strip()
            json_parsed["define"].append({variable_name: variable_value})
            continue

        try:
            # Parse the keyword value pair under a block
            if parsing_block:
                # Skip comment lines inside the block
                if line.startswith("#") and line != table_k_v:
                    continue

                # Parse the rows for the table block
                if block_type == "table":
                    if line == table_k_v:
                        parsing_table = True
                        block_data["data"] = []
                        continue
                    if parsing_table:
                        values = list(filter(None, line.split(" ")))
                        if len(values) == 5:
                            block_data["data"].append({
                                "val": values[0],
                                "err-": values[1],
                                "err+": values[2],
                                "xmin": values[3],
                                "xmax": values[4],
                            })
                        continue
                keyword_value = line.split(" ", 1)
                if keyword_value[0] in keywords:
                    if keyword_value[0] == "select":
                        # use list for storing `select` values
                        if block_data.get(keyword_value[0]):
                            block_data[keyword_value[0]].append(keyword_value[-1])
                        else:
                            block_data[keyword_value[0]] = [keyword_value[-1]]
                    else:
                        block_data[keyword_value[0]] = keyword_value[-1]
                elif block_type == "region" and len(keyword_value) == 1:
                    block_data["parent"] = keyword_value[-1]
                else:
                    raise ValueError('Parsing of Keyword {} is not supported.'.format(keyword_value[0]))
        except Exception:
            block_data = {}
            parsing_block = False
            parsing_table = False

    return json_parsed

Answer 1

You already have one useful code review, so I'll focus less on those details and more on parsing strategy and your question about organizing the parsing into smaller, more-focused functions.

When building a parser, don't forget error reporting. It's tempting when building a parser to sanitize the input as early as possible, for example by stripping lines or filtering out comment lines. If done crudely, however, that preliminary work can undermine error reporting because you lose basic facts like the line number where the error occurred. Here's a simple illustration of one way to avoid that. The initial processing step will convert the raw lines of text into simple data objects that also know their original (pre-sanitized) line number.

from typing import List, Optional
from dataclasses import dataclass, field

BLOCK_NAMES = ('object', 'region', 'info', 'table')
KEYWORDS = ('define', 'select', 'reject', 'take', 'using', 'sort', 'weight', 'histo')
TABLE_K_V = '# val    err-     err+     xmin     xmax'
COMMENT = '#'

@dataclass(frozen = True)
class Line:
    n: int
    text: str

def collect_lines(fh):
    # Takes an iterable (eg opened file handle).
    # Yields Line instances.
    # Filters out blank lines and comment lines (other than TABLE_K_V).
    for i, line in enumerate(fh):
        text = line.strip()
        if not text:
            continue
        elif text.startswith(TABLE_K_V):
            yield Line(i + 1, text)
        elif not text.startswith(COMMENT):
            yield Line(i + 1, text)

Parsing block by block rather than line by line. Parsing a block-oriented language line by line is tricky: on each line you have to know what kind of structure you are currently embedded in, which usually involves the management of various status variables. This kind of code tends to be complex, hard to modify, and thus bug-prone. A simpler approach is to organize the data very early into the larger units -- the blocks. Not all syntaxes are amenable to this kind of approach, but I think ADL is. Here's an illustration that uses a Block data object, which knows its kind and has a list of Line instances.

@dataclass
class Block:
    kind: Optional[str] = None
    lines: List[Line] = field(default_factory = list)

def collect_blocks(lines):
    # Takes an iterable of Line instances.
    # Yields Block instances.
    b = Block()
    for line in lines:
        first_word = line.text.split()[0]
        if first_word in BLOCK_NAMES:
            if b.lines:
                yield b
            b = Block(kind = first_word)
        b.lines.append(line)
    if b.lines:
        yield b

def main():
    # Usage demo.
    path = sys.argv[1]
    with open(path) as fh:
        for b in collect_blocks(collect_lines(fh)):
            print(b.kind, len(b.lines))

if __name__ == '__main__':
    main()

Next steps. Once you have a way to work with entire blocks at a time, you just need to build a parsing function for each kind of block. And those functions might use smaller functions that focus on the next level down in the hierarchy: the entities declared by KEYWORDS.

Answer 2

Global Constants

block_names = ("object", "region", "info", "table ")
keywords = ("define", "select", "reject", "take", "using", "sort", "weight", "histo")
table_k_v = "# val    err-     err+     xmin     xmax"

These should be defined in UPPER_CASE, to indicate they are constants.

Suspicious value

In block_names = ("object", "region", "info", "table "), the final value has a trailing space, none of the other values do. This is suspicious.

Reading the code, I see the test line.startswith(block_names), and reviewing the sample file, I see the word tabletype, so I'm guessing you've used "table " with the trailing space to avoid accidentally matching tabletype. However, this is confusing and fragile. A new term like "regional" could accidentally match, and a change in white space, such as using a tab character (\t) instead of a space would prevent the correct match in "table\tbtagdeepCSVmedium.

Close resources

When opening resources, use a with statement, to ensure the resource is closed. You never explicitly call f.close() in your code, so the resource may be held open for much longer than you intend. However, you should prefer using the with-syntax to avoid the need to call f.close() explicitly.

    # Read the ADL file
    with open(adl_file, 'r') as f:
        ...

Multiple ways to parse blank lines

        if line in ['\n', '\r\n']:
            if parsing_block:
                json_parsed[block_type].append(block_data)
                block_data = {}
                parsing_block = False
                parsing_table = False
            else:
                continue

        line = line.strip()
        if not line:
            continue

If a blank line, then if in a parsing block you do some work and then proceed to the line = line.strip(). If not in a parsing block, you continue the loop. After the blank line check, you strip white space from the start/end, and if that results in a blank line, you continue without checking if you are in a parsing block.

Uh ... so two line which are visually the same are handled differently if they contain spaces or tabs?

Plus you later have to handle "when new block starts without an empty line" anyway.

It would be simpler to strip whitespace, and then check for empty/blank lines, and rely on the existing "when new block starts without an empty line" code:

        line = line.strip()
        if not line:
            continue

Comments

        if line.startswith("#"):
            continue

This happens before stripping off leading white-space, so indented comments would still be processed.

From the ADL definition you linked to, comments can also appear at end of lines:

keyword3 value3 # comment about value3

As long as you can't have any # character embedded inside quoted string, it would simplify later parsing if these were removed. Since removing the comment can turn a non-blank line into a blank line, let's strip off comments first, then check for blank lines:

        if '#' in line:
            line = line[:line.index('#')]

        line = line.strip()
        if not line:
            continue

Splitting on Spaces

                        values = list(filter(None, line.split(" ")))

It took me a while to figure out what you were trying to do here, but I finally got it. You are splitting a line on the space character, which is giving you a bunch of empty strings if terms are separated by multiple spaces. The filter(None, ...) produces a generator that removes falsy empty strings, which you then turn into a list.

This does practically the same thing:

                        values = line.split()

It is better, actually, since it will treat any combination of one or more white-space characters as one separator, allowing the tables to use tabs to line things up nicely.

Last element of list

This bugged me ... keyword_value[-1] ... until I realized what you were doing with .split(" "). Using .split() fixes that issue, and you can safely use keyword_value[1] to reference the second item in the line, even if multiple spaces exist between the first and second items.

Exception Handling

    for line in f:
        ...

        try:
            ...
                else:
                    raise ValueError('Parsing of Keyword {} is not supported.'.format(keyword_value[0]))
        except Exception:
            block_data = {}
            parsing_block = False
            parsing_table = False

Catching any Exception is bad. If you make an error, like using keyword_val instead of keyword_value, or using a bad list index, the exception (NameError, IndexError) is caught, and your code will blank out the current parsing and muddle-on without ever telling you what happened. Yikes!

You are explicitly raising a ValueError. You should explicitly catch that, and only that.

However, you've only got one raise statement, which is followed immediately by the catch clause. You could eliminate the entire try...except, and move the cleanup code into the else: statement.

You might want to print out the unrecognized keywords too. You might think you've coded for them all, but you might not have.

Indentation

The ADL syntax is composed of blocks with indented lines. Using line = line.strip() removes all leading and trailing white-space, destroying the block-level formatting. You are left with relying on region, table, etc as being a block level keywords, with the hope that in the future a region will never have a table keyword inside it, or vis-versa.

Instead:

    for line in f:

        # Remove comments
        if '#' in line:
            line = line[:line.index('#')]

        if not line or line.isblank():
            continue    # White-space only line
        elif line.startswith((' ', '\t')):
            ...         # Indented line: should be inside a block
        else:
            ...         # Outdented line: should be start of a block

End-of-block processing

When you detect the end of a block, or start a new block with an uncommitted block, you append the previous block and clear the collection area ...

            json_parsed[block_type].append(block_data)
            block_data = {}
            parsing_block = False

When you start a new block, you remember the block_type, and set the parsing_block to True.

You could make block_type do double duty. Set block_type to None when you are not parsing a block, and you could remove the parsing_block flag.

A possibly better approach could be to immediately store the block in the output structure when you start a block ...

        if line.startswith(block_names):    # Based on current code
            block = line.split(maxsplit=1)
            block_data = {"name": block[1]}
            json_parsed[block[0]].append(block_data)

Then, when it is time to do the end-of-block processing, you're already done! Though block_data was initially empty when it was added, you fill in the data at subsequent steps like you are already doing.

Using block_data = None when you finished one block and haven't started the next may be useful, as it will prevent accidental corruption of your completed block.

This has eliminated both block_type and parsing_block variables.

table_k_v

This is very fragile code:

table_k_v = "# val    err-     err+     xmin     xmax"
...
                if block_type == "table":
                    if line == table_k_v:
                        parsing_table = True
                        ...
                    ...

If the table layout changes ever so slightly, perhaps adding another significant figure to the data, the comment may be adjusted without a second thought, and your code with fail to detect the table.

A slightly better approach might be to split the line on spaces, and compare that to a pre-split constant:

TABLE_K_V = "#  val  err-  err+  xmin  xmax".split()
...
                if block_type == "table":
                    if line.split() == TABLE_K_V:
                        parsing_table = True
                        ...
                    ...

You could throw in a .casefold() in case someone decides # Val Err- Err+ XMin XMax looks better in the comment line.

However, this is still fragile. It is only a comment line, and other modifications like err- to -Err would be very difficult to account for. Since we cannot rely on it, removing the comments completely (as shown above) makes sense.

In the table section, if the line consists of 5 numbers, it is a table row. If the line consists of an identifier and a value, it is a key-value pair. If it is anything else, it is likely an error.

Python 3.10

Assuming you're up to the latest version of Python, there is a new feature: Structured Pattern Matching. With it, you can match compound structures.

For example, parsing of the top-level blocks could look like:

        # Top-level block (non-indented line)

        words = line.split(maxsplits=3)
        match words:

            case ['table', name]:
                block_data = {"name": name}
                json_parsed['table'].append(block_data)
                # remainder of table parsing ...

            case ['define', name, '=', definition]:
                json_parsed['define'].append({name: definition})

            case [('object' | 'obj'), name]:
                block_data = {"name": name}
                json_parsed['object'].append(block_data)
                # remainder of object parsing ...

            case [('object' | 'obj'), name, ':', type_alias]:
                raise NotImplementedError("Type aliases don't seem supported, yet")

            case [('info' | 'region') as block_type, name]:
                block_data = {"name": name}
                json_parsed[block_type].append(block_data)
                # remainder of block parsing ...

            case _:
                print("Warning: Unrecognized block:", line)

Note the use of Or-patterns to treat both "object" and "obj" as object definitions, per the ADL definition.

Also, from the sample file, there is the block:

object vetoleptons : Union(vetoelectrons, vetomuons)

which does not appear to be properly supported in your code yet.

Answer 3

After suggestions from @AJNeufeld and @FMc, It helped me a lot to understand and make changes in the code to increase the readability. The code looks more pythonic now, I would say.

Here's the final code:

from typing import List, Optional
from dataclasses import dataclass, field
import re


TABLE_HEADER = '# val    err-     err+     xmin     xmax'


@dataclass(frozen=True)
class Line:
    text: str


@dataclass
class Block:
    block_type: Optional[str] = None  # noqa
    lines: List[Line] = field(default_factory=list)  # noqa


@dataclass
class Keyword:
    block_type: Optional[str] = None  # noqa
    line: Line = None  # noqa


def _collect_lines(file_handler):
    """Return a line generator for the block/keyword."""
    _COMMENT = '#'
    for line in file_handler:
        line = line.decode('utf-8')
        text = line.strip()
        if not text:
            continue
        elif text.startswith(TABLE_HEADER):
            yield Line(text)
        elif not text.startswith(_COMMENT):
            # Skip comment lines
            yield Line(text)


def _collect_blocks(lines):
    """Return a block/keyword generator."""
    _block_object = Block()
    for line in lines:
        _block_type = line.text.split()[0]
        if _regex_block_match(_block_type):
            if _block_object.lines:
                yield _block_object
            _block_object = Block(block_type=_block_type)
        if _regex_keyword_match(_block_type, define=True):
            if _block_object.lines:
                yield _block_object
            _keyword_object = Keyword(block_type=_block_type, line=line)
            yield _keyword_object
        if _block_object:
            _block_object.lines.append(line)
    if _block_object and _block_object.lines:
        yield _block_object


def _regex_block_match(_block_type):
    _block_type_regexp = {
        'object': re.compile(r'object'),
        'region': re.compile(r'region'),
        'info': re.compile(r'info'),
        'table': re.compile(r'table'),
    }
    for _block in _block_type_regexp:
        pattern = _block_type_regexp[_block]
        if pattern.fullmatch(_block_type):
            return True
    return False


def _regex_keyword_match(_keyword_type, define=False):
    _keyword_type_regexp = {
        'define': re.compile(r'define'),
        'select': re.compile(r'select'),
        'reject': re.compile(r'reject'),
        'take': re.compile(r'take'),
        'using': re.compile(r'using'),
        'sort': re.compile(r'sort'),
        'weight': re.compile(r'weight'),
        'histo': re.compile(r'histo'),
    }
    if define:
        pattern = _keyword_type_regexp['define']
        return pattern.fullmatch(_keyword_type)
    for _keyword in _keyword_type_regexp:
        pattern = _keyword_type_regexp[_keyword]
        if pattern.fullmatch(_keyword_type):
            return True
    return False


def _get_key_value(line):
    # Split the keywords and values
    line_data = line.text.split(' ', 1)
    if _regex_keyword_match(line_data[0]) or _regex_block_match(line_data[0]):
        return line_data
    return None


def _handle_select_keyword(select_value, _block_data):
    if _block_data.get('select'):
        _block_data['select'].append(select_value)
    else:
        _block_data['select'] = [select_value]
    return _block_data

# Project specific
def _adl_logger(_type, message):
    if _type == 'info':
        return current_app.logger.info(message)
    if _type == 'error':
        return current_app.logger.error(message)


def _object_block_parser(block):
    block_data = {}
    for line in block.lines:
        line_key_value = _get_key_value(line)
        if not line_key_value:
            _adl_logger(
                _type='info',
                message='Skipping invalid keyword value pair {}'.format(line))
            continue
        if line_key_value[0] == 'select':
            block_data = _handle_select_keyword(
                line_key_value[-1], block_data)
        else:
            block_data.update({
                line_key_value[0]: line_key_value[-1]
            })
    return block_data


def _region_block_parser(block):
    block_data = {}
    for line in block.lines:
        line_key_value = _get_key_value(line)
        if not line_key_value:
            _adl_logger(
                _type='info',
                message='Skipping invalid keyword value pair {}'.format(line))
            continue
        if len(line_key_value) == 1:
            block_data['parent'] = line_key_value[-1]
            continue
        if line_key_value[0] == 'select':
            block_data = _handle_select_keyword(
                line_key_value[-1], block_data)
        else:
            block_data.update({
                line_key_value[0]: line_key_value[-1]
            })
    return block_data


def _info_block_parser(block):
    raise NotImplementedError


def _table_block_parser(block):
    block_data = {
        'data': []
    }
    for line in block.lines:
        if line.text == TABLE_HEADER:
            # Todo: Make check more general since header can be different.
            continue
        values = line.text.split()
        if len(values) == 5:
            block_data['data'].append({
                'val': values[0],
                'err-': values[1],
                'err+': values[2],
                'xmin': values[3],
                'xmax': values[4],
            })
        else:
            line_key_value = _get_key_value(line)
            if not line_key_value:
                _adl_logger(
                    _type='info',
                    message='Skipping invalid keyword value pair {}'.format(
                        line))
                continue
            block_data.update({
                line_key_value[0]: line_key_value[-1]
            })
    return block_data


def _define_block_parser(block):
    block_data = {}
    line_data = block.line.text.split("=")
    _define_key = line_data[0].strip().split(" ")[-1]
    _define_value = line_data[1].strip()
    block_data.update({
        _define_key: _define_value
    })
    return block_data


def _adl_parser(adl_file=None):
    """Returns the ADL parsed file as JSON object

    :param type: File
    :param adl_file: Input .adl from the user.
    :return : JSON object
    """
    block_data = []
    json_parsed = {
        'objects': [],
        'regions': [],
        'infos': [],
        'tables': [],
        'define': [],
    }

    with open(adl_file, 'r') as file_handler:
        for b in _collect_blocks(_collect_lines(file_handler)):
            block_data.append(b)

    for block in block_data:
        if block.block_type == 'object':
            json_parsed['objects'].append(_object_block_parser(block))
        if block.block_type == 'region':
            json_parsed['regions'].append(_region_block_parser(block))
        if block.block_type == 'info':
            json_parsed['infos'].append(_info_block_parser(block))
        if block.block_type == 'table':
            json_parsed['tables'].append(_table_block_parser(block))
        if block.block_type == 'define':
            json_parsed['define'].append(_define_block_parser(block))

    return json_parsed