And here are the most Important Scripts I’m using for the compilation Process:
Create_Calibration_Images
from PIL import Image
import json
import os
from tqdm import tqdm
import random
def resize_and_pad(img, target_w, target_h):
"""Fit the full image into the target size and pad the remaining area."""
ratio = min(target_w / img.width, target_h / img.height)
new_width = round(img.width * ratio)
new_height = round(img.height * ratio)
resized_img = img.resize((new_width, new_height), Image.Resampling.LANCZOS)
canvas = Image.new("RGB", (target_w, target_h), (114, 114, 114))
left = (target_w - new_width) // 2
top = (target_h - new_height) // 2
canvas.paste(resized_img, (left, top))
return canvas
def find_image_files(directory, extensions):
"""Find all image files with specified extensions in directory and its subdirectories."""
image_files = []
for root, _, files in os.walk(directory):
for file in files:
if any(file.lower().endswith(ext.lower()) for ext in extensions):
image_files.append(os.path.join(root, file))
return image_files
class CalibrationImageCreator:
def __init__(self, image_dir, output_dir, image_size,
num_images, input_image_extensions):
self.image_dir = image_dir
self.output_dir = output_dir
self.image_size = image_size
self.num_images = num_images
self.input_image_extensions = input_image_extensions
def create_calibration_images(self):
#target_dir_path = os.path.join(self.dataset_dir, "calib_images")
os.makedirs(self.output_dir, exist_ok=True)
metadata_path = os.path.join(self.output_dir, ".calibration_images_config.json")
if os.path.exists(metadata_path):
os.remove(metadata_path)
#target_dataset_dir = os.path.join(self.dataset_dir, "images")
print(f"Searching for images in {self.image_dir}...")
all_image_paths = find_image_files(self.image_dir, self.input_image_extensions)
if not all_image_paths:
print(f"No images found in {self.image_dir} with extensions {self.input_image_extensions}")
return
print(f"Found {len(all_image_paths)} images")
random.shuffle(all_image_paths)
images_list = all_image_paths[:self.num_images]
actual_num_images = len(images_list)
if actual_num_images < self.num_images:
print(f"Warning: Only found {actual_num_images} images, less than the requested {self.num_images}")
print("Creating calibration images for Hailo export")
created_count = 0
for idx, filepath in tqdm(enumerate(images_list), total=actual_num_images):
try:
with Image.open(filepath) as img:
img = img.convert("RGB")
processed_img = resize_and_pad(img, self.image_size[0], self.image_size[1])
output_filepath = os.path.join(self.output_dir, f"processed_{idx}.jpg")
processed_img.save(output_filepath, format="JPEG")
created_count += 1
except Exception as e:
print(f"Error processing {filepath}: {e}")
with open(metadata_path, "w", encoding="utf-8") as metadata_file:
json.dump(
{
"resize_mode": "letterbox",
"image_size": list(self.image_size),
"requested_num_images": self.num_images,
"created_num_images": created_count,
"padding_color": [114, 114, 114],
},
metadata_file,
)
Create_Calibration_Data
import json
import os
import numpy as np
from PIL import Image
class CalibrationDataProcessor:
FLOAT32_CHUNK_SIZE = 64
def __init__(self, image_dir, output_dir, image_size):
self.image_dir = image_dir
self.output_dir = output_dir
self.image_size = image_size
self.input_image_extensions = (".jpg", ".jpeg", ".png")
@staticmethod
def _available_memory_bytes():
available_bytes = None
try:
with open("/proc/meminfo", encoding="utf-8") as memory_info:
for line in memory_info:
if line.startswith("MemAvailable:"):
available_bytes = int(line.split()[1]) * 1024
break
except (OSError, ValueError):
pass
if available_bytes is None:
try:
available_bytes = (
os.sysconf("SC_AVPHYS_PAGES") * os.sysconf("SC_PAGE_SIZE")
)
except (OSError, ValueError, AttributeError):
return 0
cgroup_limits = (
("/sys/fs/cgroup/memory.max", "/sys/fs/cgroup/memory.current"),
(
"/sys/fs/cgroup/memory/memory.limit_in_bytes",
"/sys/fs/cgroup/memory/memory.usage_in_bytes",
),
)
for limit_path, usage_path in cgroup_limits:
try:
with open(limit_path, encoding="utf-8") as limit_file:
limit_value = limit_file.read().strip()
with open(usage_path, encoding="utf-8") as usage_file:
usage_bytes = int(usage_file.read().strip())
if limit_value == "max":
continue
limit_bytes = int(limit_value)
if 0 < limit_bytes < 2**60:
available_bytes = min(
available_bytes,
max(0, limit_bytes - usage_bytes),
)
break
except (OSError, ValueError):
continue
return available_bytes
def _write_float64_dataset(
self,
image_names,
normalized_path,
processed_path,
completion_marker_path,
array_shape,
):
calibration_data = []
for image_name in image_names:
image_path = os.path.join(self.image_dir, image_name)
with Image.open(image_path) as image:
image = image.convert("RGB").resize(self.image_size)
calibration_data.append(np.asarray(image) / 255.0)
calibration_data_normalized = np.asarray(calibration_data)
np.save(normalized_path, calibration_data_normalized)
calibration_data_processed = calibration_data_normalized * 255.0
np.save(processed_path, calibration_data_processed)
with open(completion_marker_path, "w", encoding="utf-8") as marker_file:
json.dump(
{
"image_count": len(image_names),
"shape": array_shape,
"dtype": str(calibration_data_normalized.dtype),
"method": "float64",
},
marker_file,
)
def _write_chunked_float32_dataset(
self,
image_names,
normalized_path,
processed_path,
completion_marker_path,
array_shape,
):
normalized_temp_path = f"{normalized_path}.tmp.npy"
processed_temp_path = f"{processed_path}.tmp.npy"
normalized_data = np.lib.format.open_memmap(
normalized_temp_path,
mode="w+",
dtype=np.float32,
shape=array_shape,
)
processed_data = np.lib.format.open_memmap(
processed_temp_path,
mode="w+",
dtype=np.float32,
shape=array_shape,
)
chunk_size = self.FLOAT32_CHUNK_SIZE
for chunk_start in range(0, len(image_names), chunk_size):
chunk_end = min(chunk_start + chunk_size, len(image_names))
chunk_shape = (
chunk_end - chunk_start,
self.image_size[1],
self.image_size[0],
3,
)
image_batch = np.empty(chunk_shape, dtype=np.uint8)
for batch_index, image_name in enumerate(image_names[chunk_start:chunk_end]):
image_path = os.path.join(self.image_dir, image_name)
with Image.open(image_path) as image:
image = image.convert("RGB").resize(self.image_size)
image_batch[batch_index] = np.asarray(image, dtype=np.uint8)
normalized_batch = image_batch.astype(np.float32)
normalized_batch /= np.float32(255.0)
normalized_data[chunk_start:chunk_end] = normalized_batch
processed_data[chunk_start:chunk_end] = image_batch
normalized_data.flush()
processed_data.flush()
del normalized_data, processed_data
os.replace(normalized_temp_path, normalized_path)
os.replace(processed_temp_path, processed_path)
with open(completion_marker_path, "w", encoding="utf-8") as marker_file:
json.dump(
{
"image_count": len(image_names),
"shape": array_shape,
"dtype": "float32",
"method": "chunked_float32",
},
marker_file,
)
def create_calibration_dataset(self):
os.makedirs(self.output_dir, exist_ok=True)
calibration_data_path = os.path.join(self.output_dir, "calibration_data_normalized.npy")
processed_data_path = os.path.join(self.output_dir, "calibration_data_processed.npy")
completion_marker_path = os.path.join(
self.output_dir,
".calibration_data_complete.json",
)
if os.path.exists(completion_marker_path):
os.remove(completion_marker_path)
image_names = sorted(
image_name
for image_name in os.listdir(self.image_dir)
if image_name.lower().endswith(self.input_image_extensions)
and os.path.isfile(os.path.join(self.image_dir, image_name))
)
if not image_names:
raise ValueError(f"No calibration images found in {self.image_dir}")
print("Creating calibration datasets for Hailo export")
image_count = len(image_names)
array_shape = (
image_count,
self.image_size[1],
self.image_size[0],
3,
)
available_bytes = self._available_memory_bytes()
array_bytes = int(np.prod(array_shape)) * np.dtype(np.float64).itemsize
estimated_float64_peak = 3 * array_bytes + 256 * 1024**2
float32_chunk_pixels = (
min(image_count, self.FLOAT32_CHUNK_SIZE)
* self.image_size[1]
* self.image_size[0]
* 3
)
float32_array_bytes = int(np.prod(array_shape)) * np.dtype(np.float32).itemsize
estimated_float32_peak = (
2 * float32_array_bytes
+ float32_chunk_pixels * (1 + 4 + 4)
+ 256 * 1024**2
)
use_float64 = estimated_float64_peak <= available_bytes
gib = 1024**3
print(
f"Available memory: {available_bytes / gib:.2f} GiB; "
f"Estimated float64 peak: {estimated_float64_peak / gib:.2f} GiB; "
f"Estimated float32 peak: {estimated_float32_peak / gib:.2f} GiB"
)
if use_float64:
print("Using in-memory float64 calibration processing")
self._write_float64_dataset(
image_names,
calibration_data_path,
processed_data_path,
completion_marker_path,
array_shape,
)
else:
print("Using chunked float32 calibration processing")
self._write_chunked_float32_dataset(
image_names,
calibration_data_path,
processed_data_path,
completion_marker_path,
array_shape,
)
print(
f"Normalized calibration dataset saved with shape: "
f"{array_shape} to {calibration_data_path}"
)
print(
f"Processed calibration dataset saved with shape: "
f"{array_shape} to {processed_data_path}"
)
ONNX_Parser
import os
from hailo_sdk_client import ClientRunner
class ONNXParser:
def __init__(self, model_name, onnx_path, hailo_output_folder,
chosen_hw_arch, width, height):
self.model_name = model_name
self.onnx_path = onnx_path
self.chosen_hw_arch = chosen_hw_arch
self.width = width
self.height = height
self.hailo_output_folder = hailo_output_folder
def parse(self):
runner = ClientRunner(hw_arch=self.chosen_hw_arch)
start_node_names = ["/model.0/conv/Conv"]
end_node_names = [
"/model.23/cv2.0/cv2.0.2/Conv",
"/model.23/cv3.0/cv3.0.2/Conv",
"/model.23/cv2.1/cv2.1.2/Conv",
"/model.23/cv3.1/cv3.1.2/Conv",
"/model.23/cv2.2/cv2.2.2/Conv",
"/model.23/cv3.2/cv3.2.2/Conv",
]
try:
runner.translate_onnx_model(
self.onnx_path,
self.model_name,
start_node_names=start_node_names,
end_node_names=end_node_names,
net_input_shapes={"/model.0/conv/Conv": [1, 3, self.width, self.height]},
)
print("Model translation successful.")
except Exception as e:
print(f"Error during model translation: {e}")
raise
resulting_hailo_model_har_name = (
f"{self.model_name}_hailo_model_{self.width}x{self.height}.har"
)
resulting_hailo_model_har_name = os.path.join(self.hailo_output_folder, resulting_hailo_model_har_name)
try:
runner.save_har(resulting_hailo_model_har_name)
print(f"HAR file saved as: {resulting_hailo_model_har_name}")
except Exception as e:
print(f"Error saving HAR file: {e}")
Modell_Optimizer
import os
from hailo_sdk_client import ClientRunner
class ModelOptimizer:
def __init__(self, model_name, hailo_output_folder, hailo_model_har_path, calibration_dataset_path,
nms_config_path, width, height, compression_level,
batch_size_optimization, calibset_size, optimization_level,
batch_size_flavor=None,
fineTuning_policy="disabled", batch_size_fineTuning=None,
epochs_fineTuning=None,
adaround_policy="disabled", batch_size_adaround=None):
self.hailo_output_folder = hailo_output_folder
self.hailo_model_har_path = hailo_model_har_path
self.calibration_dataset_path = calibration_dataset_path
self.nms_config_path = nms_config_path
self.model_name = model_name
self.width = width
self.height = height
self.calibset_size = calibset_size
self.batch_size_flavor = batch_size_flavor
self.batch_size_optimization = batch_size_optimization
self.batch_size_fineTuning = batch_size_fineTuning
self.batch_size_adaround = batch_size_adaround
self.optimization_level = optimization_level
self.compression_level = compression_level
self.fineTuning_policy = fineTuning_policy
self.adaround_policy = adaround_policy
self.epochs_fineTuning = epochs_fineTuning
def optimize(self):
if not os.path.isfile(self.hailo_model_har_path):
raise FileNotFoundError(
f"Hailo model HAR not found: {self.hailo_model_har_path}"
)
if not os.path.isfile(self.calibration_dataset_path):
raise FileNotFoundError(
f"Calibration dataset not found: {self.calibration_dataset_path}"
)
if not os.path.isfile(self.nms_config_path):
raise FileNotFoundError(f"NMS config not found: {self.nms_config_path}")
if self.fineTuning_policy == "enabled" and self.batch_size_fineTuning is None:
raise ValueError(
"batch_size_fineTuning is required when fineTuning_policy is enabled"
)
if self.fineTuning_policy == "enabled" and self.epochs_fineTuning is None:
raise ValueError(
"epochs_fineTuning is required when fineTuning_policy is enabled"
)
if self.adaround_policy == "enabled" and self.batch_size_adaround is None:
raise ValueError(
"batch_size_adaround is required when adaround_policy is enabled"
)
runner = ClientRunner(har=self.hailo_model_har_path)
model_script = (
"normalization1 = normalization([0.0, 0.0, 0.0], [255.0, 255.0, 255.0])\n"
"change_output_activation(conv120, sigmoid)\n"
"change_output_activation(conv143, sigmoid)\n"
"change_output_activation(conv165, sigmoid)\n"
f'nms_postprocess("{self.nms_config_path}", meta_arch=yolov8, engine=cpu)\n'
)
flavor_batch_size = (
f", batch_size={self.batch_size_flavor}"
if self.batch_size_flavor is not None
else ""
)
model_script += (
f"model_optimization_flavor(optimization_level={self.optimization_level}, "
f"compression_level={self.compression_level}{flavor_batch_size})\n"
f"model_optimization_config(calibration, calibset_size={self.calibset_size}, "
f"batch_size={self.batch_size_optimization})\n"
)
if self.fineTuning_policy != "disabled":
fine_tuning_options = [f"dataset_size={self.calibset_size}"]
if self.epochs_fineTuning is not None:
fine_tuning_options.append(f"epochs={self.epochs_fineTuning}")
if self.batch_size_fineTuning is not None:
fine_tuning_options.append(f"batch_size={self.batch_size_fineTuning}")
fine_tuning_options_text = (
f", {', '.join(fine_tuning_options)}" if fine_tuning_options else ""
)
model_script += (
f"post_quantization_optimization(finetune, policy={self.fineTuning_policy}, "
"loss_layer_names=[conv113, conv136, conv159, conv117, conv140, conv162, conv120, conv143, conv165], "
"loss_types=[l2rel, l2rel, l2rel, l2rel, l2rel, l2rel, ce, ce, ce], "
f"loss_factors=[1, 1, 1, 2, 2, 2, 2, 2, 2]{fine_tuning_options_text})\n"
)
if self.adaround_policy != "disabled":
adaround_batch_size = (
f", batch_size={self.batch_size_adaround}"
if self.batch_size_adaround is not None
else ""
)
model_script += (
f"post_quantization_optimization(adaround, policy={self.adaround_policy}"
f"{adaround_batch_size})\n"
)
runner.load_model_script(model_script)
runner.optimize(self.calibration_dataset_path)
quantized_model_output_har_path = os.path.join(
self.hailo_output_folder,
f"{self.model_name}_quantized_optimized_model_"
f"{self.width}x{self.height}_{self.calibset_size}_{self.batch_size_optimization}_"
f"{self.optimization_level}_{self.compression_level}.har",
)
#Output Format: num_calibration_images, batch_size, optimization_level, compression_level
runner.save_har(quantized_model_output_har_path)
print(f"Quantized HAR file saved to: {quantized_model_output_har_path}")
Compiling
import os
from hailo_sdk_client import ClientRunner
class ModelCompiler:
def __init__(self, model_name, quantized_model_har_path, output_dir,
width, height, num_calibration_images, batch_size,
optimization_level, compression_level, compiler_optimization_level):
self.model_name = model_name
self.quantized_model_har_path = quantized_model_har_path
self.output_dir = output_dir
self.width = width
self.height = height
self.num_calibration_images = num_calibration_images
self.batch_size = batch_size
self.optimization_level = optimization_level
self.compression_level = compression_level
self.compiler_optimization_level = compiler_optimization_level
def compile(self):
if not os.path.isfile(self.quantized_model_har_path):
raise FileNotFoundError(
f"Quantized model HAR not found: {self.quantized_model_har_path}"
)
runner = ClientRunner(har=self.quantized_model_har_path)
model_script = (
f"performance_param(compiler_optimization_level={self.compiler_optimization_level})\n"
)
runner.load_model_script(model_script)
print("[info] ClientRunner initialized successfully.")
try:
hef = runner.compile()
print("[info] Compilation completed successfully.")
except Exception as e:
print(f"[error] Failed to compile the model: {e}")
raise
os.makedirs(self.output_dir, exist_ok=True)
hef_path = os.path.join(
self.output_dir,
f"{self.model_name}_Finished_{self.width}x{self.height}_"
f"{self.num_calibration_images}_{self.batch_size}_"
f"{self.optimization_level}_{self.compression_level}.hef",
)
with open(hef_path, "wb") as hef_file:
hef_file.write(hef)
print(f"Compiled HEF saved to: {hef_path}")
return hef_path
NMS_Config
{
"nms_scores_th": 0.65,
"nms_iou_th": 0.5,
"image_dims": [
640,
640
],
"max_proposals_per_class": 25,
"classes": 10,
"regression_length": 16,
"background_removal": false,
"background_removal_index": 0,
"bbox_decoders": [
{
"name": "bbox_decoder117",
"stride": 8,
"reg_layer": "conv117",
"cls_layer": "conv120"
},
{
"name": "bbox_decoder140",
"stride": 16,
"reg_layer": "conv140",
"cls_layer": "conv143"
},
{
"name": "bbox_decoder162",
"stride": 32,
"reg_layer": "conv162",
"cls_layer": "conv165"
}
]
}
The first Python snippets I gathered from doing research and running them one after one. Recently I used the help of AI to get my initial Python Snipets into this Pipeline
