exir: remove JSON program conversion path

Change-Id: Iebb6ff9151b76b352ef5dbb4d9bd23e2e622c326
Signed-off-by: Chizkiyahu Raful <chizkiyahu.raful@arm.com>

Author: chizkiyahu

exir/_serialize/_flatbuffer.py

@@ -12,7 +12,6 @@
 import importlib.resources
 import os
 import re
-import shutil
 import stat
 import subprocess
 import tempfile
@@ -384,72 +383,6 @@ def _flatc_decompile(
     )
 
 
-def _program_json_to_flatbuffer(
-    program_json: str,
-    *,
-    constant_tensor_alignment: Optional[int] = None,
-    delegate_alignment: Optional[int] = None,
-) -> _FlatbufferResult:
-    """Converts Program-compatible JSON into binary flatbuffer data.
-
-    Args:
-        program_json: The JSON to convert. Must be compatible with the root
-            table type of //executorch/schema/program.fbs.
-        constant_tensor_alignment: If provided, the alignment to use for tensor
-            data embedded in the output flatbuffer data. If not provided, uses
-            the alignment in the schema.
-        delegate_alignment: If provided, the alignment to use for delegate
-            data embedded in the output flatbuffer data. If not provided, uses
-            the alignment in the schema.
-
-    Returns: The flatbuffer data and associated metadata.
-    """
-    with tempfile.TemporaryDirectory() as temp_dir:
-        schema_info = _prepare_schema(
-            out_dir=temp_dir,
-            constant_tensor_alignment=constant_tensor_alignment,
-            delegate_alignment=delegate_alignment,
-        )
-        file_stem = "data"
-        json_path = os.path.join(temp_dir, file_stem + ".json")
-        output_path = os.path.join(temp_dir, file_stem + ".pte")
-
-        with open(json_path, "wb") as json_file:
-            json_file.write(program_json.encode("ascii"))
-
-        try:
-            _flatc_compile(temp_dir, schema_info.root_path, json_path)
-        except Exception as err:
-            # It's helpful to save the breaking files for debugging. Optionally
-            # move them out of the auto-deleting temporary directory. Don't do
-            # this by default because some input files can be many GB in size,
-            # and these copies won't be auto-deleted.
-            should_save = os.getenv(_SAVE_FLATC_ENV, "").strip() not in {"", "0"}
-            extra_message = ""
-            if should_save:
-                try:
-                    saved_dir = tempfile.mkdtemp(prefix="exir-saved-flatc-")
-                    for f in os.listdir(temp_dir):
-                        shutil.move(src=os.path.join(temp_dir, f), dst=saved_dir)
-                    extra_message += f" Moved input files to '{saved_dir}'."
-                except Exception as err2:
-                    extra_message += (
-                        f" (Failed to save input files for debugging: {err2})"
-                    )
-            else:
-                extra_message += (
-                    f" Set {_SAVE_FLATC_ENV}=1 to save input files on failure."
-                )
-
-            raise RuntimeError(
-                f"Failed to compile {json_path} to {output_path}." + extra_message
-            ) from err
-        with open(output_path, "rb") as output_file:
-            return _FlatbufferResult(
-                data=output_file.read(), max_alignment=schema_info.max_alignment
-            )
-
-
 def _replace_infinity_in_json_file(content: bytes) -> bytes:
     """Replace -inf and inf with "inf" and "-inf" in the JSON file. program.fbs
     is used to convert from flatbuffer to JSON. +-inf float values are not

exir/_serialize/_program.py

@@ -16,7 +16,7 @@
 from typing import ClassVar, Dict, List, Literal, Optional, Sequence, Tuple
 
 from executorch.exir._serialize._cord import Cord
-from executorch.exir._serialize._dataclass import _DataclassEncoder, _json_to_dataclass
+from executorch.exir._serialize._dataclass import _DataclassEncoder
 from executorch.exir._serialize._flatbuffer import _FlatbufferResult
 from executorch.exir._serialize._flatbuffer_program import (
     _flatbuffer_to_program,
@@ -86,12 +86,6 @@ def _program_to_json(program: Program) -> str:
     return json.dumps(program, cls=_DataclassEncoder)
 
 
-def _json_to_program(program_json: bytes) -> Program:
-    """Returns a Program deserialized from the given JSON string."""
-    # construct program class recursively from dict
-    return _json_to_dataclass(json.loads(program_json), cls=Program)
-
-
 def _insert_flatbuffer_header(
     flatbuffer_data: bytes, magic_regex: str, header_data: bytes
 ) -> bytes:
@@ -797,9 +791,7 @@ def _extract_delegate_payload(
         program_size = len(pte_data)
 
     # Parse the program flatbuffer
-    program: Program = _json_to_program(
-        _program_flatbuffer_to_json(pte_data[:program_size])
-    )
+    program: Program = _flatbuffer_to_program(pte_data[:program_size])
 
     # Search for the matching delegate
     match_count = 0

exir/_serialize/test/test_flatbuffer.py

@@ -7,19 +7,13 @@
 # LICENSE file in the root directory of this source tree.
 
 import os
-import re
-import shutil
 import tempfile
 import unittest
 from typing import Dict, Optional, Sequence
 from unittest.mock import patch
 
 from executorch.exir._serialize import _flatbuffer
-from executorch.exir._serialize._flatbuffer import (
-    _program_json_to_flatbuffer,
-    _ResourceFiles,
-    _SchemaInfo,
-)
+from executorch.exir._serialize._flatbuffer import _ResourceFiles, _SchemaInfo
 
 
 def read_file(dir: str, filename: str) -> bytes:
@@ -277,60 +271,3 @@ def test_bad_delegate_alignment_fails(self) -> None:
                             out_dir,
                             delegate_alignment=bad_alignment,
                         )
-
-
-class TestProgramJsonToFlatbuffer(unittest.TestCase):
-    @patch.dict(os.environ, {_flatbuffer._SAVE_FLATC_ENV: "1"})
-    def test_save_json_on_failure(self) -> None:
-        err_msg: Optional[str] = None
-        try:
-            _program_json_to_flatbuffer("} some bad json {")
-            self.fail("Should have raised an exception")
-        except RuntimeError as err:
-            err_msg = err.args[0]
-
-        self.assertIsNotNone(err_msg)
-        match = re.search(r"Moved input files to '(.*?)'", err_msg)
-        self.assertTrue(match, msg=f"Unexpected error message: {err_msg}")
-        path = match.group(1)
-
-        files = frozenset(os.listdir(path))
-        # Delete the files otherwise they'll accumulate every time the
-        # test is run.
-        shutil.rmtree(path)
-        # Check for a couple of the files that should be there.
-        self.assertIn("data.json", files)
-        self.assertIn("program.fbs", files)
-
-    @patch.dict(os.environ, {_flatbuffer._SAVE_FLATC_ENV: "1"})
-    def test_unable_to_save_json_on_failure(self) -> None:
-        err_msg: Optional[str] = None
-        try:
-            with patch.object(
-                _flatbuffer.shutil,
-                "move",
-                side_effect=Exception("shutil.move mock failure"),
-            ):
-                _program_json_to_flatbuffer("} some bad json {")
-            self.fail("Should have raised an exception")
-        except RuntimeError as err:
-            err_msg = err.args[0]
-
-        self.assertIsNotNone(err_msg)
-        self.assertIn("Failed to save input files", err_msg)
-
-    @patch.dict(os.environ, {_flatbuffer._SAVE_FLATC_ENV: ""})
-    def test_no_save_json_on_failure(self) -> None:
-        err_msg: Optional[str] = None
-        try:
-            _program_json_to_flatbuffer("} some bad json {")
-            self.fail("Should have raised an exception")
-        except RuntimeError as err:
-            err_msg = err.args[0]
-
-        self.assertIsNotNone(err_msg)
-        self.assertIn(
-            f"Set {_flatbuffer._SAVE_FLATC_ENV}=1 to save input files", err_msg
-        )
-        self.assertNotIn("Moved input files", err_msg)
-        self.assertNotIn("Failed to save input files", err_msg)

exir/_serialize/test/test_flatbuffer_program.py

@@ -4,18 +4,12 @@
 # This source code is licensed under the BSD-style license found in the
 # LICENSE file in the root directory of this source tree.
 
-import json
 import unittest
 
-from executorch.exir._serialize._flatbuffer import (
-    _program_flatbuffer_to_json,
-    _program_json_to_flatbuffer,
-)
 from executorch.exir._serialize._flatbuffer_program import (
     _flatbuffer_to_program,
     _program_to_flatbuffer,
 )
-from executorch.exir._serialize._program import _json_to_program, _program_to_json
 from executorch.exir.backend.compile_spec_schema import CompileSpec
 from executorch.exir.schema import (
     AllocationDetails,
@@ -160,58 +154,13 @@ def _make_program(self) -> Program:
             named_data=[],
         )
 
-    def _flatbuffer_to_dict(self, flatbuffer_data: bytes) -> dict:
-        return json.loads(_program_flatbuffer_to_json(flatbuffer_data))
-
-    def test_roundtrip_via_json(self) -> None:
-        program = self._make_program()
-        result = _program_to_flatbuffer(
-            program, constant_tensor_alignment=32, delegate_alignment=64
-        )
-        self.assertGreater(len(result.data), 8)
-        self.assertEqual(result.data[4:6], b"ET")
-        self.assertGreaterEqual(result.max_alignment, 64)
-
-        program2 = _json_to_program(_program_flatbuffer_to_json(result.data))
-        self.assertEqual(program2, program)
-
     def test_roundtrip_via_direct_python(self) -> None:
         program = self._make_program()
         result = _program_to_flatbuffer(
             program, constant_tensor_alignment=32, delegate_alignment=64
         )
         self.assertEqual(_flatbuffer_to_program(result.data), program)
 
-    def test_flatbuffer_paths_match(self) -> None:
-        program = self._make_program()
-        cases = [
-            (None, None),
-            (32, 64),
-        ]
-        for constant_tensor_alignment, delegate_alignment in cases:
-            with self.subTest(
-                constant_tensor_alignment=constant_tensor_alignment,
-                delegate_alignment=delegate_alignment,
-            ):
-                result = _program_to_flatbuffer(
-                    program,
-                    constant_tensor_alignment=constant_tensor_alignment,
-                    delegate_alignment=delegate_alignment,
-                )
-                result2 = _program_json_to_flatbuffer(
-                    _program_to_json(program),
-                    constant_tensor_alignment=constant_tensor_alignment,
-                    delegate_alignment=delegate_alignment,
-                )
-                direct_dict = self._flatbuffer_to_dict(result.data)
-                json_path_dict = self._flatbuffer_to_dict(result2.data)
-                self.assertEqual(
-                    direct_dict,
-                    json_path_dict,
-                    "Flatbuffer JSON differs between direct and JSON paths",
-                )
-                self.assertEqual(result.max_alignment, result2.max_alignment)
-
     def test_bad_alignment_fails(self) -> None:
         program = Program(
             version=0,