convert hex to int in calculator (#463)

Co-authored-by: dtrai2 <[email protected]>

CHANGELOG.md

@@ -3,6 +3,9 @@
 ## next release
 
 ### Features
+
+* add possibility to convert hex to int in `calculator` processor with new added function `from_hex`
+
 ### Improvements
 ### Bugfix
 

logprep/processor/calculator/fourFn.py

@@ -46,6 +46,7 @@
     "exp": math.exp,
     "abs": abs,
     "trunc": int,
+    "from_hex": lambda a: int(a, 16),
     "round": round,
     "sgn": lambda a: -1 if a < -epsilon else 1 if a > epsilon else 0,
     # functionsl with multiple arguments
@@ -80,7 +81,7 @@ class BNF(Forward):
     #                    Optional(e + Word("+-"+nums, nums)))
     # or use provided pyparsing_common.number, but convert back to str:
     # fnumber = ppc.number().addParseAction(lambda t: str(t[0]))
-    fnumber = Regex(r"[+-]?\d+(?:\.\d*)?(?:[eE][+-]?\d+)?")
+    fnumber = Regex(r"[+-]?[a-zA-Z0-9]+(?:\.\d*)?(?:[eE][+-]?\d+)?")
     ident = Word(alphas, alphanums + "_$")
 
     plus, minus, mult, div = map(Literal, "+-*/")
@@ -119,12 +120,15 @@ def evaluate_stack(self):
             args = reversed([self.evaluate_stack() for _ in range(num_args)])
             return fn[op](*args)
         if op[0].isalpha():
-            raise Exception(f"invalid identifier '{op}'")
+            raise Exception(f"invalid identifier '{op}'")  # pylint: disable=broad-exception-raised
         # try to evaluate as int first, then as float if int fails
         try:
             return int(op)
         except ValueError:
-            return float(op)
+            try:
+                return float(op)
+            except ValueError:
+                return op
 
     def __new__(cls):
         if not hasattr(cls, "instance"):
@@ -134,7 +138,7 @@ def __new__(cls):
     def __init__(self) -> None:
         super().__init__()
         self.exprStack = []
-        expr_list = delimitedList(Group(self))
+        expr_list = delimitedList(Group(self))  # pylint: disable=E1121
 
         # add parse action that replaces the function identifier with a (name, number of args) tuple
         def insert_fn_argcount_tuple(t):

tests/unit/processor/calculator/test_calculator.py

@@ -171,6 +171,42 @@
         {"duration": "0.01"},
         {"duration": "0.01"},
     ),
+    (
+        "convert hex to int",
+        {
+            "filter": "message",
+            "calculator": {
+                "calc": "from_hex(0x${field1})",
+                "target_field": "new_field",
+            },
+        },
+        {"message": "This is a message", "field1": "ff"},
+        {"message": "This is a message", "field1": "ff", "new_field": 255},
+    ),
+    (
+        "convert hex to int with prefix",
+        {
+            "filter": "message",
+            "calculator": {
+                "calc": "from_hex(${field1})",
+                "target_field": "new_field",
+            },
+        },
+        {"message": "This is a message", "field1": "0xff"},
+        {"message": "This is a message", "field1": "0xff", "new_field": 255},
+    ),
+    (
+        "convert hex to int with prefix",
+        {
+            "filter": "message",
+            "calculator": {
+                "calc": "from_hex(0x${field1})",
+                "target_field": "new_field",
+            },
+        },
+        {"message": "This is a message", "field1": "FF"},
+        {"message": "This is a message", "field1": "FF", "new_field": 255},
+    ),
 ]
 
 
@@ -350,6 +386,7 @@ def test_testcases_failure_handling(
             ("10+sin(PI/4)^2", 10 + math.sin(math.pi / 4) ** 2),
             ("trunc(E)", int(math.e)),
             ("trunc(-E)", int(-math.e)),
+            ("from_hex(4B)", 75),
             ("round(E)", round(math.e)),
             ("round(-E)", round(-math.e)),
             ("E^PI", math.e**math.pi),
@@ -378,6 +415,6 @@ def test_testcases_failure_handling(
     )
     def test_fourfn(self, expression, expected):
         bnf = BNF()
-        _ = bnf.parseString(expression, parseAll=True)
+        _ = bnf.parseString(expression, parseAll=True)  # pylint: disable=E1123,E1121
         result = bnf.evaluate_stack()
         assert result == expected