perf jevents metric: Add ability to rewrite metrics in terms of others

import decimal

import json

import re

from typing import Dict, List, Optional, Set, Union

from typing import Dict, List, Optional, Set, Tuple, Union

class Expression:

    """Returns true when two expressions are the same."""

    raise NotImplementedError()

  def Substitute(self, name: str, expression: 'Expression') -> 'Expression':

    raise NotImplementedError()

  def __str__(self) -> str:

    return self.ToPerfJson()

          other.lhs) and self.rhs.Equals(other.rhs)

    return False

  def Substitute(self, name: str, expression: Expression) -> Expression:

    if self.Equals(expression):

      return Event(name)

    lhs = self.lhs.Substitute(name, expression)

    rhs = None

    if self.rhs:

      rhs = self.rhs.Substitute(name, expression)

    return Operator(self.operator, lhs, rhs)

class Select(Expression):

  """Represents a select ternary in the parse tree."""

          other.false_val) and self.true_val.Equals(other.true_val)

    return False

  def Substitute(self, name: str, expression: Expression) -> Expression:

    if self.Equals(expression):

      return Event(name)

    true_val = self.true_val.Substitute(name, expression)

    cond = self.cond.Substitute(name, expression)

    false_val = self.false_val.Substitute(name, expression)

    return Select(true_val, cond, false_val)

class Function(Expression):

  """A function in an expression like min, max, d_ratio."""

      return result

    return False

  def Substitute(self, name: str, expression: Expression) -> Expression:

    if self.Equals(expression):

      return Event(name)

    lhs = self.lhs.Substitute(name, expression)

    rhs = None

    if self.rhs:

      rhs = self.rhs.Substitute(name, expression)

    return Function(self.fn, lhs, rhs)

def _FixEscapes(s: str) -> str:

  s = re.sub(r'([^\\]),', r'\1\\,', s)

  def Equals(self, other: Expression) -> bool:

    return isinstance(other, Event) and self.name == other.name

  def Substitute(self, name: str, expression: Expression) -> Expression:

    return self

class Constant(Expression):

  """A constant within the expression tree."""

  def Equals(self, other: Expression) -> bool:

    return isinstance(other, Constant) and self.value == other.value

  def Substitute(self, name: str, expression: Expression) -> Expression:

    return self

class Literal(Expression):

  """A runtime literal within the expression tree."""

  def Equals(self, other: Expression) -> bool:

    return isinstance(other, Literal) and self.value == other.value

  def Substitute(self, name: str, expression: Expression) -> Expression:

    return self

def min(lhs: Union[int, float, Expression], rhs: Union[int, float,

                                                       Expression]) -> Function:

class _RewriteIfExpToSelect(ast.NodeTransformer):

  """Transformer to convert if-else nodes to Select expressions."""

  def visit_IfExp(self, node):

    # pylint: disable=invalid-name

  for kw in keywords:

    py = re.sub(rf'Event\(r"{kw}"\)', kw, py)

  try:

    parsed = ast.parse(py, mode='eval')

  except SyntaxError as e:

    raise SyntaxError(f'Parsing expression:\n{orig}') from e

  _RewriteIfExpToSelect().visit(parsed)

  parsed = ast.fix_missing_locations(parsed)

  return _Constify(eval(compile(parsed, orig, 'eval')))

def RewriteMetricsInTermsOfOthers(metrics: List[Tuple[str, Expression]]

                                  )-> Dict[str, Expression]:

  """Shorten metrics by rewriting in terms of others.

  Args:

    metrics (list): pairs of metric names and their expressions.

  Returns:

    Dict: mapping from a metric name to a shortened expression.

  """

  updates: Dict[str, Expression] = dict()

  for outer_name, outer_expression in metrics:

    updated = outer_expression

    while True:

      for inner_name, inner_expression in metrics:

        if inner_name.lower() == outer_name.lower():

          continue

        if inner_name in updates:

          inner_expression = updates[inner_name]

        updated = updated.Substitute(inner_name, inner_expression)

      if updated.Equals(outer_expression):

        break

      if outer_name in updates and updated.Equals(updates[outer_name]):

        break

      updates[outer_name] = updated

  return updates

import unittest

from metric import Constant

from metric import Event

from metric import Expression

from metric import ParsePerfJson

from metric import RewriteMetricsInTermsOfOthers

class TestMetricExpressions(unittest.TestCase):

    after = '0 * SLOTS'

    self.assertEqual(ParsePerfJson(before).Simplify().ToPerfJson(), after)

  def test_RewriteMetricsInTermsOfOthers(self):

    Expression.__eq__ = lambda e1, e2: e1.Equals(e2)

    before = [('m1', ParsePerfJson('a + b + c + d')),

              ('m2', ParsePerfJson('a + b + c'))]

    after = {'m1': ParsePerfJson('m2 + d')}

    self.assertEqual(RewriteMetricsInTermsOfOthers(before), after)

    Expression.__eq__ = None

if __name__ == '__main__':

  unittest.main()