Skip to content
Snippets Groups Projects
Commit 716ed4ac authored by Cyril Moineau's avatar Cyril Moineau
Browse files

Add option to run MakeFile with Docker. + Move MAX_RETRIES has a static variable.

parent 88fb7e80
No related branches found
No related tags found
2 merge requests!340.2.0,!29Fix Docker with makefile
Pipeline #73907 passed
Stage: static_analysis
    Stage: build
      Stage: test
        Stage: coverage
          Hide whitespace changes
          Inline Side-by-side
          aidge_export_arm_cortexm/benchmark.py
          + 55
          − 36
          View file @ 716ed4ac
          import os
          from subprocess import run
          from subprocess import run, DEVNULL, CalledProcessError
          import numpy as np
          import aidge_core
          import aidge_core.export_utils
          ......@@ -15,7 +15,20 @@ from aidge_export_arm_cortexm.flash import BOARD_CONFIG_PATH
          # Default target board for export and benchmarking.
          # Can be overridden externally before calling any function that uses it.
          # Example: aidge_export_arm_cortexm.benchmark.board = "stm32f7"
          BENCHMARK_BOARD= "stm32h7"
          BENCHMARK_BOARD = "stm32h7"
          USE_DOCKER = True
          # Number max of trial to catch the UART
          MAX_RETRIES = 5
          def _image_exists(image_name):
          """Helper function to check if a Docker image exist
          """
          result = run(
          ['docker', 'image', 'inspect', image_name],
          stdout=DEVNULL,
          stderr=DEVNULL
          )
          return result.returncode == 0
          def generate_call_function_arm_cortex_m(export_folder: str, call_function: str, board: str) -> None:
          generate_file(
          ......@@ -30,7 +43,7 @@ def measure_inference_time(model: aidge_core.GraphView, input_data: list[str, np
          export_folder :str = f"{operator_type.lower()}_export_arm_inference"
          board=BENCHMARK_BOARD
          model.set_backend("cpu")
          # create input Tensor list for the GraphView
          # === 1. Creation and injection of inputs ===
          ......@@ -56,7 +69,7 @@ def measure_inference_time(model: aidge_core.GraphView, input_data: list[str, np
          # === 3. Scheduler generation ===
          scheduler = aidge_core.SequentialScheduler(model)
          scheduler.generate_scheduling()
          # === 4. Exporting code (DNN/, Makefile, Src/ etc) ===
          print(" ├─Exporting model ...")
          aidge_export_arm_cortexm.export(
          ......@@ -97,19 +110,31 @@ def measure_inference_time(model: aidge_core.GraphView, input_data: list[str, np
          outputs_dtype=outputs_dtype,
          outputs_size=outputs_size
          )
          # === 7. Compilation firmware ===
          print(" ├─Compiling firmware...")
          log_path = Path(export_folder)/"export_log.log"
          # Clean logs
          if log_path.exists():
          log_path.unlink()
          with log_path.open("a") as log_file:
          run(['make', 'build'], cwd=export_folder, check=True, stdout=log_file, stderr=log_file)
          try:
          if USE_DOCKER:
          if not _image_exists("arm:arm-none-eabi_compiler"):
          run(['make', 'build_image_docker'], cwd=export_folder, check=True, stdout=log_file, stderr=log_file)
          run(['make', 'build_export_docker'], cwd=export_folder, check=True, stdout=log_file, stderr=log_file)
          else:
          run(['make', 'build'], cwd=export_folder, check=True, stdout=log_file, stderr=log_file)
          except CalledProcessError as e:
          raise RuntimeError(f"Fail to build export {export_folder}.\nError log available at: {str(log_path)}.") from e
          # === 8. Flash STM32 + UART and Reading outputs from UART ===
          # === 8. Flash STM32 + UART and Reading outputs from UART ===
          # Attempt to flash and capture UART output multiple times to handle flashing instability.
          #
          #
          # In some cases, the firmware may not start correctly due to an unreliable flashing process,
          # leading to no UART output or incomplete logs. This loop retries flashing and UART capture
          # up to MAX_RETRIES times.
          ......@@ -120,32 +145,28 @@ def measure_inference_time(model: aidge_core.GraphView, input_data: list[str, np
          # 3. If the output file is missing or contains no valid outputs, another attempt is made.
          #
          # The loop exits early as soon as valid outputs are captured.
          MAX_RETRIES = 5
          timings = None
          for attempt in range(1, MAX_RETRIES + 1):
          print(f" ├─Flashing and capturing attempt {attempt}/{MAX_RETRIES}")
          aidge_export_arm_cortexm.flash_and_capture(BOARD_CONFIG_PATH, export_folder=export_folder)
          # Check if the file uart_output exists
          # Check if the file uart_output exists
          uart_output_file = Path(export_folder) / "uart_output.txt"
          if not uart_output_file.is_file():
          aidge_core.Log.error(f"UART output file not found: {uart_output_file}")
          continue # restart the loop
          # === 9. Reading outputs from UART ===
          # === 9. Reading outputs from UART ===
          parsed_uart = aidge_export_arm_cortexm.parsing_uart_output(uart_output_file)
          timings = parsed_uart["timings"]
          if timings: # timings not empty
          break
          break
          else :
          aidge_core.Log.error("No timings captured in uart_output.txt")
          continue
          print(" ├─Completed UART output capture.")
          if not timings:
          ......@@ -155,13 +176,13 @@ def measure_inference_time(model: aidge_core.GraphView, input_data: list[str, np
          print(" └─Inference time done.")
          return timings
          def compute_output(model: aidge_core.GraphView, input_data: list[tuple[str, np.ndarray]]) -> list[np.ndarray]:
          operator_type: str = model.get_ordered_outputs()[0][0].get_operator().type()
          export_folder :str = f"{operator_type.lower()}_export_arm"
          board=BENCHMARK_BOARD
          model.set_backend("cpu")
          # === 1. Creation and injection of inputs ===
          ......@@ -179,7 +200,7 @@ def compute_output(model: aidge_core.GraphView, input_data: list[tuple[str, np.n
          tensor.set_backend("cpu")
          tensor.set_datatype(aidge_core.dtype.float32)
          ordered_inputs.append(tensor)
          for i, (node, idx) in enumerate(model.get_ordered_inputs()):
          node.get_operator().set_input(idx, ordered_inputs[i])
          ......@@ -236,9 +257,9 @@ def compute_output(model: aidge_core.GraphView, input_data: list[tuple[str, np.n
          run(['make', 'build'], cwd=export_folder, check=True, stdout=log_file, stderr=log_file)
          # === 8. Flash STM32 + UART and Reading outputs from UART ===
          # Attempt to flash and capture UART output multiple times to handle flashing instability.
          #
          #
          # In some cases, the firmware may not start correctly due to an unreliable flashing process,
          # leading to no UART output or incomplete logs. This loop retries flashing and UART capture
          # up to MAX_RETRIES times.
          ......@@ -249,9 +270,7 @@ def compute_output(model: aidge_core.GraphView, input_data: list[tuple[str, np.n
          # 3. If the output file is missing or contains no valid outputs, another attempt is made.
          #
          # The loop exits early as soon as valid outputs are captured.
          MAX_RETRIES = 5
          outputs = None
          for attempt in range(1, MAX_RETRIES + 1):
          ......@@ -263,22 +282,22 @@ def compute_output(model: aidge_core.GraphView, input_data: list[tuple[str, np.n
          if not uart_output_file.is_file():
          aidge_core.Log.error(f"UART output file not found: {uart_output_file}")
          continue # Restart the loop
          # === 9. Reading outputs from UART ===
          parsed_uart = aidge_export_arm_cortexm.parsing_uart_output(uart_output_file)
          outputs = parsed_uart["outputs"]
          if outputs: # outputs not empty
          break
          break
          else :
          aidge_core.Log.error("No outputs captured in uart_output.txt")
          continue
          print(" └─Completed UART output capture.")
          if not outputs:
          aidge_core.Log.error("No outputs captured. Check uart_ouput.txt and please restart the benchmark; the firmware might not have started properly on the board")
          aidge_core.Log.error("No outputs captured. Check uart_ouput.txt and please restart the benchmark; the firmware might not have started properly on the board")
          raise SystemExit(1)
          # === 10. Reshape the outputs ===
          for i, (node, idx) in enumerate(model.get_ordered_outputs()):
          ......@@ -290,7 +309,7 @@ def compute_output(model: aidge_core.GraphView, input_data: list[tuple[str, np.n
          dims_permutted = [dims[0], dims[2], dims[1]]
          elif nb_dims == 4:
          dims_permutted = [dims[0], dims[2], dims[3], dims[1]]
          if np.prod(dims) != outputs[i].size:
          aidge_core.Log.fatal("Incompatible export output size ({outputs[i].size}) with required shape {dims}", outputs[i].size, dims)
          raise SystemExit(1)
          ......@@ -303,5 +322,5 @@ def compute_output(model: aidge_core.GraphView, input_data: list[tuple[str, np.n
          outputs[i] = outputs[i].transpose(0, 3, 1, 2)
          print(" └─Compute output done.")
          return outputs
          \ No newline at end of file
          0% or .
          You are about to add 0 people to the discussion. Proceed with caution.
          Finish editing this message first!
          Please register or to comment

          Copyright © Eclipse Foundation, Inc. All Rights Reserved.     Privacy Policy | Terms of Use | Copyright Agent