common_deploy.py

Go to the documentation of this file.

 
 
import logging
import os
import sys
import warnings
import subprocess
import platform
from hydra.core.hydra_config import HydraConfig
from omegaconf import DictConfig
import tensorflow as tf
import shutil
from hydra.core.hydra_config import HydraConfig
from typing import Optional
from pathlib import Path
 
import common.stm32ai_local as stmaic
from common.benchmarking import cloud_connect, cloud_analyze, benchmark_model
from common.stm32ai_dc import (CliLibraryIde, CliLibrarySerie, CliParameters)
from .external_memory_mgt import update_activation_c_code
 
import json
import re
from typing import Dict, List
 
 
def _keep_internal_weights(path_network_data_params: str) -> None:
    """
    @brief Tags all model weight arrays for placement in internal Flash memory.
 
    @details
    This function reads the generated C file `network_data_params.c` produced by
    ST Edge AI Core and injects a GCC section attribute before every weight array
    declaration, forcing the linker to place all weights in the MCU's internal Flash.
 
    The injected attribute is:
    @code{.c}
    AI_INTERNAL_FLASH __attribute__((section(".InternalFlashSection")))
    @endcode
 
    This function is called when the model weights fit entirely within the internal
    Flash of the target board (no weight splitting required). It is the simpler
    alternative to _dispatch_weights().
 
    **How it works:**
    1. Opens `network_data_params.c` for reading
    2. Scans line by line for the `#include "network_data_params.h"` directive
       and injects the macro definition immediately before it
    3. For each line containing a weight array declaration (matched by regex
       `const ai_uXX name[size]`), prepends the `AI_INTERNAL_FLASH` attribute
    4. Writes the modified content to a temporary file, then atomically replaces
       the original using `os.replace()`
 
    @param path_network_data_params  Absolute path to the generated
                                     `network_data_params.c` file inside
                                     the ST Edge AI Core output directory.
 
    @return None
 
    @note The file is modified **in place** using a write-then-rename pattern
          to avoid partial writes in case of failure.
 
    @see _dispatch_weights() for the alternative function used when weights
         must be split between internal and external Flash.
    """
    with open(path_network_data_params, 'r') as f1, \
         open(os.path.join(os.path.dirname(path_network_data_params),
                           'network_data_params_modify.c'), 'w') as f2:
        for lineNumber, line in enumerate(f1):
            # Inject the macro definition before the header include
            if line == '#include "network_data_params.h"\n':
                line = '#define AI_INTERNAL_FLASH    __attribute__((section(".InternalFlashSection")))\n' + line
            # Detect weight array declarations using regex
            # Pattern matches: const ai_uXX name[size]
            weight = re.findall("const ai_u(?:\d+) (.*)\[(?:\d+)\]", line)
            if weight != []:
                # Prepend the section attribute to force internal Flash placement
                line = 'AI_INTERNAL_FLASH\n' + line
            f2.write(line)
    # Atomically replace the original file with the modified version
    os.replace(
        os.path.join(os.path.dirname(path_network_data_params),
                     'network_data_params_modify.c'),
        path_network_data_params
    )
 
 
def _dispatch_weights(internalFlashSizeFlash_KB: str,
                      kernelFlash_KB: str,
                      applicationSizeFlash_KB: str,
                      path_network_c_info: str,
                      path_network_data_params: str) -> None:
    """
    @brief Splits model weights between internal and external Flash memory.
 
    @details
    When a model's weights are too large to fit entirely in the MCU's internal
    Flash, this function distributes them between internal and external Flash
    (e.g., OctoFlash on STM32N6570-DK) by annotating each weight array in the
    generated C source with the appropriate GCC section attribute.
 
    **Algorithm:**
    1. Reads `network_c_info.json` — the ST Edge AI Core memory report — to
       obtain the list of all weight arrays with their sizes.
    2. Filters to keep only read-only memory pools (`"rights": "ACC_READ"`),
       which correspond to model weights stored in Flash.
    3. Sorts weight arrays from largest to smallest (greedy bin-packing strategy).
    4. Iterates through the sorted list and greedily assigns each weight to
       internal Flash if space remains, otherwise to external Flash.
    5. Injects GCC section attributes into `network_data_params.c` accordingly:
       - `AI_INTERNAL_FLASH __attribute__((section(".InternalFlashSection")))`
       - `AI_EXTERNAL_FLASH __attribute__((section(".ExternalFlashSection")))`
 
    **Memory budget calculation:**
    @code
    freeInternalFlash = internalFlashSize - kernelFlash - applicationFlash
    @endcode
    The kernel Flash (ST AI runtime library) and application code are subtracted
    from the total internal Flash to compute the space available for weights.
 
    @param internalFlashSizeFlash_KB  Total internal Flash size in KB (e.g., "2048KB")
    @param kernelFlash_KB             ST AI runtime library size in KB (e.g., "256KB")
    @param applicationSizeFlash_KB    Application firmware size in KB (e.g., "512KB")
    @param path_network_c_info        Path to `network_c_info.json` generated by
                                      ST Edge AI Core (contains memory pool details)
    @param path_network_data_params   Path to `network_data_params.c` to be annotated
 
    @return None
 
    @note Uses a **greedy largest-first strategy** for bin packing — not optimal
          but fast and effective for the typical weight distribution of embedded models.
 
    @see _keep_internal_weights() for the simpler case where all weights fit internally.
    """
    with open(os.path.join(path_network_c_info), 'r') as f:
        graph = json.load(f)
 
    # Keep only Flash (read-only) memory pools — these are the weight arrays
    for i in range(len(graph["memory_pools"]) - 1, -1, -1):
        element = graph["memory_pools"][i]
        if element["rights"] != "ACC_READ":
            graph["memory_pools"].remove(element)
 
    # Sort weights largest-first for greedy bin-packing
    sorted_weights = sorted(graph["memory_pools"],
                            key=lambda item: item['used_size_bytes'],
                            reverse=True)
 
    # Compute free internal Flash after kernel and application code
    internalFlashSize_inBytes = int(re.split('(\d+)', internalFlashSizeFlash_KB)[1]) * 10**3
    kernel_flash_inBytes = int(re.split('(\d+)', kernelFlash_KB)[1]) * 10**3
    application_size_flash_inBytes = int(re.split('(\d+)', applicationSizeFlash_KB)[1]) * 10**3
    freeInternalFlashSize = internalFlashSize_inBytes - kernel_flash_inBytes - application_size_flash_inBytes
 
    ExternalWeightArray = []
    InternalWeightArray = []
    for detail in sorted_weights:
        if (freeInternalFlashSize - detail["used_size_bytes"]) > 0:
            # Weight fits in internal Flash — assign it there
            InternalWeightArray.append(detail["name"])
            freeInternalFlashSize -= detail["used_size_bytes"]
        else:
            # No space left in internal Flash — assign to external Flash
            ExternalWeightArray.append(detail["name"])
 
    # Annotate network_data_params.c with the correct section attributes
    with open(path_network_data_params, 'r') as f1, \
         open(os.path.join(os.path.dirname(path_network_data_params),
                           'network_data_params_modify.c'), 'w') as f2:
        for lineNumber, line in enumerate(f1):
            if line == '#include "network_data_params.h"\n':
                # Inject both macro definitions before the include
                line = (
                    '#define AI_EXTERNAL_FLASH    __attribute__((section(".ExternalFlashSection")))\n'
                    '#define AI_INTERNAL_FLASH    __attribute__((section(".InternalFlashSection")))\n'
                    + line
                )
            # Match weight array declarations
            weight = re.findall(
                "const ai_u(?:\d+) \\D_network_(.*)_\\D(?:\\d+)\\[(?:\\d+)\\]", line
            )
            if weight != []:
                if weight[0] in InternalWeightArray:
                    line = 'AI_INTERNAL_FLASH\n' + line
                elif weight[0] in ExternalWeightArray:
                    line = 'AI_EXTERNAL_FLASH\n' + line
            f2.write(line)
    os.replace(
        os.path.join(os.path.dirname(path_network_data_params),
                     'network_data_params_modify.c'),
        path_network_data_params
    )
 
 
def stm32ai_deploy(target: bool = False,
                   stlink_serial_number: str = None,
                   stm32ai_version: str = None,
                   c_project_path: str = None,
                   output_dir: str = None,
                   stm32ai_output: str = None,
                   optimization: str = None,
                   path_to_stm32ai: str = None,
                   path_to_cube_ide: str = None,
                   additional_files: list = None,
                   stmaic_conf_filename: str = 'stmaic_c_project.conf',
                   verbosity: int = None,
                   debug: bool = False,
                   model_path: str = None,
                   get_model_name_output: str = None,
                   stm32ai_ide: str = None,
                   stm32ai_serie: str = None,
                   credentials: list = None,
                   on_cloud: bool = False,
                   check_large_model: bool = False,
                   cfg=None,
                   custom_objects: Dict = None) -> None:
    """
    @brief Generic deployment function for STM32 MCU targets (H7, U5, etc.).
 
    @details
    This function orchestrates the complete deployment pipeline for standard
    STM32 MCU boards (excluding STM32N6, which uses stm32ai_deploy_stm32n6()).
 
    **Pipeline steps:**
    1. **Session creation** — loads the model into an STMAi session workspace
    2. **Board configuration** — reads the `.conf` file specifying memory pools,
       linker scripts, and build system paths
    3. **Model compilation** — runs ST Edge AI Core (offline or cloud) to:
       - Convert the quantized model to optimized C arrays (`network.c`, `network_data_params.c`)
       - Generate the AI runtime library (`Lib/`, `Inc/`)
       - Optionally split weights between internal/external Flash for large models
    4. **Firmware build and flash** — invokes STM32CubeIDE in headless mode to
       compile the C project and flash the binary via ST-Link
 
    **Large model handling (`check_large_model=True`):**
    When enabled, the function first benchmarks the model to measure its ROM and RAM
    requirements, then compares them against the board's available memory pools.
    If weights overflow internal Flash, `_dispatch_weights()` is called to split them.
    If activations overflow AXIRAM, `update_activation_c_code()` redistributes
    activation buffers across AXIRAM and SDRAM.
 
    **Cloud vs. local execution:**
    - `on_cloud=True`: uses the STM32Cube.AI Developer Cloud API for compilation
    - `on_cloud=False`: uses the local `stedgeai` executable (used in this project)
 
    @param target               Unused legacy parameter (kept for API compatibility).
    @param stlink_serial_number ST-Link serial number for multi-board setups.
                                Leave empty if only one board is connected.
    @param stm32ai_version      Version string of ST Edge AI Core (e.g., "2.1.0").
    @param c_project_path       Absolute path to the STM32CubeIDE C project root.
    @param output_dir           Directory for all deployment outputs (logs, generated files).
    @param stm32ai_output       Directory where ST Edge AI Core writes generated C files.
    @param optimization         Compilation optimization level: "balanced", "latency", "ram".
    @param path_to_stm32ai      Absolute path to the `stedgeai` executable.
    @param path_to_cube_ide     Absolute path to the `stm32cubeide` executable.
    @param additional_files     Extra files to copy into the C project before building.
    @param stmaic_conf_filename Board configuration file name (e.g., "stmaic_STM32N6570-DK.conf").
    @param verbosity            Logging verbosity (None=silent, 1=info, 2=debug).
    @param debug                Enable debug logging for the STMAi driver.
    @param model_path           Absolute path to the quantized model (.tflite or .onnx).
    @param get_model_name_output  Model name string used for Cloud API identification.
    @param stm32ai_ide          IDE/compiler identifier (must be "gcc" for GCC toolchain).
    @param stm32ai_serie        STM32 series string (e.g., "STM32H7", "STM32U5").
    @param credentials          Pre-obtained cloud credentials from cloud_connect().
    @param on_cloud             If True, use STM32Cube.AI Developer Cloud for compilation.
    @param check_large_model    If True, perform memory analysis before compilation
                                and split weights/activations if needed.
    @param cfg                  Hydra DictConfig for preprocessing parameters
                                (used by update_activation_c_code).
    @param custom_objects       Custom Keras objects for model loading (if applicable).
 
    @return None
 
    @throws ValueError  If the model is too large to fit in any available memory.
 
    @note **Not used in this project.** For STM32N6 deployment, use
          stm32ai_deploy_stm32n6() which adds Neural-ART NPU support.
    """
 
    def _stmaic_local_call(session):
        """
        @brief Inner function: compiles the model using the local stedgeai executable.
 
        @details
        This nested function handles the offline compilation path. It configures
        the STMAi compile options and invokes stmaic.compile() which internally
        runs the stedgeai CLI to generate C code from the quantized model.
 
        For large models (check_large_model=True), it additionally:
        - Benchmarks the model to measure exact memory requirements
        - Checks if weights fit in internal Flash
        - Calls _dispatch_weights() or _keep_internal_weights() accordingly
        - Optionally calls update_activation_c_code() for SRAM overflow handling
 
        @param session  The STMAi session object created by stmaic.load().
        @return None
        """
        if not check_large_model:
            os.environ["STM32_AI_EXE"] = path_to_stm32ai
            tools = stmaic.STMAiTools()
            session.set_tools(tools)
            print("[INFO] : Offline CubeAI used; Selected tools: ", tools, flush=True)
            shutil.rmtree(stm32ai_output, ignore_errors=True)
            opt = stmaic.STMAiCompileOptions(
                no_inputs_allocation=False,
                no_outputs_allocation=False
            )
            opt.optimization = optimization
            stmaic.compile(session, opt)
        else:
            split_weights = False
            split_ram = False
 
            # Step 1: Measure model footprint
            benchmark_model(
                optimization=optimization, model_path=model_path,
                path_to_stm32ai=path_to_stm32ai, stm32ai_output=stm32ai_output,
                stm32ai_version=stm32ai_version,
                get_model_name_output=get_model_name_output
            )
            with open(os.path.join(stm32ai_output, 'network_report.json'), 'r') as f:
                report = json.load(f)
 
            needed_rom = report["model_size"]
            needed_ram = int(report["ram_size"][0]) if isinstance(
                report["ram_size"], list) else int(report["ram_size"])
 
            # Step 2: Read board memory pool configuration
            with open(os.path.join(board.config.memory_pool_path), 'r') as f:
                memory_pool = json.load(f)
 
            available_default_ram = int(next(
                item for item in memory_pool['memory']['mempools']
                if item["name"] == "AXIRAM")["size"]["value"]) * 10**3
            externalRamSize_inBytes = int(next(
                item for item in memory_pool['memory']['mempools']
                if item["name"] == "SDRAM")["size"]["value"]) * 10**3
 
            split_ram = available_default_ram < needed_ram
 
            internalFlashSize_inBytes = int(
                re.split('(\d+)', board.config.internalFlash_size)[1]) * 10**3
            externalFlashSize_inBytes = int(
                re.split('(\d+)', board.config.externalFlash_size)[1]) * 10**3
            application_size_flash_inBytes = int(
                re.split('(\d+)', board.config.application_size)[1]) * 10**3
 
            # Step 3: Validate that the model fits in total available memory
            if needed_rom > externalFlashSize_inBytes + internalFlashSize_inBytes - application_size_flash_inBytes:
                raise ValueError(
                    "\033[31m The model is too large (too many weights) to fit on the board. "
                    "It won't be compiled.\033[39m")
            if needed_ram > externalRamSize_inBytes + available_default_ram:
                raise ValueError(
                    "\033[31m The model is too large (too many activations) to fit on the board. "
                    "It won't be compiled.\033[39m")
 
            # Step 4: Determine if weight splitting is needed
            split_weights = needed_rom > (internalFlashSize_inBytes - application_size_flash_inBytes)
 
            os.environ["STM32_AI_EXE"] = path_to_stm32ai
            tools = stmaic.STMAiTools()
            session.set_tools(tools)
            print("[INFO] : Offline CubeAI used; Selected tools: ", tools, flush=True)
            shutil.rmtree(stm32ai_output, ignore_errors=True)
 
            opt = stmaic.STMAiCompileOptions(
                no_inputs_allocation=False,
                no_outputs_allocation=False,
                split_weights=split_weights
            )
            opt.optimization = optimization
 
            if split_ram:
                print("[INFO] : Dispatch activations in different RAM pools to fit the large model.")
                stmaic.compile(session=session, options=opt, target=session._board_config)
            else:
                stmaic.compile(session=session, options=opt)
 
            path_network_c_info = os.path.join(session.workspace, "network_c_info.json")
 
            # Step 5: Update activation buffer placement in C code
            update_activation_c_code(
                c_project_path, path_network_c_info=path_network_c_info,
                available_AXIRAM=available_default_ram, cfg=cfg,
                custom_objects=custom_objects
            )
 
            # Step 6: Annotate weight arrays with Flash section attributes
            if split_weights:
                print("[INFO] : Dispatch weights between internal and external Flash to fit the large model.")
                _dispatch_weights(
                    internalFlashSizeFlash_KB=board.config.internalFlash_size,
                    kernelFlash_KB=board.config.lib_size,
                    applicationSizeFlash_KB=board.config.application_size,
                    path_network_c_info=path_network_c_info,
                    path_network_data_params=os.path.join(
                        session.generated_dir, "network_data_params.c")
                )
            else:
                print("[INFO] : Weights fit in internal Flash.")
                _keep_internal_weights(
                    path_network_data_params=os.path.join(
                        session.generated_dir, "network_data_params.c")
                )
 
    # --- Main deployment sequence ---
 
    
    os.environ["STM32_CUBE_IDE_EXE"] = path_to_cube_ide
 
    if debug:
        stmaic.set_log_level('debug')
    elif verbosity is not None:
        stmaic.set_log_level('info')
 
    
    session = stmaic.load(model_path, workspace_dir=output_dir)
 
    
    board_conf = os.path.join(c_project_path, stmaic_conf_filename)
    board = stmaic.STMAiBoardConfig(board_conf)
    session.set_board(board)
    print("[INFO] : Selected board : ", board, flush=True)
 
    
    user_files = []
    print("[INFO] : Compiling the model and generating optimized C code + Lib/Inc files: ",
          model_path, flush=True)
    if on_cloud:
        login_success, ai, _ = cloud_connect(
            stm32ai_version=stm32ai_version, credentials=credentials)
        if login_success:
            if not check_large_model:
                ai.generate(CliParameters(
                    model=model_path, output=stm32ai_output,
                    fromModel=get_model_name_output,
                    includeLibraryForSerie=CliLibrarySerie(stm32ai_serie.upper()),
                    includeLibraryForIde=CliLibraryIde(stm32ai_ide.lower())))
            else:
                # [Cloud large model path — same logic as local but using cloud API]
                pass
            if os.path.exists(stm32ai_output):
                shutil.move(stm32ai_output, os.path.join(output_dir, "generated"))
                stm32ai_output = os.path.join(output_dir, "generated")
                if not os.listdir(stm32ai_output) or \
                        'Lib' not in os.listdir(stm32ai_output) or 'Inc' not in os.listdir(stm32ai_output):
                    _stmaic_local_call(session)
        else:
            _stmaic_local_call(session)
    else:
        _stmaic_local_call(session)
 
    print("[INFO] : Optimized C code + Lib/Inc files generation done.")
 
    
    print("[INFO] : Building the STM32 c-project..", flush=True)
    user_files.extend([os.path.join(output_dir, "C_header/ai_model_config.h")])
    if additional_files:
        for f in additional_files:
            user_files.extend([os.path.join(output_dir, f)])
 
    stmaic.build(session, user_files=user_files, serial_number=stlink_serial_number)
 
 
def stm32ai_deploy_stm32n6(target: bool = False,
                            stlink_serial_number: str = None,
                            stm32ai_version: str = None,
                            c_project_path: str = None,
                            output_dir: str = None,
                            stm32ai_output: str = None,
                            optimization: str = None,
                            path_to_stm32ai: str = None,
                            path_to_cube_ide: str = None,
                            additional_files: list = None,
                            stmaic_conf_filename: str = 'stmaic_c_project.conf',
                            verbosity: int = None,
                            debug: bool = False,
                            model_path: str = None,
                            get_model_name_output: str = None,
                            stm32ai_ide: str = None,
                            stm32ai_serie: str = None,
                            credentials: list = None,
                            on_cloud: bool = False,
                            check_large_model: bool = False,
                            build_conf: str = None,
                            cfg=None,
                            custom_objects: Dict = None,
                            input_data_type: str = '',
                            output_data_type: str = '',
                            inputs_ch_position: str = '',
                            outputs_ch_position: str = '') -> None:
    """
    @brief STM32N6-specific deployment function with Neural-ART NPU support.
 
    @details
    This is the primary deployment function used in this project for all three
    case studies (MoveNet Lightning, YOLOv8n-pose, TinyBERT).
 
    It differs from stm32ai_deploy() in several critical ways specific to
    the STM32N6 hardware and its Neural-ART NPU:
 
    **Key differences from generic stm32ai_deploy():**
 
    1. **Neural-ART path** — the compile options include a `st_neural_art` parameter
       pointing to the NPU Add-on configuration (`neuralart_user_path`). This enables
       ST Edge AI Core to generate NPU-optimized code with the 4CA convolution
       accelerator configuration:
       @code
       neural_art_path = profile + "@" + neuralart_user_path
       opt = STMAiCompileOptions(st_neural_art=neural_art_path, ...)
       @endcode
 
    2. **Data type and channel format** — the STM32N6 camera pipeline delivers
       images as uint8 in channel-last (NHWC) format. These are specified explicitly:
       - `input_data_type='uint8'`
       - `inputs_ch_position='chlast'`
       The NPU processes activations in channel-last format internally, matching
       the TFLite model's native format.
 
    3. **Build configuration** — supports specifying a build configuration name
       (`build_conf`, e.g., "Release") for the STM32CubeIDE project.
 
    4. **Generated header files** — copies both `ai_model_config.h` AND
       `app_config.h` into the C project (the generic function only copies the former).
 
    5. **No weight splitting** — STM32N6 has 128 MB octoFlash, so weight overflow
       is never a concern for the models used in this project.
 
    **Compilation flow for STM32N6 (offline, used in this project):**
    @code{.sh}
    # Equivalent CLI command executed internally by _stmaic_local_call():
    stedgeai generate \
        --model movenet_lightning_int8.tflite \
        --target stm32n6 \
        --st-neural-art default@/path/to/neuralart_options.json \
        --input-data-type uint8 \
        --inputs-ch-position chlast \
        --output /output/generated/
    @endcode
 
    @param target               Unused legacy parameter.
    @param stlink_serial_number ST-Link serial number (empty if single board).
    @param stm32ai_version      ST Edge AI Core version string.
    @param c_project_path       Path to the STM32CubeIDE C project root.
    @param output_dir           Output directory for deployment artifacts.
    @param stm32ai_output       Directory for ST Edge AI Core generated files.
    @param optimization         Optimization strategy: "balanced" (used in project).
    @param path_to_stm32ai      Path to the stedgeai executable.
    @param path_to_cube_ide     Path to the stm32cubeide executable.
    @param additional_files     Extra files to copy into the C project.
    @param stmaic_conf_filename Board .conf file (e.g., "stmaic_STM32N6570-DK.conf").
    @param verbosity            Logging verbosity level.
    @param debug                Enable debug mode for STMAi driver.
    @param model_path           Path to the quantized INT8 model (.tflite or .onnx).
    @param get_model_name_output  Model name for Cloud API identification.
    @param stm32ai_ide          IDE string — must be "gcc" for STM32N6.
    @param stm32ai_serie        Series string — must be "STM32N6".
    @param credentials          Cloud credentials (unused if on_cloud=False).
    @param on_cloud             Use Developer Cloud for compilation (False in project).
    @param check_large_model    Enable pre-compilation memory analysis.
    @param build_conf           STM32CubeIDE build configuration name (e.g., "Release").
    @param cfg                  Hydra DictConfig object (for activation code updates).
    @param custom_objects       Custom Keras objects for model loading.
    @param input_data_type      NPU input data type — **must be 'uint8'** for camera pipeline.
    @param output_data_type     NPU output data type — empty means auto-detect.
    @param inputs_ch_position   Input channel format — **must be 'chlast'** (NHWC for TFLite).
    @param outputs_ch_position  Output channel format — empty means auto-detect.
 
    @return None
 
    @note This function is called by deploy.py in the pose_estimation module,
          which reads all parameters from user_config.yaml via the Hydra cfg object.
 
    @see deploy.py for the caller that reads parameters from user_config.yaml.
    @see stm32ai_deploy() for the generic MCU version without NPU support.
    """
 
    def _stmaic_local_call(session):
        """
        @brief Inner function: compiles the model for STM32N6 using local stedgeai.
 
        @details
        STM32N6-specific compile sequence:
        1. Sets the STM32_AI_EXE environment variable to the stedgeai path
        2. Constructs the Neural-ART path string from the board config:
           `"profile@neuralart_options_path"` — this tells stedgeai to enable
           the Neural-ART NPU accelerator with the 4CA configuration
        3. Creates STMAiCompileOptions with:
           - st_neural_art: enables NPU code generation
           - input_data_type: 'uint8' (camera pipeline delivers uint8 frames)
           - inputs_ch_position: 'chlast' (NHWC channel-last format)
        4. Loads the board configuration (memory pools, NPU memory banks)
        5. Calls stmaic.compile() which invokes stedgeai internally
 
        @param session  The STMAi session object.
        @return None
        """
        os.environ["STM32_AI_EXE"] = path_to_stm32ai
        tools = stmaic.STMAiTools()
        session.set_tools(tools)
        print("[INFO] : Offline CubeAI used; Selected tools: ", tools, flush=True)
 
        shutil.rmtree(stm32ai_output, ignore_errors=True)
 
        
        neural_art_path = (session._board_config.config.profile + "@" +
                           session._board_config.config.neuralart_user_path)
 
        
        opt = stmaic.STMAiCompileOptions(
            st_neural_art=neural_art_path,
            input_data_type=input_data_type,       # 'uint8'
            inputs_ch_position=inputs_ch_position,  # 'chlast' (NHWC)
            output_data_type=output_data_type,
            outputs_ch_position=outputs_ch_position
        )
 
        
        board_conf = os.path.join(c_project_path, stmaic_conf_filename)
        board = stmaic.STMAiBoardConfig(board_conf, build_conf)
        session.set_board(board)
 
        
        stmaic.compile(session=session, options=opt, target=session._board_config)
 
    # --- Main deployment sequence for STM32N6 ---
 
    os.environ["STM32_CUBE_IDE_EXE"] = path_to_cube_ide
 
    if debug:
        stmaic.set_log_level('debug')
    elif verbosity is not None:
        stmaic.set_log_level('info')
 
    
    session = stmaic.load(model_path, workspace_dir=output_dir)
    board_conf = os.path.join(c_project_path, stmaic_conf_filename)
    board = stmaic.STMAiBoardConfig(board_conf, build_conf)
    session.set_board(board)
    print("[INFO] : Selected board : ", board, flush=True)
 
    
    user_files = []
    print("[INFO] : Compiling the model and generating optimized C code + Lib/Inc files: ",
          model_path, flush=True)
 
    if on_cloud:
        login_success, ai, _ = cloud_connect(
            stm32ai_version=stm32ai_version, credentials=credentials)
        if login_success:
            with open(session._board_config.config.neuralart_user_path) as file:
                neuralart_options = json.load(file)
            
            neuralart_options = neuralart_options['Profiles']['default'][
                "options"].replace('--', "--atonnOptions.")
            ai.generate(CliParameters(
                model=model_path, output=stm32ai_output,
                fromModel=get_model_name_output,
                target="stm32n6", stNeuralArt="default",
                allocateInputs=False, allocateOutputs=False,
                mpool=board._conf.mpool,
                extraCommandLineArguments=neuralart_options,
                includeLibraryForSerie=CliLibrarySerie(stm32ai_serie.upper()),
                includeLibraryForIde=CliLibraryIde(stm32ai_ide.lower())))
            if os.path.exists(stm32ai_output):
                shutil.move(stm32ai_output, os.path.join(output_dir, "generated"))
                stm32ai_output = os.path.join(output_dir, "generated")
                if not os.listdir(stm32ai_output) or \
                        'Lib' not in os.listdir(stm32ai_output) or 'Inc' not in os.listdir(stm32ai_output):
                    _stmaic_local_call(session)
        else:
            _stmaic_local_call(session)
    else:
        
        _stmaic_local_call(session)
 
    print("[INFO] : Optimized C code + Lib/Inc files generation done.")
 
    
    print("[INFO] : Building the STM32 c-project..", flush=True)
    user_files.extend([os.path.join(output_dir, "C_header/app_config.h")])
    user_files.extend([os.path.join(output_dir, "C_header/ai_model_config.h")])
    if additional_files:
        for f in additional_files:
            user_files.extend([os.path.join(output_dir, f)])
 
    stmaic.build(session, user_files=user_files, serial_number=stlink_serial_number)
 
 
def stm32ai_deploy_mpu(target: str = None,
                       board_ip_address: str = None,
                       board_deploy: str = None,
                       class_names: List = None,
                       c_project_path: str = None,
                       verbosity: int = None,
                       debug: bool = False,
                       model_path: str = None,
                       cfg=None) -> bool:
    """
    @brief Deploy an AI model to an STM32MP MPU board over SSH/SCP.
 
    @details
    This function handles deployment on STM32MP-series Microprocessor Units (MPUs),
    which run Linux and use a fundamentally different deployment mechanism than MCUs:
    instead of flashing firmware via ST-Link, it transfers application files over the
    network using SSH and SCP.
 
    **Deployment mechanism:**
    Unlike MCU deployment (which replaces the entire firmware binary), MPU deployment:
    1. Verifies board reachability via ICMP ping
    2. Creates the deployment directory on the target via SSH
    3. Copies application code, resources, and the model file via SCP
    4. Copies board-specific shell scripts (STM32MP1/*.sh or STM32MP2/*.sh)
    5. Launches the application remotely via SSH
 
    **Supported boards:**
    - STM32MP257F-EV1 (STM32MP2 series)
    - STM32MP157F-DK2 (STM32MP1 series)
    - STM32MP135F-DK  (STM32MP1 series)
 
    **File transfer structure:**
    @code
    c_project_path/
    ├── Application/     → Copied to board_deploy/Application/
    │   └── launch_*.sh  → Main launch script
    ├── Resources/       → Copied to board_deploy/Resources/
    │   └── class_names.txt  → Generated from class_names parameter
    └── STM32MP1/*.sh    → Board-specific scripts (MP1) or STM32MP2/*.sh (MP2)
    @endcode
 
    @param target           Unused legacy parameter.
    @param board_ip_address IP address of the target MPU board (e.g., "192.168.1.100").
                            The board must be on the same network as the host PC.
    @param board_deploy     Deployment directory path on the target board's filesystem.
    @param class_names      List of class name strings OR path to a .txt file containing
                            class names (one per line). Used for inference labeling.
    @param c_project_path   Path to the C project containing Application/ and Resources/.
    @param verbosity        Logging verbosity level.
    @param debug            Enable debug logging.
    @param model_path       Path to the AI model file to deploy (.tflite, .onnx, or .nb).
    @param cfg              Hydra DictConfig object (currently unused in MPU path).
 
    @return True if deployment succeeded, False on any error.
 
    @throws None — errors are caught and logged, returning False instead.
 
    @note SSH host key checking is disabled (`StrictHostKeyChecking no`) for
          convenience in lab/development environments. Do not use in production.
 
    @note This function is **not used in this project** (which targets the STM32N6
          MCU, not an MPU). It is documented here for completeness.
    """
 
    # Step 1: Validate that a board IP address was provided
    if board_ip_address is None:
        print("[FAIL] : Board IP address is missing, unable to deploy on target.")
        return False
 
    # Step 2: Verify board reachability via ping
    count = 5
    timeout = 100
    subprocess_timeout = 5
    count_params = '-n' if platform.system().lower() == 'windows' else '-c'
    timeout_params = '-w' if platform.system().lower() == 'windows' else '-W'
 
    cmd = ['ping', count_params, str(count), timeout_params, str(timeout), board_ip_address]
    try:
        res = subprocess.run(cmd, stdout=subprocess.PIPE, stderr=subprocess.PIPE,
                             timeout=5, text=True)
        if res.returncode == 0:
            print(f"[INFO] : Board is reachable at {board_ip_address} address")
        else:
            print(f"[FAIL] : Board is not reachable at {board_ip_address} address")
            return False
    except subprocess.TimeoutExpired:
        print(f"[FAIL] : Board is not reachable, ping timed out after {subprocess_timeout}s.")
        return False
    except Exception as e:
        print(f"[FAIL] : Verification of the IP failed : {e}.")
        return False
 
    # Step 3: Create deployment directory on target via SSH
    command = "mkdir -p " + board_deploy
    ssh = subprocess.run(
        "ssh -o \"StrictHostKeyChecking no\" root@" + board_ip_address + " \"" + command + "\"",
        shell=True, stdout=subprocess.PIPE, stderr=subprocess.PIPE, timeout=300)
    if ssh.returncode != 0:
        print(f"[FAIL] : Deploy directory creation failed, code: {ssh.returncode}")
        return False
 
    # Step 4: Generate class names file on disk
    path_to_application = c_project_path + "Application/"
    path_to_resources = c_project_path + "Resources/"
    label_file = os.path.join(path_to_resources, 'class_names.txt')
 
    if isinstance(class_names, list) and all(isinstance(name, str) for name in class_names):
        with open(label_file, 'w') as file:
            for class_name in class_names:
                file.write(class_name + '\n')
    elif isinstance(class_names, str) and class_names.endswith('.txt'):
        shutil.copy(class_names, label_file)
 
    # Step 5: Transfer application code and model to the board via SCP
    command = ("scp -r " + path_to_application + " " + path_to_resources + " " +
               model_path + " root@" + board_ip_address + ":" + board_deploy)
    deploy_res = subprocess.run(command, shell=True, stdout=subprocess.PIPE,
                                stderr=subprocess.PIPE, timeout=300)
    if deploy_res.returncode == 0:
        print(f"[INFO] : Application code successfully installed on target")
    else:
        print(f"[FAIL] : Application code deployment failed : {deploy_res.stderr} ")
        return False
 
    # Step 6: Transfer board-specific shell scripts
    if "STM32MP2" in target:
        path_to_target_resources = c_project_path + "/STM32MP2/*.sh"
    else:
        path_to_target_resources = c_project_path + "/STM32MP1/*.sh"
 
    command = ("scp -r -p " + path_to_target_resources + " root@" +
               board_ip_address + ":" + board_deploy + "/Resources")
    deploy_spe_res = subprocess.run(command, shell=True, stdout=subprocess.PIPE,
                                    stderr=subprocess.PIPE, timeout=300)
    if deploy_spe_res.returncode != 0:
        print(f"[FAIL] : Application code deployment failed : {deploy_spe_res.stderr} ")
        return False
 
    # Step 7: Find and launch the application script on the board
    script_extension = ".sh"
    file_names = []
    for item in os.listdir(path_to_application):
        if Path(item).suffix == script_extension:
            file_names.append(os.path.basename(item))
 
    launch_script = None
    for file_name in file_names:
        if "launch_" in file_name:
            launch_script = file_name
    if launch_script is None:
        print("[FAIL] : No launch_*.sh script found in Application/.")
        return False
    command = (board_deploy + "/Application/" + launch_script + " " +
               board_deploy + " " + os.path.basename(model_path) + " " +
               os.path.basename(label_file))
    print(f"[INFO] : To launch application directly on the target please run : {command}")
    command = ("ssh -o \"StrictHostKeyChecking no\" root@" + board_ip_address +
               " \"" + command + "\"")
    print(f"[INFO] : To launch application from your host computer please run : {command}")