Repository: gammapy/gammapy
Pull request #5855 (2025-05-08 11:12:58)

Utility to filter a table parsing string of  logical operations
The pull request Gammapy PR #5780 introduces a new feature that allows users to filter rows in an astropy.table.Table using logical expressions. The goal is to provide a flexible and expressive way to select subsets of data based on conditions involving table columns.

This PR introduces a function to simplify this operation.
Here's what it does:
- Takes an Astropy Table and a logical expression string.
- Evaluates the expression per row using Python's native syntax
- Returns a filtered table based on the evaluated condition.

It uses AST-based evaluation to parse and evaluate only safe, supported Python expressions.
Supports:
- Logical operators: and, or
- Comparison operators: <, >, ==, !=, in, etc.
- Lists and constants
- Any column in the table

The code, tests, and most of this text are AI generated.

codecov[bot]: ## [Codecov](https://app.codecov.io/gh/gammapy/gammapy/pull/5855?dropdown=coverage&src=pr&el=h1&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=gammapy) Report
Attention: Patch coverage is `97.50000%` with `1 line` in your changes missing coverage. Please review.
> Project coverage is 94.67%. Comparing base [(`010deef`)](https://app.codecov.io/gh/gammapy/gammapy/commit/010deef9fdfd07fa3e755bacdf91fd1787113028?dropdown=coverage&el=desc&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=gammapy) to head [(`453038b`)](https://app.codecov.io/gh/gammapy/gammapy/commit/453038b5fd38546a56c0c8e13f2033122508f502?dropdown=coverage&el=desc&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=gammapy).
> Report is 19 commits behind head on main.

| [Files with missing lines](https://app.codecov.io/gh/gammapy/gammapy/pull/5855?dropdown=coverage&src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=gammapy) | Patch % | Lines |
|---|---|---|
| [gammapy/utils/scripts.py](https://app.codecov.io/gh/gammapy/gammapy/pull/5855?src=pr&el=tree&filepath=gammapy%2Futils%2Fscripts.py&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=gammapy#diff-Z2FtbWFweS91dGlscy9zY3JpcHRzLnB5) | 97.50% | [1 Missing :warning: ](https://app.codecov.io/gh/gammapy/gammapy/pull/5855?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=gammapy) |

<details><summary>Additional details and impacted files</summary>


```diff
@@           Coverage Diff           @@
##             main    #5855   +/-   ##
=======================================
  Coverage   94.67%   94.67%           
=======================================
  Files         239      239           
  Lines       37940    37980   +40     
=======================================
+ Hits        35918    35957   +39     
- Misses       2022     2023    +1     
```

</details>

[:umbrella: View full report in Codecov by Sentry](https://app.codecov.io/gh/gammapy/gammapy/pull/5855?dropdown=coverage&src=pr&el=continue&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=gammapy).   
:loudspeaker: Have feedback on the report? [Share it here](https://about.codecov.io/codecov-pr-comment-feedback/?utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=gammapy).

<details><summary> :rocket: New features to boost your workflow: </summary>

- :snowflake: [Test Analytics](https://docs.codecov.com/docs/test-analytics): Detect flaky tests, report on failures, and find test suite problems.
</details>
registerrier: This needs a rebase @QRemy 

Patch:
diff --git a/gammapy/utils/scripts.py b/gammapy/utils/scripts.py
index ac9044523..b73bdebab 100644
--- a/gammapy/utils/scripts.py
+++ b/gammapy/utils/scripts.py
@@ -1,11 +1,15 @@
 # Licensed under a 3-clause BSD style license - see LICENSE.rst
 """Utilities to create scripts and command-line tools."""
 
+
+import ast
 import codecs
+import operator
 import os.path
 import functools
 import types
 import warnings
+import numpy as np
 from base64 import urlsafe_b64encode
 from pathlib import Path
 from uuid import uuid4
@@ -274,3 +278,92 @@ def raise_import_error(module_name, is_property=False):
     """
     kind = "property" if is_property else "method"
     raise ImportError(f"The '{module_name}' module is required to use this {kind}.")
+
+
+# Mapping of AST operators to NumPy functions
+_OPERATORS = {
+    ast.And: np.logical_and,
+    ast.Or: np.logical_or,
+    ast.Eq: operator.eq,
+    ast.NotEq: operator.ne,
+    ast.Lt: operator.lt,
+    ast.LtE: operator.le,
+    ast.Gt: operator.gt,
+    ast.GtE: operator.ge,
+    ast.In: lambda a, b: np.isin(a, b),
+    ast.NotIn: lambda a, b: ~np.isin(a, b),
+}
+
+
+def logic_parser(table, expression):
+    """
+    Parse and apply a logical expression to filter rows from an Astropy Table.
+
+    This function evaluates a logical expression on each row of the input table
+    and returns a new table containing only the rows that satisfy the expression.
+    The expression can reference any column in the table by name and supports
+    logical operators (`and`, `or`), comparison operators (`<`, `>`, `==`, `!=`, `in`),
+    lists, and constants.
+
+    Parameters
+    ----------
+    table : `~astropy.table.Table`
+        The input table to filter.
+    expression : str
+        The logical expression to evaluate on each row. The expression can reference
+        any column in the table by name.
+
+    Returns
+    -------
+    table : `~astropy.table.Table`
+        A table view containing only the rows that satisfy the expression. If no rows
+        match the condition, an empty table with the same column names and data types
+        as the input table is returned.
+
+    Examples
+    --------
+    Given a table with columns 'OBS_ID' and 'EVENT_TYPE':
+
+    >>> from astropy.table import Table
+    >>> data = {'OBS_ID': [1, 2, 3, 4], 'EVENT_TYPE': ['1', '3', '4', '2']}
+    >>> table = Table(data)
+    >>> expression = '(OBS_ID < 3) and (OBS_ID > 1) and ((EVENT_TYPE in ["3", "4"]) or (EVENT_TYPE == "3"))'
+    >>> filtered_table = logic_parser(table, expression)
+    >>> print(filtered_table)
+    OBS_ID EVENT_TYPE
+    ------ ----------
+         2          3
+
+    """
+
+    def eval_node(node):
+        if isinstance(node, ast.BoolOp):
+            op_func = _OPERATORS[type(node.op)]
+            values = [eval_node(v) for v in node.values]
+            result = values[0]
+            for v in values[1:]:
+                result = op_func(result, v)
+            return result
+        elif isinstance(node, ast.Compare):
+            left = eval_node(node.left)
+            for op, comparator in zip(node.ops, node.comparators):
+                right = eval_node(comparator)
+                op_func = _OPERATORS[type(op)]
+                left = op_func(left, right)
+            return left
+        elif isinstance(node, ast.Name):
+            if node.id not in table.colnames:
+                raise KeyError(
+                    f"Column '{node.id}' not found in the table. Available columns: {table.colnames}"
+                )
+            return table[node.id]
+        elif isinstance(node, ast.Constant):
+            return node.value
+        elif isinstance(node, ast.List):
+            return [eval_node(elt) for elt in node.elts]
+        else:
+            raise ValueError(f"Unsupported expression type: {type(node)}")
+
+    expr_ast = ast.parse(expression, mode="eval")
+    mask = eval_node(expr_ast.body)
+    return table[mask]


Test patch:
diff --git a/gammapy/utils/tests/test_scripts.py b/gammapy/utils/tests/test_scripts.py
index 6551d748d..cc6043fd8 100644
--- a/gammapy/utils/tests/test_scripts.py
+++ b/gammapy/utils/tests/test_scripts.py
@@ -3,6 +3,7 @@ import warnings
 import pytest
 import yaml
 from numpy.testing import assert_allclose
+from astropy.table import Table
 from gammapy.utils.scripts import (
     get_images_paths,
     make_path,
@@ -11,6 +12,7 @@ from gammapy.utils.scripts import (
     requires_module,
     to_yaml,
     write_yaml,
+    logic_parser,
 )
 
 
@@ -107,3 +109,90 @@ def test_requires_module():
         @requires_module("nonexistent_module")
         class InvalidUsage:
             pass
+
+
+def test_logic_parser():
+    data = {"OBS_ID": [1, 2, 3, 4], "EVENT_TYPE": ["1", "3", "4", "2"]}
+    table = Table(data)
+
+    # Test 'and' operation
+    result = logic_parser(table, "(OBS_ID < 3) and (OBS_ID > 1)")
+    assert len(result) == 1
+    assert result["OBS_ID"][0] == 2
+
+    # Test 'or' operation
+    result = logic_parser(table, "(OBS_ID < 2) or (OBS_ID > 3)")
+    assert len(result) == 2
+    assert result["OBS_ID"][0] == 1
+    assert result["OBS_ID"][1] == 4
+
+    # Test '==' operation
+    result = logic_parser(table, 'EVENT_TYPE == "3"')
+    assert len(result) == 1
+    assert result["EVENT_TYPE"][0] == "3"
+
+    # Test '!=' operation
+    result = logic_parser(table, 'EVENT_TYPE != "3"')
+    assert len(result) == 3
+    assert result["EVENT_TYPE"][0] == "1"
+    assert result["EVENT_TYPE"][1] == "4"
+    assert result["EVENT_TYPE"][2] == "2"
+
+    # Test '<' operation
+    result = logic_parser(table, "OBS_ID < 3")
+    assert len(result) == 2
+    assert result["OBS_ID"][0] == 1
+    assert result["OBS_ID"][1] == 2
+
+    # Test '<=' operation
+    result = logic_parser(table, "OBS_ID <= 3")
+    assert len(result) == 3
+    assert result["OBS_ID"][0] == 1
+    assert result["OBS_ID"][1] == 2
+    assert result["OBS_ID"][2] == 3
+
+    # Test '>' operation
+    result = logic_parser(table, "OBS_ID > 2")
+    assert len(result) == 2
+    assert result["OBS_ID"][0] == 3
+    assert result["OBS_ID"][1] == 4
+
+    # Test '>=' operation
+    result = logic_parser(table, "OBS_ID >= 2")
+    assert len(result) == 3
+    assert result["OBS_ID"][0] == 2
+    assert result["OBS_ID"][1] == 3
+    assert result["OBS_ID"][2] == 4
+
+    # Test 'in' operation
+    result = logic_parser(table, 'EVENT_TYPE in ["3", "4"]')
+    assert len(result) == 2
+    assert result["EVENT_TYPE"][0] == "3"
+    assert result["EVENT_TYPE"][1] == "4"
+
+    # Test 'not in' operation
+    result = logic_parser(table, 'EVENT_TYPE not in ["3", "4"]')
+    assert len(result) == 2
+    assert result["EVENT_TYPE"][0] == "1"
+    assert result["EVENT_TYPE"][1] == "2"
+
+    # Test no match
+    result = logic_parser(table, "(OBS_ID < 3) and (OBS_ID > 2)")
+    assert len(result) == 0
+
+
+def test_logic_parser_error():
+    # Create a sample table
+    data = {"OBS_ID": [1, 2, 3, 4], "EVENT_TYPE": ["1", "3", "4", "2"]}
+    table = Table(data)
+
+    # Define an expression with a non-existent key
+    expression = "(NON_EXISTENT_KEY < 3) and (OBS_ID > 1)"
+
+    with pytest.raises(KeyError) as excinfo:
+        logic_parser(table, expression)
+
+    with pytest.raises(ValueError) as excinfo:
+        expression = "unsupported_expression()"
+        logic_parser(table, expression)
+    assert "Unsupported expression type" in str(excinfo.value)