Lightweight ninja writer in Python

Summary:
- Create python classes for ninja vocbulary in `ninja_syntax.py`. These
  classes will be serialized to a ninja file
- Create a Writer class in `ninja_writer.py`. The current API supports
  adding variables,rules,build actions, etc. This can be extended in the
  future (See `test_ninja_writer.py` for examples)

Future Work:
 - Update the `Subninja` class once chDir is supported (aosp/2064612)
 - Support a width parameter that will be used to wrap long lines of
   text. This will improve readability of the generated files

Expected Use Case: Multi-tree build orchestrator

Test: python ./test_ninja_syntax.py
Test: python ./test_ninja_writer.py

Change-Id: I90c7ee69ddeb7c20c3fd4fca5a911dddbf2253bd

orchestrator/ninja/ninja_syntax.py

@@ -0,0 +1,172 @@
+#!/usr/bin/env python
+#
+# Copyright (C) 2022 The Android Open Source Project
+#
+# Licensed under the Apache License, Version 2.0 (the "License");
+# you may not use this file except in compliance with the License.
+# You may obtain a copy of the License at
+#
+#      http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+
+from abc import ABC, abstractmethod
+
+from collections.abc import Iterator
+from typing import List
+
+TAB = "  "
+
+class Node(ABC):
+  '''An abstract class that can be serialized to a ninja file
+  All other ninja-serializable classes inherit from this class'''
+
+  @abstractmethod
+  def stream(self) -> Iterator[str]:
+    pass
+
+class Variable(Node):
+  '''A ninja variable that can be reused across build actions
+  https://ninja-build.org/manual.html#_variables'''
+
+  def __init__(self, name:str, value:str, indent=0):
+    self.name = name
+    self.value = value
+    self.indent = indent
+
+  def stream(self) -> Iterator[str]:
+    indent = TAB * self.indent
+    yield f"{indent}{self.name} = {self.value}"
+
+class RuleException(Exception):
+  pass
+
+# Ninja rules recognize a limited set of variables
+# https://ninja-build.org/manual.html#ref_rule
+# Keep this list sorted
+RULE_VARIABLES = ["command",
+                  "depfile",
+                  "deps",
+                  "description",
+                  "dyndep",
+                  "generator",
+                  "msvc_deps_prefix",
+                  "restat",
+                  "rspfile",
+                  "rspfile_content"]
+
+class Rule(Node):
+  '''A shorthand for a command line that can be reused
+  https://ninja-build.org/manual.html#_rules'''
+
+  def __init__(self, name:str):
+    self.name = name
+    self.variables = []
+
+  def add_variable(self, name: str, value: str):
+    if name not in RULE_VARIABLES:
+      raise RuleException(f"{name} is not a recognized variable in a ninja rule")
+
+    self.variables.append(Variable(name=name, value=value, indent=1))
+
+  def stream(self) -> Iterator[str]:
+    self._validate_rule()
+
+    yield f"rule {self.name}"
+    # Yield rule variables sorted by `name`
+    for var in sorted(self.variables, key=lambda x: x.name):
+      # variables yield a single item, next() is sufficient
+      yield next(var.stream())
+
+  def _validate_rule(self):
+    # command is a required variable in a ninja rule
+    self._assert_variable_is_not_empty(variable_name="command")
+
+  def _assert_variable_is_not_empty(self, variable_name: str):
+    if not any(var.name == variable_name for var in self.variables):
+      raise RuleException(f"{variable_name} is required in a ninja rule")
+
+class BuildActionException(Exception):
+  pass
+
+class BuildAction(Node):
+  '''Describes the dependency edge between inputs and output
+  https://ninja-build.org/manual.html#_build_statements'''
+
+  def __init__(self, output: str, rule: str, inputs: List[str]=None, implicits: List[str]=None, order_only: List[str]=None):
+    self.output = output
+    self.rule = rule
+    self.inputs = self._as_list(inputs)
+    self.implicits = self._as_list(implicits)
+    self.order_only = self._as_list(order_only)
+    self.variables = []
+
+  def add_variable(self, name: str, value: str):
+    '''Variables limited to the scope of this build action'''
+    self.variables.append(Variable(name=name, value=value, indent=1))
+
+  def stream(self) -> Iterator[str]:
+    self._validate()
+
+    build_statement = f"build {self.output}: {self.rule}"
+    if len(self.inputs) > 0:
+      build_statement += " "
+      build_statement += " ".join(self.inputs)
+    if len(self.implicits) > 0:
+      build_statement += " | "
+      build_statement += " ".join(self.implicits)
+    if len(self.order_only) > 0:
+      build_statement += " || "
+      build_statement += " ".join(self.order_only)
+    yield build_statement
+    # Yield variables sorted by `name`
+    for var in sorted(self.variables, key=lambda x: x.name):
+      # variables yield a single item, next() is sufficient
+      yield next(var.stream())
+
+  def _validate(self):
+    if not self.output:
+      raise BuildActionException("Output is required in a ninja build statement")
+    if not self.rule:
+      raise BuildActionException("Rule is required in a ninja build statement")
+
+  def _as_list(self, list_like):
+    if list_like is None:
+      return []
+    if isinstance(list_like, list):
+      return list_like
+    return [list_like]
+
+class Pool(Node):
+  '''https://ninja-build.org/manual.html#ref_pool'''
+
+  def __init__(self, name: str, depth: int):
+    self.name = name
+    self.depth = Variable(name="depth", value=depth, indent=1)
+
+  def stream(self) -> Iterator[str]:
+    yield f"pool {self.name}"
+    yield next(self.depth.stream())
+
+class Subninja(Node):
+
+  def __init__(self, subninja: str, chDir: str):
+    self.subninja = subninja
+    self.chDir = chDir
+
+  # TODO(spandandas): Update the syntax when aosp/2064612 lands
+  def stream() -> Iterator[str]:
+    yield f"subninja {self.subninja}"
+
+class Line(Node):
+  '''Generic class that can be used for comments/newlines/default_target etc'''
+
+  def __init__(self, value:str):
+    self.value = value
+
+  def stream(self) -> Iterator[str]:
+    yield self.value

orchestrator/ninja/ninja_writer.py

@@ -0,0 +1,55 @@
+#!/usr/bin/env python
+#
+# Copyright (C) 2022 The Android Open Source Project
+#
+# Licensed under the Apache License, Version 2.0 (the "License");
+# you may not use this file except in compliance with the License.
+# You may obtain a copy of the License at
+#
+#      http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+
+from ninja_syntax import Variable, BuildAction, Rule, Pool, Subninja, Line
+
+# TODO: Format the output according to a configurable width variable
+# This will ensure that the generated content fits on a screen and does not
+# require horizontal scrolling
+class Writer:
+
+  def __init__(self, file):
+    self.file = file
+    self.nodes = [] # type Node
+
+  def add_variable(self, variable: Variable):
+    self.nodes.append(variable)
+
+  def add_rule(self, rule: Rule):
+    self.nodes.append(rule)
+
+  def add_build_action(self, build_action: BuildAction):
+    self.nodes.append(build_action)
+
+  def add_pool(self, pool: Pool):
+    self.nodes.append(pool)
+
+  def add_comment(self, comment: str):
+    self.nodes.append(Line(value=f"# {comment}"))
+
+  def add_default(self, default: str):
+    self.nodes.append(Line(value=f"default {default}"))
+
+  def add_newline(self):
+    self.nodes.append(Line(value=""))
+
+  def add_subninja(self, subninja: Subninja):
+    self.nodes.append(subninja)
+
+  def write(self):
+    for node in self.nodes:
+      for line in node.stream():
+        print(line, file=self.file)

orchestrator/ninja/test_ninja_syntax.py

@@ -0,0 +1,107 @@
+#!/usr/bin/env python
+#
+# Copyright (C) 2022 The Android Open Source Project
+#
+# Licensed under the Apache License, Version 2.0 (the "License");
+# you may not use this file except in compliance with the License.
+# You may obtain a copy of the License at
+#
+#      http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+
+import unittest
+
+from ninja_syntax import Variable, Rule, RuleException, BuildAction, BuildActionException, Pool
+
+class TestVariable(unittest.TestCase):
+
+  def test_assignment(self):
+    variable = Variable(name="key", value="value")
+    self.assertEqual("key = value", next(variable.stream()))
+    variable = Variable(name="key", value="value with spaces")
+    self.assertEqual("key = value with spaces", next(variable.stream()))
+    variable = Variable(name="key", value="$some_other_variable")
+    self.assertEqual("key = $some_other_variable", next(variable.stream()))
+
+  def test_indentation(self):
+    variable = Variable(name="key", value="value", indent=0)
+    self.assertEqual("key = value", next(variable.stream()))
+    variable = Variable(name="key", value="value", indent=1)
+    self.assertEqual("  key = value", next(variable.stream()))
+
+class TestRule(unittest.TestCase):
+
+  def test_rulename_comes_first(self):
+    rule = Rule(name="myrule")
+    rule.add_variable("command", "/bin/bash echo")
+    self.assertEqual("rule myrule", next(rule.stream()))
+
+  def test_command_is_a_required_variable(self):
+    rule = Rule(name="myrule")
+    with self.assertRaises(RuleException):
+      next(rule.stream())
+
+  def test_bad_rule_variable(self):
+    rule = Rule(name="myrule")
+    with self.assertRaises(RuleException):
+      rule.add_variable(name="unrecognize_rule_variable", value="value")
+
+  def test_rule_variables_are_indented(self):
+    rule = Rule(name="myrule")
+    rule.add_variable("command", "/bin/bash echo")
+    stream = rule.stream()
+    self.assertEqual("rule myrule", next(stream)) # top-level rule should not be indented
+    self.assertEqual("  command = /bin/bash echo", next(stream))
+
+  def test_rule_variables_are_sorted(self):
+    rule = Rule(name="myrule")
+    rule.add_variable("description", "Adding description before command")
+    rule.add_variable("command", "/bin/bash echo")
+    stream = rule.stream()
+    self.assertEqual("rule myrule", next(stream)) # rule always comes first
+    self.assertEqual("  command = /bin/bash echo", next(stream))
+    self.assertEqual("  description = Adding description before command", next(stream))
+
+class TestBuildAction(unittest.TestCase):
+
+  def test_no_inputs(self):
+    build = BuildAction(output="out", rule="phony")
+    stream = build.stream()
+    self.assertEqual("build out: phony", next(stream))
+    # Empty output
+    build = BuildAction(output="", rule="phony")
+    with self.assertRaises(BuildActionException):
+      next(build.stream())
+    # Empty rule
+    build = BuildAction(output="out", rule="")
+    with self.assertRaises(BuildActionException):
+      next(build.stream())
+
+  def test_inputs(self):
+    build = BuildAction(output="out", rule="cat", inputs=["input1", "input2"])
+    self.assertEqual("build out: cat input1 input2", next(build.stream()))
+    build = BuildAction(output="out", rule="cat", inputs=["input1", "input2"], implicits=["implicits1", "implicits2"], order_only=["order_only1", "order_only2"])
+    self.assertEqual("build out: cat input1 input2 | implicits1 implicits2 || order_only1 order_only2", next(build.stream()))
+
+  def test_variables(self):
+    build = BuildAction(output="out", rule="cat", inputs=["input1", "input2"])
+    build.add_variable(name="myvar", value="myval")
+    stream = build.stream()
+    next(stream)
+    self.assertEqual("  myvar = myval", next(stream))
+
+class TestPool(unittest.TestCase):
+
+  def test_pool(self):
+    pool = Pool(name="mypool", depth=10)
+    stream = pool.stream()
+    self.assertEqual("pool mypool", next(stream))
+    self.assertEqual("  depth = 10", next(stream))
+
+if __name__ == "__main__":
+  unittest.main()

orchestrator/ninja/test_ninja_writer.py

@@ -0,0 +1,54 @@
+#!/usr/bin/env python
+#
+# Copyright (C) 2022 The Android Open Source Project
+#
+# Licensed under the Apache License, Version 2.0 (the "License");
+# you may not use this file except in compliance with the License.
+# You may obtain a copy of the License at
+#
+#      http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+
+import unittest
+
+from io import StringIO
+
+from ninja_writer import Writer
+from ninja_syntax import Variable, Rule, BuildAction
+
+class TestWriter(unittest.TestCase):
+
+  def test_simple_writer(self):
+    with StringIO() as f:
+      writer = Writer(f)
+      writer.add_variable(Variable(name="cflags", value="-Wall"))
+      writer.add_newline()
+      cc = Rule(name="cc")
+      cc.add_variable(name="command", value="gcc $cflags -c $in -o $out")
+      writer.add_rule(cc)
+      writer.add_newline()
+      build_action = BuildAction(output="foo.o", rule="cc", inputs=["foo.c"])
+      writer.add_build_action(build_action)
+      writer.write()
+      self.assertEqual('''cflags = -Wall
+
+rule cc
+  command = gcc $cflags -c $in -o $out
+
+build foo.o: cc foo.c
+''', f.getvalue())
+
+  def test_comment(self):
+    with StringIO() as f:
+      writer = Writer(f)
+      writer.add_comment("This is a comment in a ninja file")
+      writer.write()
+      self.assertEqual("# This is a comment in a ninja file\n", f.getvalue())
+
+if __name__ == "__main__":
+  unittest.main()