From 00a35063191b62121530a143afffc7d0c4cc0318 Mon Sep 17 00:00:00 2001 From: ff225 Date: Fri, 10 Jul 2026 10:05:59 +0200 Subject: [PATCH 1/2] refactor(client): unify HTTP clients on a shared http_client core (#46) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit The four camera client variants each duplicated ~15 helper functions (config loading, server discovery, TFLite interpreters, the binary protocol/payload, registration, split inference, result upload) โ€” a fix in http_client.py had to be replicated in four files. Make http_client.py the single source of truth and reduce the camera variants to thin wrappers that import the shared logic and keep only their own frame source and capture loop: - http_clientCAMlaptop / http_clientCAMpi: OpenCV webcam / Picamera2 loops - http_clientCAMlaptopBench / http_clientCAMpiBench: same, plus CLI overrides for benchmark sweeps (--id/--model/--res) and profiler export on shutdown Add http_client.set_model_config() so the benchmark variants can override the TFLite folder / input resolution (with submodel-prefix and layer auto-detection) before interpreters are preloaded. Hardware-only imports (cv2, picamera2) are guarded so the modules load for tooling off-device and fail with a clear error at camera open. The camera capture loops are preserved verbatim; only the duplicated helpers were removed (~2,300 fewer duplicated lines). Existing tests that import http_client pass unchanged. --- src/client/python/http_client.py | 40 ++ src/client/python/http_clientCAMlaptop.py | 642 +---------------- .../python/http_clientCAMlaptopBench.py | 674 ++---------------- src/client/python/http_clientCAMpi.py | 629 +--------------- src/client/python/http_clientCAMpiBench.py | 671 ++--------------- 5 files changed, 207 insertions(+), 2449 deletions(-) diff --git a/src/client/python/http_client.py b/src/client/python/http_client.py index 8d4995f8..fbfbdae0 100644 --- a/src/client/python/http_client.py +++ b/src/client/python/http_client.py @@ -905,6 +905,46 @@ def compute_model_hash(): return _cached_model_hash +def set_model_config(tflite_subdir=None, input_size=None): + """Override the model directory and/or input resolution before startup. + + Used by the benchmark client variants to sweep models/resolutions from the + command line. When a TFLite subdirectory is given, the submodel prefix and + last offloading layer are auto-detected from its ``*.tflite`` files. Any + cache derived from the previous model/resolution is invalidated so that the + next call rebuilds it. + """ + global TFLITE_DIR, SUBMODEL_PREFIX, LAST_OFFLOADING_LAYER + global INPUT_HEIGHT, INPUT_WIDTH + global _interpreter_cache, _cached_model_hash, _cached_image_rgb + global _cached_rgb565_buffer, _send_image_buffer + + if tflite_subdir is not None: + TFLITE_DIR = SCRIPT_DIR / tflite_subdir + tflite_files = sorted(TFLITE_DIR.glob("*.tflite")) + if tflite_files: + SUBMODEL_PREFIX = tflite_files[0].stem.rsplit("_", 1)[0] + LAST_OFFLOADING_LAYER = len(tflite_files) - 1 + print(f"๐Ÿ” Detected submodel prefix: {SUBMODEL_PREFIX}") + print( + f"๐Ÿ“Š Detected layers: {len(tflite_files)} " + f"(last offloading layer: {LAST_OFFLOADING_LAYER})" + ) + else: + print(f"โŒ WARNING: TFLite folder is empty: {TFLITE_DIR}") + + if input_size is not None: + INPUT_HEIGHT = input_size + INPUT_WIDTH = input_size + + # Invalidate caches derived from the previous model/resolution. + _interpreter_cache = {} + _cached_model_hash = None + _cached_image_rgb = None + _cached_rgb565_buffer = None + _send_image_buffer = None + + # ----- # MAIN # ----- diff --git a/src/client/python/http_clientCAMlaptop.py b/src/client/python/http_clientCAMlaptop.py index 95d743de..3ff5fbcd 100644 --- a/src/client/python/http_clientCAMlaptop.py +++ b/src/client/python/http_clientCAMlaptop.py @@ -1,97 +1,50 @@ -import numpy as np -from PIL import Image -import struct -import gzip -import requests -from requests.adapters import HTTPAdapter -from urllib3.util.retry import Retry -import time -import random +"""SCIoT HTTP client โ€” laptop webcam variant. + +Thin wrapper around ``http_client`` (the canonical client). All shared logic โ€” +config, server discovery, TFLite interpreters, the binary protocol/payload, +registration, split inference and result upload โ€” is imported from there. This +module only adds the laptop webcam frame source and its capture loop. +""" + import os import sys -import yaml -from pathlib import Path -from delay_simulator import DelaySimulator -import socket -import hashlib -import ipaddress -from concurrent.futures import ThreadPoolExecutor, as_completed -import cProfile -from profiler import AdvancedProfiler +import time -# Get the directory where this script is located -SCRIPT_DIR = Path(__file__).resolve().parent -PROJECT_ROOT = SCRIPT_DIR.parent.parent.parent -CONFIG_PATH = SCRIPT_DIR / "http_config.yaml" +import numpy as np +from PIL import Image +from concurrent.futures import ThreadPoolExecutor + +from profiler import AdvancedProfiler -# Load configuration from YAML file -with open(CONFIG_PATH, "r") as f: - config = yaml.safe_load(f) +import http_client as base +# Optional positional override of the device id (kept from the original client), +# applied before importing the shared constants below so they reflect it. if len(sys.argv) > 1: - config["client"]["device_id"] = sys.argv[1] - -DEVICE_ID = config["client"]["device_id"] -SERVER_HOST = config["http"]["server_host"] -SERVER_PORT = config["http"]["server_port"] - -# ===================================================== -# OPTIMIZATION: Use requests.Session for connection pooling. -# This reuses TCP connections across requests (HTTP keep-alive) -# instead of creating a new TCP connection for each request. -# ===================================================== -session = requests.Session() - -# ===================================================== -# OPTIMIZATION: Configure HTTPAdapter with retry policy and pool sizing. -# - pool_connections / pool_maxsize sized for a single-server client -# - Lightweight retry (2 attempts) with exponential backoff on transient errors -# ===================================================== -_retry_strategy = Retry( - total=2, - backoff_factor=0.1, - status_forcelist=[502, 503, 504], - allowed_methods=["GET", "POST"], + base.set_device_id(sys.argv[1]) + +from http_client import ( + load_config, + SERVER, + DEVICE_ID, + TFLITE_DIR, + INPUT_HEIGHT, + INPUT_WIDTH, + LAST_OFFLOADING_LAYER, + register_device, + generate_message_id, + get_offloading_layer, + run_split_inference, + send_inference_result, + _preload_interpreters, + _build_rgb565_buffer, ) -_adapter = HTTPAdapter( - pool_connections=1, - pool_maxsize=4, - max_retries=_retry_strategy, -) -session.mount("http://", _adapter) -session.mount("https://", _adapter) - -# ===================================================== -# OPTIMIZATION: Dedicated discovery session with a larger pool -# for parallel subnet scanning (up to 50 threads). -# Avoids ephemeral socket churn from bare requests.post/get calls. -# ===================================================== -_discovery_session = requests.Session() -_discovery_adapter = HTTPAdapter(pool_connections=10, pool_maxsize=50) -_discovery_session.mount("http://", _discovery_adapter) -_discovery_session.mount("https://", _discovery_adapter) - -# ===================================================== -# CONFIGURATION: Optional gzip compression for payloads. -# Saves bandwidth on constrained IoT networks at the cost of -# ~1-2 ms CPU per request. Toggle via http_config.yaml. -# ===================================================== -COMPRESSION_ENABLED = config.get("compression", {}).get("enabled", False) - -_tf = None -_cv2 = None - -def _load_tensorflow(): - """Load TensorFlow lazily so startup progress is visible on Apple Silicon.""" - global _tf - if _tf is None: - print("Loading TensorFlow/TFLite runtime...", flush=True) - import tensorflow as tensorflow_module +# Offloading-layer requests run in a dedicated single-worker pool, overlapped +# with image acquisition (see the main loop). +_request_pool = ThreadPoolExecutor(max_workers=1) - _tf = tensorflow_module - print(f"TensorFlow loaded: {_tf.__version__}", flush=True) - return _tf +_cv2 = None def _load_cv2(): @@ -106,525 +59,6 @@ def _load_cv2(): return _cv2 -def load_config(config_filename="http_config.yaml"): - """Legge il file di configurazione YAML usando il percorso assoluto.""" - # Trova la cartella esatta dove risiede QUESTO script (http_client.py) - script_dir = os.path.dirname(os.path.abspath(__file__)) - # Unisce la cartella al nome del file - config_path = os.path.join(script_dir, config_filename) - - if os.path.exists(config_path): - with open(config_path, "r") as file: - return yaml.safe_load(file) - else: - print(f"โš ๏ธ [ATTENZIONE] File di configurazione non trovato in: {config_path}") - print("โš ๏ธ Uso le impostazioni di default (cProfile disabilitato).") - return {} - -# Discovery functions -def get_local_network(): - """Get the local network IP range for scanning.""" - try: - # Get local IP address - s = socket.socket(socket.AF_INET, socket.SOCK_DGRAM) - s.connect(("8.8.8.8", 80)) - local_ip = s.getsockname()[0] - s.close() - - # Convert to network/24 (assuming typical home network) - network = ipaddress.IPv4Network(f"{local_ip}/24", strict=False) - return network, local_ip - except Exception as e: - print(f"Error getting local network: {e}") - return None, None - - -def check_server_at_ip(ip, port, timeout=1.0): - """Check if a server is responding at the given IP:port.""" - try: - # First try TCP connection - sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM) - sock.settimeout(timeout) - result = sock.connect_ex((str(ip), port)) - sock.close() - - if result == 0: - # Port is open, verify it's our server by trying multiple methods - # OPTIMIZATION: Use _discovery_session instead of bare requests.*() - # to reuse TCP connections across the 50-thread parallel scan. - try: - # Try 1: POST to registration endpoint (proper way) - test_url = ( - f"http://{ip}:{port}{config['http']['endpoints']['registration']}" - ) - response = _discovery_session.post( - test_url, json={"device_id": "discovery_test"}, timeout=2 - ) - if response.status_code in [ - 200, - 201, - 400, - 409, - ]: # Any reasonable server response - print( - f" Found server at {ip}:{port} (registration: {response.status_code})" - ) - return str(ip) - except requests.exceptions.RequestException: - pass - - try: - # Try 2: GET request to root or any endpoint - response = _discovery_session.get(f"http://{ip}:{port}/", timeout=2) - if response.status_code < 500: # Any response except server error - print( - f" Found server at {ip}:{port} (root: {response.status_code})" - ) - return str(ip) - except requests.exceptions.RequestException: - pass - - try: - # Try 3: Check offloading_layer endpoint - test_url = f"http://{ip}:{port}{config['http']['endpoints']['offloading_layer']}" - response = _discovery_session.get(test_url, timeout=2) - if response.status_code in [200, 400, 404, 405]: - print( - f" Found server at {ip}:{port} (offload endpoint: {response.status_code})" - ) - return str(ip) - except requests.exceptions.RequestException: - pass - - return None - except Exception as e: - return None - - -def discover_server(port): - """Scan local network to discover server.""" - print("=" * 60) - print("SERVER DISCOVERY MODE") - print("=" * 60) - - # First, try common local addresses - print(f"Checking localhost and common addresses on port {port}...\n") - common_hosts = ["127.0.0.1", "localhost", "0.0.0.0"] - - for host in common_hosts: - print(f" Trying {host}:{port}...") - result = check_server_at_ip(host, port, timeout=2.0) - if result: - print(f"\nServer found at {result}:{port}") - return result - - print("\nNo server on localhost, scanning local network...\n") - - network, local_ip = get_local_network() - if not network: - print("ERROR: Could not determine local network") - return None - - print(f"Local IP: {local_ip}") - print(f"Scanning network: {network}") - print(f"This may take 10-30 seconds...\n") - - # Scan network with threading for speed - found_servers = [] - total_hosts = network.num_addresses - 2 # Exclude network and broadcast - checked = 0 - - with ThreadPoolExecutor(max_workers=50) as executor: - futures = { - executor.submit(check_server_at_ip, ip, port): ip - for ip in network.hosts() - if str(ip) != local_ip - } - - for future in as_completed(futures): - checked += 1 - if checked % 25 == 0: - print(f"Progress: {checked}/{total_hosts} hosts checked...") - - result = future.result() - if result: - found_servers.append(result) - print(f"\nSERVER FOUND: {result}:{port}") - # Cancel remaining futures for faster completion - for f in futures: - f.cancel() - break - - print(f"\nScan complete: {checked}/{total_hosts} hosts checked") - - if found_servers: - selected = found_servers[0] - print(f"\nSelected server: {selected}:{port}") - return selected - else: - print("\nNo server found on local network") - print(" Please check:") - print(" 1. Server is running") - print(" 2. Server is on the same network") - print(f" 3. Server is listening on port {port}") - return None - - -# Run discovery if SERVER_HOST is None -if SERVER_HOST is None or SERVER_HOST == "None" or SERVER_HOST == "": - print("\nSERVER_HOST is not configured - starting discovery...\n") - discovered_host = discover_server(SERVER_PORT) - - if discovered_host: - SERVER_HOST = discovered_host - print(f"\nUsing discovered server: {SERVER_HOST}:{SERVER_PORT}\n") - else: - print("\nWARNING: No server discovered - will run in LOCAL-ONLY mode") - SERVER_HOST = "localhost" # Fallback to avoid errors - -SERVER = f"http://{SERVER_HOST}:{SERVER_PORT}" - -MODEL_CONFIG = config["model"] -INPUT_HEIGHT = MODEL_CONFIG["input_height"] -INPUT_WIDTH = MODEL_CONFIG["input_width"] -LAST_OFFLOADING_LAYER = MODEL_CONFIG["last_offloading_layer"] - -IMAGE_PATH = SCRIPT_DIR / MODEL_CONFIG["image_name"] -TFLITE_DIR = SCRIPT_DIR / MODEL_CONFIG["tflite_subdir"] -SUBMODEL_PREFIX = MODEL_CONFIG["submodel_prefix"] - -ENDPOINTS = config["http"]["endpoints"] - -# ===================================================== -# OPTIMIZATION: Configurable device input sending. -# When disabled, skips the send_image() POST entirely, saving -# ~18 KB bandwidth and one round-trip per iteration. -# ===================================================== -SEND_DEVICE_INPUT = config["http"].get("send_device_input", True) - -# ===================================================== -# OPTIMIZATION: Thread pool for overlapping the GET offloading_layer -# request with image preparation / inference setup. -# Saves one round-trip of latency per iteration. -# ===================================================== -_request_pool = ThreadPoolExecutor(max_workers=1) - -# ===================================================== -# OPTIMIZATION: Pre-compute the device ID header bytes once. -# The device ID never changes at runtime, so we encode it and pack -# its length prefix once instead of repeating this work every call. -# ===================================================== -_DEVICE_ID_BYTES = DEVICE_ID.encode("ascii") -_DEVICE_ID_HEADER = struct.pack("i", len(_DEVICE_ID_BYTES)) + _DEVICE_ID_BYTES - -# Initialize delay simulators -DELAY_CONFIG = config.get("delay_simulation", {}) -computation_delay = DelaySimulator(DELAY_CONFIG.get("device_computation")) -network_delay = DelaySimulator(DELAY_CONFIG.get("network")) - -if computation_delay.enabled: - print(f"Computation delay enabled: {computation_delay.get_delay_info()}") -if network_delay.enabled: - print(f"Network delay enabled: {network_delay.get_delay_info()}") - -# ===================================================== -# OPTIMIZATION: Pre-load and cache TFLite interpreters. -# Interpreters are expensive to create (file I/O + memory allocation). -# We create them once at startup and reuse across inference cycles. -# ===================================================== -_interpreter_cache = {} - - -def _get_interpreter(layer_index): - """Get a cached TFLite interpreter for the given layer index.""" - if layer_index not in _interpreter_cache: - tensorflow_module = _load_tensorflow() - model_path = str(TFLITE_DIR / f"{SUBMODEL_PREFIX}_{layer_index}.tflite") - interpreter = tensorflow_module.lite.Interpreter(model_path=model_path) - interpreter.allocate_tensors() - _interpreter_cache[layer_index] = { - "interpreter": interpreter, - "input_details": interpreter.get_input_details(), - "output_details": interpreter.get_output_details(), - } - return _interpreter_cache[layer_index] - - -def _preload_interpreters(): - """Pre-load all TFLite interpreters at startup.""" - print(f"Pre-loading {LAST_OFFLOADING_LAYER + 1} TFLite interpreters...", flush=True) - for i in range(LAST_OFFLOADING_LAYER + 1): - _get_interpreter(i) - if i == 0 or (i + 1) % 10 == 0 or i == LAST_OFFLOADING_LAYER: - print(f" Loaded interpreter {i + 1}/{LAST_OFFLOADING_LAYER + 1}", flush=True) - print("All interpreters loaded and cached.", flush=True) - - -# Function to generate a random message ID -def generate_message_id(): - return "".join(random.choices("ABCDEFGHIJKLMNOPQRSTUVWXYZ0123456789", k=4)) - - -# --------------------- -# Registration -# --------------------- -def register_device(): - url = f"{SERVER}{ENDPOINTS['registration']}" - model_hash = compute_model_hash() - payload = {"device_id": DEVICE_ID, "model_hash": model_hash} - try: - r = session.post(url, json=payload, timeout=5) - if r.status_code == 404: - print("WARNING: Server does not have this model - running LOCAL-ONLY") - return False - return r.status_code == 200 - except requests.exceptions.RequestException as e: - return False - - -# ------------------------------------------------ -# Convert image to RGB565 and send to server -# ------------------------------------------------ - -# ===================================================== -# OPTIMIZATION: Pre-compute static RGB565 data once, reuse across calls. -# Only the timestamp header needs to change per call. -# The send buffer is pre-allocated so send_image() only writes the -# 8-byte timestamp in-place instead of copying the entire image. -# ===================================================== -_cached_rgb565_buffer = None -_send_image_buffer = None - - -def _build_rgb565_buffer(): - """Build the RGB565 buffer from the image (done once at startup).""" - global _cached_rgb565_buffer, _send_image_buffer - img = Image.open(str(IMAGE_PATH)).resize((INPUT_HEIGHT, INPUT_WIDTH)).convert("RGB") - data = np.array(img) - # OPTIMIZATION: Vectorized NumPy conversion instead of nested Python loops. - # ~50-100x faster for typical image sizes (e.g. 96x96). - r = data[:, :, 0].astype(np.uint16) - g = data[:, :, 1].astype(np.uint16) - b = data[:, :, 2].astype(np.uint16) - rgb565 = ((r >> 3) << 11) | ((g >> 2) << 5) | (b >> 3) - _cached_rgb565_buffer = rgb565.astype(">u2").tobytes() - # Pre-allocate the send buffer (8 bytes timestamp + image data). - # send_image() will only overwrite the first 8 bytes each call. - _send_image_buffer = bytearray(8 + len(_cached_rgb565_buffer)) - _send_image_buffer[8:] = _cached_rgb565_buffer - - -def send_image(): - url = f"{SERVER}{ENDPOINTS['device_input']}" - - # OPTIMIZATION: Write timestamp in-place into the pre-allocated buffer. - # Avoids allocating and copying ~18 KB on every call. - struct.pack_into("d", _send_image_buffer, 0, time.time()) - - headers = {"Content-Type": "application/octet-stream"} - if COMPRESSION_ENABLED: - payload = gzip.compress(bytes(_send_image_buffer)) - headers["Content-Encoding"] = "gzip" - else: - payload = _send_image_buffer - - try: - r = session.post( - url, - data=payload, - headers=headers, - timeout=30, - ) - return True - except requests.exceptions.RequestException as e: - return False - - -# ---------------------------- -# Request Best Offloading Layer -# ---------------------------- - - -def get_offloading_layer(): - url = f"{SERVER}{ENDPOINTS['offloading_layer']}?device_id={DEVICE_ID}" - try: - r = session.get(url, timeout=10) - if r.status_code == 200: - best_layer = r.json().get("offloading_layer_index", LAST_OFFLOADING_LAYER) - - if best_layer > LAST_OFFLOADING_LAYER: - best_layer = LAST_OFFLOADING_LAYER - return best_layer - else: - return LAST_OFFLOADING_LAYER - - except requests.exceptions.RequestException: - return LAST_OFFLOADING_LAYER - - -# TEST: Simulate random best offloading layer change -def get_offloading_layer_random(): - url = f"{SERVER}{ENDPOINTS['offloading_layer']}" - r = session.get(url) - if r.status_code == 200: - best_layer = r.json().get("offloading_layer_index", LAST_OFFLOADING_LAYER) - print("Best layer received:", best_layer) - return random.randint(0, LAST_OFFLOADING_LAYER) - else: - print("Error requesting layer:", r.status_code) - return LAST_OFFLOADING_LAYER - - -# ---------------------- -# Send output -# --------------------- - -# ===================================================== -# OPTIMIZATION: Cache the loaded and processed image array. -# The image doesn't change between iterations, so we load it once. -# ===================================================== -_cached_image_rgb = None - - -def load_image_rgb(path): - global _cached_image_rgb - if _cached_image_rgb is None: - img = Image.open(str(path)).resize((INPUT_HEIGHT, INPUT_WIDTH)).convert("RGB") - img_np = np.asarray(img).astype(np.float32) / 255.0 - img_np = np.expand_dims(img_np, axis=0) # [1,H,W,3] - _cached_image_rgb = img_np - return _cached_image_rgb - - -def run_split_inference(image, tflite_dir, stop_layer): - input_data = image - inference_times = [] - - # Handle -1 as "run all layers until the end" - if stop_layer == -1: - stop_layer = LAST_OFFLOADING_LAYER - - for i in range(stop_layer + 1): - # OPTIMIZATION: Use cached interpreter instead of recreating each time - cached = _get_interpreter(i) - interpreter = cached["interpreter"] - input_details = cached["input_details"] - output_details = cached["output_details"] - - input_data = input_data.astype(input_details[0]["dtype"]) - interpreter.set_tensor(input_details[0]["index"], input_data) - - t0 = time.time() - - # Apply artificial computation delay - if computation_delay.enabled: - delay = computation_delay.apply_delay() - - interpreter.invoke() - t1 = time.time() - inference_times.append(t1 - t0) - input_data = interpreter.get_tensor(output_details[0]["index"]) - return input_data, inference_times - - -def send_inference_result( - output_data, inference_times, layer_index, message_id, acq_time_ms -): - # Apply network delay before sending - send_start = time.time() - if network_delay.enabled: - delay = network_delay.apply_delay() - - url = f"{SERVER}{ENDPOINTS['device_inference_result']}" - timestamp = time.time() - - # Build the binary payload. - # _DEVICE_ID_HEADER is pre-computed at module level (length prefix + encoded bytes). - # The two adjacent int/float fields (layer_index, acq_time) are packed in a single call. - output_bytes = output_data.tobytes() - times_bytes = np.array(inference_times, dtype=np.float32).tobytes() - - # ===================================================== - # OPTIMIZATION: Pre-allocate the buffer at full size and use pack_into() - # to write fields in-place, avoiding 6+ incremental bytearray copies. - # ===================================================== - msg_id_bytes = message_id.encode("ascii").ljust(4, b"\x00") - header_size = ( - 8 # timestamp (d) - + len(_DEVICE_ID_HEADER) # device id length prefix + bytes - + 4 # message_id (4 ascii bytes) - + 4 - + 4 # layer_index (i) + acq_time (f) - + 4 # output length prefix (I) - ) - total_size = header_size + len(output_bytes) + 4 + len(times_bytes) - buffer = bytearray(total_size) - - offset = 0 - struct.pack_into("d", buffer, offset, timestamp) - offset += 8 - buffer[offset : offset + len(_DEVICE_ID_HEADER)] = _DEVICE_ID_HEADER - offset += len(_DEVICE_ID_HEADER) - buffer[offset : offset + 4] = msg_id_bytes - offset += 4 - struct.pack_into("if", buffer, offset, layer_index, acq_time_ms / 1000.0) - offset += 8 - struct.pack_into("I", buffer, offset, len(output_bytes)) - offset += 4 - buffer[offset : offset + len(output_bytes)] = output_bytes - offset += len(output_bytes) - struct.pack_into("i", buffer, offset, len(times_bytes)) - offset += 4 - buffer[offset : offset + len(times_bytes)] = times_bytes - - # ===================================================== - # OPTIMIZATION: Optional gzip compression to reduce bandwidth. - # Controlled via http_config.yaml -> compression.enabled - # ===================================================== - headers = {"Content-Type": "application/octet-stream"} - if COMPRESSION_ENABLED: - payload = gzip.compress(bytes(buffer)) - headers["Content-Encoding"] = "gzip" - else: - payload = buffer - - try: - r = session.post( - url, - data=payload, - headers=headers, - timeout=5, - ) - send_end = time.time() - network_time = send_end - send_start - data_size = len(payload) - network_speed = ( - (data_size / network_time) / 1024 if network_time > 0 else 0 - ) # KB/s - return True, network_time, network_speed, data_size - except requests.exceptions.RequestException as e: - return False, 0, 0, 0 - - -# ===================================================== -# OPTIMIZATION: Cache the model hash after first computation. -# The TFLite model files never change at runtime, so we avoid -# re-reading all files from disk on every registration/reconnect. -# ===================================================== -_cached_model_hash = None - - -def compute_model_hash(): - global _cached_model_hash - if _cached_model_hash is None: - hasher = hashlib.md5() - for tflite_file in sorted(TFLITE_DIR.glob("*.tflite")): - with open(tflite_file, "rb") as f: - hasher.update(f.read()) - _cached_model_hash = hasher.hexdigest() - return _cached_model_hash - - def open_laptop_camera(camera_index=0): cv2_module = _load_cv2() print(f"Opening laptop webcam at index {camera_index}...", flush=True) diff --git a/src/client/python/http_clientCAMlaptopBench.py b/src/client/python/http_clientCAMlaptopBench.py index 94ae5d49..6079a981 100644 --- a/src/client/python/http_clientCAMlaptopBench.py +++ b/src/client/python/http_clientCAMlaptopBench.py @@ -1,643 +1,65 @@ -import tensorflow as tf -import numpy as np -from PIL import Image -import struct -import gzip -import requests -from requests.adapters import HTTPAdapter -from urllib3.util.retry import Retry -import time -import random +"""SCIoT HTTP client โ€” laptop webcam benchmark/sweep variant. + +Thin wrapper around ``http_client`` (the canonical client). Adds CLI overrides +(``--id``/``--model``/``--res``) for benchmark sweeps and exports profiler +reports on shutdown; all shared logic is imported from ``http_client``. +""" + import os -import yaml -from pathlib import Path -from delay_simulator import DelaySimulator -import socket -import hashlib -import ipaddress -from concurrent.futures import ThreadPoolExecutor, as_completed -import cProfile -from profiler import AdvancedProfiler -import cv2 import sys +import time import argparse -from pathlib import Path +import numpy as np +from PIL import Image +from concurrent.futures import ThreadPoolExecutor -# Get the directory where this script is located -SCRIPT_DIR = Path(__file__).resolve().parent -PROJECT_ROOT = SCRIPT_DIR.parent.parent.parent -CONFIG_PATH = SCRIPT_DIR / "http_config.yaml" +# OpenCV ships only with the client-laptop extra (opencv-python). Import it +# defensively so this module still loads for tooling/analysis without that +# extra; the camera open in main() fails with a clear error if cv2 is missing. +try: + import cv2 +except ModuleNotFoundError: + cv2 = None -# Load configuration from YAML file -with open(CONFIG_PATH, "r") as f: - config = yaml.safe_load(f) +from profiler import AdvancedProfiler +import http_client as base +# CLI overrides, applied before importing the shared constants below so they +# reflect the selected device id / model / resolution. parser = argparse.ArgumentParser(description="SCIoT Edge Client") - -# Ora Python sa cos'รจ "parser" e puรฒ aggiungerci gli argomenti -parser.add_argument("--id", type=str, help="Sovrascrive il Device ID") -parser.add_argument("--model", type=str, help="Sovrascrive la cartella TFLite") -parser.add_argument("--res", type=int, help="Sovrascrive la risoluzione di input") - -args, unknown = parser.parse_known_args() - +parser.add_argument("--id", type=str, help="Override the device ID") +parser.add_argument("--model", type=str, help="Override the TFLite folder") +parser.add_argument("--res", type=int, help="Override the input resolution") +args, _unknown = parser.parse_known_args() if args.id: - config["client"]["device_id"] = args.id -if args.model: - config["model"]["tflite_subdir"] = args.model - - # โ”€โ”€โ”€ RILEVAMENTO AUTOMATICO PREFISSO E LAYER โ”€โ”€โ”€ - model_dir = Path(__file__).resolve().parent / args.model - tflite_files = list(model_dir.glob("*.tflite")) - - if tflite_files: - # 1. Rileva il nome esatto (es. "submodel") - prefix_trovato = tflite_files[0].stem.rsplit('_', 1)[0] - config["model"]["submodel_prefix"] = prefix_trovato - print(f"๐Ÿ” Prefisso rilevato automaticamente: {prefix_trovato}") - - # 2. Conta quanti file ci sono per evitare i crash! - numero_max_layer = len(tflite_files) - 1 - config["model"]["last_offloading_layer"] = numero_max_layer - print(f"๐Ÿ“Š Layer rilevati: {len(tflite_files)} (Ultimo: {numero_max_layer})") - else: - print(f"โŒ ATTENZIONE: La cartella {args.model} รจ vuota!") -if args.res: - config["model"]["input_height"] = args.res - config["model"]["input_width"] = args.res - -LAST_OFFLOADING_LAYER = config["model"]["last_offloading_layer"] - -DEVICE_ID = config["client"]["device_id"] -SERVER_HOST = config["http"]["server_host"] -SERVER_PORT = config["http"]["server_port"] - -# ===================================================== -# OPTIMIZATION: Use requests.Session for connection pooling. -# This reuses TCP connections across requests (HTTP keep-alive) -# instead of creating a new TCP connection for each request. -# ===================================================== -session = requests.Session() - -# ===================================================== -# OPTIMIZATION: Configure HTTPAdapter with retry policy and pool sizing. -# - pool_connections / pool_maxsize sized for a single-server client -# - Lightweight retry (2 attempts) with exponential backoff on transient errors -# ===================================================== -_retry_strategy = Retry( - total=2, - backoff_factor=0.1, - status_forcelist=[502, 503, 504], - allowed_methods=["GET", "POST"], -) -_adapter = HTTPAdapter( - pool_connections=1, - pool_maxsize=4, - max_retries=_retry_strategy, + base.set_device_id(args.id) +if args.model or args.res: + base.set_model_config(tflite_subdir=args.model, input_size=args.res) + +from http_client import ( + load_config, + IMAGE_PATH, + SERVER, + DEVICE_ID, + TFLITE_DIR, + INPUT_HEIGHT, + INPUT_WIDTH, + LAST_OFFLOADING_LAYER, + register_device, + generate_message_id, + get_offloading_layer, + run_split_inference, + send_inference_result, + _preload_interpreters, + _build_rgb565_buffer, + load_image_rgb, ) -session.mount("http://", _adapter) -session.mount("https://", _adapter) - -# ===================================================== -# OPTIMIZATION: Dedicated discovery session with a larger pool -# for parallel subnet scanning (up to 50 threads). -# Avoids ephemeral socket churn from bare requests.post/get calls. -# ===================================================== -_discovery_session = requests.Session() -_discovery_adapter = HTTPAdapter(pool_connections=10, pool_maxsize=50) -_discovery_session.mount("http://", _discovery_adapter) -_discovery_session.mount("https://", _discovery_adapter) - -# ===================================================== -# CONFIGURATION: Optional gzip compression for payloads. -# Saves bandwidth on constrained IoT networks at the cost of -# ~1-2 ms CPU per request. Toggle via http_config.yaml. -# ===================================================== -COMPRESSION_ENABLED = config.get("compression", {}).get("enabled", False) - - -def load_config(config_filename="http_config.yaml"): - """Legge il file di configurazione YAML usando il percorso assoluto.""" - # Trova la cartella esatta dove risiede QUESTO script (http_client.py) - script_dir = os.path.dirname(os.path.abspath(__file__)) - # Unisce la cartella al nome del file - config_path = os.path.join(script_dir, config_filename) - - if os.path.exists(config_path): - with open(config_path, "r") as file: - return yaml.safe_load(file) - else: - print(f"โš ๏ธ [ATTENZIONE] File di configurazione non trovato in: {config_path}") - print("โš ๏ธ Uso le impostazioni di default (cProfile disabilitato).") - return {} - -# Discovery functions -def get_local_network(): - """Get the local network IP range for scanning.""" - try: - # Get local IP address - s = socket.socket(socket.AF_INET, socket.SOCK_DGRAM) - s.connect(("8.8.8.8", 80)) - local_ip = s.getsockname()[0] - s.close() - - # Convert to network/24 (assuming typical home network) - network = ipaddress.IPv4Network(f"{local_ip}/24", strict=False) - return network, local_ip - except Exception as e: - print(f"Error getting local network: {e}") - return None, None - - -def check_server_at_ip(ip, port, timeout=1.0): - """Check if a server is responding at the given IP:port.""" - try: - # First try TCP connection - sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM) - sock.settimeout(timeout) - result = sock.connect_ex((str(ip), port)) - sock.close() - - if result == 0: - # Port is open, verify it's our server by trying multiple methods - # OPTIMIZATION: Use _discovery_session instead of bare requests.*() - # to reuse TCP connections across the 50-thread parallel scan. - try: - # Try 1: POST to registration endpoint (proper way) - test_url = ( - f"http://{ip}:{port}{config['http']['endpoints']['registration']}" - ) - response = _discovery_session.post( - test_url, json={"device_id": "discovery_test"}, timeout=2 - ) - if response.status_code in [ - 200, - 201, - 400, - 409, - ]: # Any reasonable server response - print( - f" Found server at {ip}:{port} (registration: {response.status_code})" - ) - return str(ip) - except requests.exceptions.RequestException: - pass - try: - # Try 2: GET request to root or any endpoint - response = _discovery_session.get(f"http://{ip}:{port}/", timeout=2) - if response.status_code < 500: # Any response except server error - print( - f" Found server at {ip}:{port} (root: {response.status_code})" - ) - return str(ip) - except requests.exceptions.RequestException: - pass - - try: - # Try 3: Check offloading_layer endpoint - test_url = f"http://{ip}:{port}{config['http']['endpoints']['offloading_layer']}" - response = _discovery_session.get(test_url, timeout=2) - if response.status_code in [200, 400, 404, 405]: - print( - f" Found server at {ip}:{port} (offload endpoint: {response.status_code})" - ) - return str(ip) - except requests.exceptions.RequestException: - pass - - return None - except Exception as e: - return None - - -def discover_server(port): - """Scan local network to discover server.""" - print("=" * 60) - print("SERVER DISCOVERY MODE") - print("=" * 60) - - # First, try common local addresses - print(f"Checking localhost and common addresses on port {port}...\n") - common_hosts = ["127.0.0.1", "localhost", "0.0.0.0"] - - for host in common_hosts: - print(f" Trying {host}:{port}...") - result = check_server_at_ip(host, port, timeout=2.0) - if result: - print(f"\nServer found at {result}:{port}") - return result - - print("\nNo server on localhost, scanning local network...\n") - - network, local_ip = get_local_network() - if not network: - print("ERROR: Could not determine local network") - return None - - print(f"Local IP: {local_ip}") - print(f"Scanning network: {network}") - print(f"This may take 10-30 seconds...\n") - - # Scan network with threading for speed - found_servers = [] - total_hosts = network.num_addresses - 2 # Exclude network and broadcast - checked = 0 - - with ThreadPoolExecutor(max_workers=50) as executor: - futures = { - executor.submit(check_server_at_ip, ip, port): ip - for ip in network.hosts() - if str(ip) != local_ip - } - - for future in as_completed(futures): - checked += 1 - if checked % 25 == 0: - print(f"Progress: {checked}/{total_hosts} hosts checked...") - - result = future.result() - if result: - found_servers.append(result) - print(f"\nSERVER FOUND: {result}:{port}") - # Cancel remaining futures for faster completion - for f in futures: - f.cancel() - break - - print(f"\nScan complete: {checked}/{total_hosts} hosts checked") - - if found_servers: - selected = found_servers[0] - print(f"\nSelected server: {selected}:{port}") - return selected - else: - print("\nNo server found on local network") - print(" Please check:") - print(" 1. Server is running") - print(" 2. Server is on the same network") - print(f" 3. Server is listening on port {port}") - return None - - -# Run discovery if SERVER_HOST is None -if SERVER_HOST is None or SERVER_HOST == "None" or SERVER_HOST == "": - print("\nSERVER_HOST is not configured - starting discovery...\n") - discovered_host = discover_server(SERVER_PORT) - - if discovered_host: - SERVER_HOST = discovered_host - print(f"\nUsing discovered server: {SERVER_HOST}:{SERVER_PORT}\n") - else: - print("\nWARNING: No server discovered - will run in LOCAL-ONLY mode") - SERVER_HOST = "localhost" # Fallback to avoid errors - -SERVER = f"http://{SERVER_HOST}:{SERVER_PORT}" - -MODEL_CONFIG = config["model"] -INPUT_HEIGHT = MODEL_CONFIG["input_height"] -INPUT_WIDTH = MODEL_CONFIG["input_width"] -LAST_OFFLOADING_LAYER = MODEL_CONFIG["last_offloading_layer"] - -IMAGE_PATH = SCRIPT_DIR / MODEL_CONFIG["image_name"] -TFLITE_DIR = SCRIPT_DIR / MODEL_CONFIG["tflite_subdir"] -SUBMODEL_PREFIX = MODEL_CONFIG["submodel_prefix"] - -ENDPOINTS = config["http"]["endpoints"] - -# ===================================================== -# OPTIMIZATION: Configurable device input sending. -# When disabled, skips the send_image() POST entirely, saving -# ~18 KB bandwidth and one round-trip per iteration. -# ===================================================== -SEND_DEVICE_INPUT = config["http"].get("send_device_input", True) - -# ===================================================== -# OPTIMIZATION: Thread pool for overlapping the GET offloading_layer -# request with image preparation / inference setup. -# Saves one round-trip of latency per iteration. -# ===================================================== _request_pool = ThreadPoolExecutor(max_workers=1) -# ===================================================== -# OPTIMIZATION: Pre-compute the device ID header bytes once. -# The device ID never changes at runtime, so we encode it and pack -# its length prefix once instead of repeating this work every call. -# ===================================================== -_DEVICE_ID_BYTES = DEVICE_ID.encode("ascii") -_DEVICE_ID_HEADER = struct.pack("i", len(_DEVICE_ID_BYTES)) + _DEVICE_ID_BYTES - -# Initialize delay simulators -DELAY_CONFIG = config.get("delay_simulation", {}) -computation_delay = DelaySimulator(DELAY_CONFIG.get("device_computation")) -network_delay = DelaySimulator(DELAY_CONFIG.get("network")) - -if computation_delay.enabled: - print(f"Computation delay enabled: {computation_delay.get_delay_info()}") -if network_delay.enabled: - print(f"Network delay enabled: {network_delay.get_delay_info()}") - -# ===================================================== -# OPTIMIZATION: Pre-load and cache TFLite interpreters. -# Interpreters are expensive to create (file I/O + memory allocation). -# We create them once at startup and reuse across inference cycles. -# ===================================================== -_interpreter_cache = {} - - -def _get_interpreter(layer_index): - """Get a cached TFLite interpreter for the given layer index.""" - if layer_index not in _interpreter_cache: - model_path = str(TFLITE_DIR / f"{SUBMODEL_PREFIX}_{layer_index}.tflite") - interpreter = tf.lite.Interpreter(model_path=model_path) - interpreter.allocate_tensors() - _interpreter_cache[layer_index] = { - "interpreter": interpreter, - "input_details": interpreter.get_input_details(), - "output_details": interpreter.get_output_details(), - } - return _interpreter_cache[layer_index] - - -def _preload_interpreters(): - """Pre-load all TFLite interpreters at startup.""" - print(f"Pre-loading {LAST_OFFLOADING_LAYER + 1} TFLite interpreters...") - for i in range(LAST_OFFLOADING_LAYER + 1): - _get_interpreter(i) - print(f"All interpreters loaded and cached.") - - -# Function to generate a random message ID -def generate_message_id(): - return "".join(random.choices("ABCDEFGHIJKLMNOPQRSTUVWXYZ0123456789", k=4)) - - -# --------------------- -# Registration -# --------------------- -def register_device(): - url = f"{SERVER}{ENDPOINTS['registration']}" - model_hash = compute_model_hash() - payload = {"device_id": DEVICE_ID, "model_hash": model_hash} - try: - r = session.post(url, json=payload, timeout=5) - if r.status_code == 404: - print("WARNING: Server does not have this model - running LOCAL-ONLY") - return False - return r.status_code == 200 - except requests.exceptions.RequestException as e: - return False - - -# ------------------------------------------------ -# Convert image to RGB565 and send to server -# ------------------------------------------------ - -# ===================================================== -# OPTIMIZATION: Pre-compute static RGB565 data once, reuse across calls. -# Only the timestamp header needs to change per call. -# The send buffer is pre-allocated so send_image() only writes the -# 8-byte timestamp in-place instead of copying the entire image. -# ===================================================== -_cached_rgb565_buffer = None -_send_image_buffer = None - - -def _build_rgb565_buffer(): - """Build the RGB565 buffer from the image (done once at startup).""" - global _cached_rgb565_buffer, _send_image_buffer - img = Image.open(str(IMAGE_PATH)).resize((INPUT_HEIGHT, INPUT_WIDTH)).convert("RGB") - data = np.array(img) - # OPTIMIZATION: Vectorized NumPy conversion instead of nested Python loops. - # ~50-100x faster for typical image sizes (e.g. 96x96). - r = data[:, :, 0].astype(np.uint16) - g = data[:, :, 1].astype(np.uint16) - b = data[:, :, 2].astype(np.uint16) - rgb565 = ((r >> 3) << 11) | ((g >> 2) << 5) | (b >> 3) - _cached_rgb565_buffer = rgb565.astype(">u2").tobytes() - # Pre-allocate the send buffer (8 bytes timestamp + image data). - # send_image() will only overwrite the first 8 bytes each call. - _send_image_buffer = bytearray(8 + len(_cached_rgb565_buffer)) - _send_image_buffer[8:] = _cached_rgb565_buffer - - -def send_image(): - url = f"{SERVER}{ENDPOINTS['device_input']}" - - # OPTIMIZATION: Write timestamp in-place into the pre-allocated buffer. - # Avoids allocating and copying ~18 KB on every call. - struct.pack_into("d", _send_image_buffer, 0, time.time()) - - headers = {"Content-Type": "application/octet-stream"} - if COMPRESSION_ENABLED: - payload = gzip.compress(bytes(_send_image_buffer)) - headers["Content-Encoding"] = "gzip" - else: - payload = _send_image_buffer - - try: - r = session.post( - url, - data=payload, - headers=headers, - timeout=30, - ) - return True - except requests.exceptions.RequestException as e: - return False - - -# ---------------------------- -# Request Best Offloading Layer -# ---------------------------- -# _server_was_down = False - - -def get_offloading_layer(): - url = f"{SERVER}{ENDPOINTS['offloading_layer']}?device_id={DEVICE_ID}" - try: - r = session.get(url, timeout=10) - if r.status_code == 200: - best_layer = r.json().get("offloading_layer_index", LAST_OFFLOADING_LAYER) - - if best_layer > LAST_OFFLOADING_LAYER: - best_layer = LAST_OFFLOADING_LAYER - return best_layer - else: - return LAST_OFFLOADING_LAYER - - except requests.exceptions.RequestException: - return LAST_OFFLOADING_LAYER - - -# TEST: Simulate random best offloading layer change -def get_offloading_layer_random(): - url = f"{SERVER}{ENDPOINTS['offloading_layer']}" - r = session.get(url) - if r.status_code == 200: - best_layer = r.json().get("offloading_layer_index", LAST_OFFLOADING_LAYER) - print("Best layer received:", best_layer) - return random.randint(0, LAST_OFFLOADING_LAYER) - else: - print("Error requesting layer:", r.status_code) - return LAST_OFFLOADING_LAYER - - -# ---------------------- -# Send output -# --------------------- - -# ===================================================== -# OPTIMIZATION: Cache the loaded and processed image array. -# The image doesn't change between iterations, so we load it once. -# ===================================================== -_cached_image_rgb = None - - -def load_image_rgb(path): - global _cached_image_rgb - if _cached_image_rgb is None: - img = Image.open(str(path)).resize((INPUT_HEIGHT, INPUT_WIDTH)).convert("RGB") - img_np = np.asarray(img).astype(np.float32) / 255.0 - img_np = np.expand_dims(img_np, axis=0) # [1,H,W,3] - _cached_image_rgb = img_np - return _cached_image_rgb - - -def run_split_inference(image, tflite_dir, stop_layer): - input_data = image - inference_times = [] - # Handle -1 as "run all layers until the end" - if stop_layer == -1: - stop_layer = LAST_OFFLOADING_LAYER - - for i in range(stop_layer + 1): - # OPTIMIZATION: Use cached interpreter instead of recreating each time - cached = _get_interpreter(i) - interpreter = cached["interpreter"] - input_details = cached["input_details"] - output_details = cached["output_details"] - - input_data = input_data.astype(input_details[0]["dtype"]) - interpreter.set_tensor(input_details[0]["index"], input_data) - - t0 = time.time() - - # Apply artificial computation delay - if computation_delay.enabled: - delay = computation_delay.apply_delay() - - interpreter.invoke() - t1 = time.time() - inference_times.append(t1 - t0) - input_data = interpreter.get_tensor(output_details[0]["index"]) - return input_data, inference_times - - -def send_inference_result( - output_data, inference_times, layer_index, message_id, acq_time_ms -): - # Apply network delay before sending - send_start = time.time() - if network_delay.enabled: - delay = network_delay.apply_delay() - - url = f"{SERVER}{ENDPOINTS['device_inference_result']}" - timestamp = time.time() - - # Build the binary payload. - # _DEVICE_ID_HEADER is pre-computed at module level (length prefix + encoded bytes). - # The two adjacent int/float fields (layer_index, acq_time) are packed in a single call. - output_bytes = output_data.tobytes() - times_bytes = np.array(inference_times, dtype=np.float32).tobytes() - - # ===================================================== - # OPTIMIZATION: Pre-allocate the buffer at full size and use pack_into() - # to write fields in-place, avoiding 6+ incremental bytearray copies. - # ===================================================== - msg_id_bytes = message_id.encode("ascii").ljust(4, b"\x00") - header_size = ( - 8 # timestamp (d) - + len(_DEVICE_ID_HEADER) # device id length prefix + bytes - + 4 # message_id (4 ascii bytes) - + 4 - + 4 # layer_index (i) + acq_time (f) - + 4 # output length prefix (I) - ) - total_size = header_size + len(output_bytes) + 4 + len(times_bytes) - buffer = bytearray(total_size) - - offset = 0 - struct.pack_into("d", buffer, offset, timestamp) - offset += 8 - buffer[offset : offset + len(_DEVICE_ID_HEADER)] = _DEVICE_ID_HEADER - offset += len(_DEVICE_ID_HEADER) - buffer[offset : offset + 4] = msg_id_bytes - offset += 4 - struct.pack_into("if", buffer, offset, layer_index, acq_time_ms / 1000.0) - offset += 8 - struct.pack_into("I", buffer, offset, len(output_bytes)) - offset += 4 - buffer[offset : offset + len(output_bytes)] = output_bytes - offset += len(output_bytes) - struct.pack_into("i", buffer, offset, len(times_bytes)) - offset += 4 - buffer[offset : offset + len(times_bytes)] = times_bytes - - # ===================================================== - # OPTIMIZATION: Optional gzip compression to reduce bandwidth. - # Controlled via http_config.yaml -> compression.enabled - # ===================================================== - headers = {"Content-Type": "application/octet-stream"} - if COMPRESSION_ENABLED: - payload = gzip.compress(bytes(buffer)) - headers["Content-Encoding"] = "gzip" - else: - payload = buffer - - try: - r = session.post( - url, - data=payload, - headers=headers, - timeout=5, - ) - send_end = time.time() - network_time = send_end - send_start - data_size = len(payload) - network_speed = ( - (data_size / network_time) / 1024 if network_time > 0 else 0 - ) # KB/s - return True, network_time, network_speed, data_size - except requests.exceptions.RequestException as e: - return False, 0, 0, 0 - - -# ===================================================== -# OPTIMIZATION: Cache the model hash after first computation. -# The TFLite model files never change at runtime, so we avoid -# re-reading all files from disk on every registration/reconnect. -# ===================================================== -_cached_model_hash = None - - -def compute_model_hash(): - global _cached_model_hash - if _cached_model_hash is None: - hasher = hashlib.md5() - for tflite_file in sorted(TFLITE_DIR.glob("*.tflite")): - with open(tflite_file, "rb") as f: - hasher.update(f.read()) - _cached_model_hash = hasher.hexdigest() - return _cached_model_hash - - -# ----- -# MAIN -# ----- def main(): print("=" * 60) print("SCIoT Client Starting") @@ -862,4 +284,4 @@ def main(): sys.exit(0) if __name__ == "__main__": - main() \ No newline at end of file + main() diff --git a/src/client/python/http_clientCAMpi.py b/src/client/python/http_clientCAMpi.py index a3b9cf18..32eff2d9 100644 --- a/src/client/python/http_clientCAMpi.py +++ b/src/client/python/http_clientCAMpi.py @@ -1,608 +1,47 @@ -import tensorflow as tf -import numpy as np -from PIL import Image -import struct -import gzip -import requests -from requests.adapters import HTTPAdapter -from urllib3.util.retry import Retry -import time -import random -import os -import yaml -from pathlib import Path -from delay_simulator import DelaySimulator -import socket -import hashlib -import ipaddress -from concurrent.futures import ThreadPoolExecutor, as_completed -import cProfile -from profiler import AdvancedProfiler -from picamera2 import Picamera2 +"""SCIoT HTTP client โ€” Raspberry Pi camera (Picamera2) variant. -# Get the directory where this script is located -SCRIPT_DIR = Path(__file__).resolve().parent -PROJECT_ROOT = SCRIPT_DIR.parent.parent.parent -CONFIG_PATH = SCRIPT_DIR / "http_config.yaml" +Thin wrapper around ``http_client`` (the canonical client). All shared logic is +imported from there; this module only adds the Picamera2 frame source and its +capture loop. +""" -# Load configuration from YAML file -with open(CONFIG_PATH, "r") as f: - config = yaml.safe_load(f) - -# Se lanciamo il client con un argomento extra (es. python client.py my_device), utile per il file tes_file.py -import sys - -if len(sys.argv) > 1: - config["client"]["device_id"] = sys.argv[1] +import os +import time -DEVICE_ID = config["client"]["device_id"] -SERVER_HOST = config["http"]["server_host"] -SERVER_PORT = config["http"]["server_port"] +import numpy as np +from PIL import Image +from concurrent.futures import ThreadPoolExecutor -# ===================================================== -# OPTIMIZATION: Use requests.Session for connection pooling. -# This reuses TCP connections across requests (HTTP keep-alive) -# instead of creating a new TCP connection for each request. -# ===================================================== -session = requests.Session() +from profiler import AdvancedProfiler -# ===================================================== -# OPTIMIZATION: Configure HTTPAdapter with retry policy and pool sizing. -# - pool_connections / pool_maxsize sized for a single-server client -# - Lightweight retry (2 attempts) with exponential backoff on transient errors -# ===================================================== -_retry_strategy = Retry( - total=2, - backoff_factor=0.1, - status_forcelist=[502, 503, 504], - allowed_methods=["GET", "POST"], -) -_adapter = HTTPAdapter( - pool_connections=1, - pool_maxsize=4, - max_retries=_retry_strategy, +# Picamera2 is provided by Raspberry Pi OS (not PyPI). Import it defensively so +# this module still loads for tooling/analysis off-device; the camera init in +# main() fails with a clear error if Picamera2 is missing. +try: + from picamera2 import Picamera2 +except ModuleNotFoundError: + Picamera2 = None + +from http_client import ( + load_config, + SERVER, + DEVICE_ID, + TFLITE_DIR, + INPUT_HEIGHT, + INPUT_WIDTH, + LAST_OFFLOADING_LAYER, + register_device, + generate_message_id, + get_offloading_layer, + run_split_inference, + send_inference_result, + _preload_interpreters, + _build_rgb565_buffer, ) -session.mount("http://", _adapter) -session.mount("https://", _adapter) -# ===================================================== -# OPTIMIZATION: Dedicated discovery session with a larger pool -# for parallel subnet scanning (up to 50 threads). -# Avoids ephemeral socket churn from bare requests.post/get calls. -# ===================================================== -_discovery_session = requests.Session() -_discovery_adapter = HTTPAdapter(pool_connections=10, pool_maxsize=50) -_discovery_session.mount("http://", _discovery_adapter) -_discovery_session.mount("https://", _discovery_adapter) - -# ===================================================== -# CONFIGURATION: Optional gzip compression for payloads. -# Saves bandwidth on constrained IoT networks at the cost of -# ~1-2 ms CPU per request. Toggle via http_config.yaml. -# ===================================================== -COMPRESSION_ENABLED = config.get("compression", {}).get("enabled", False) - - -def load_config(config_filename="http_config.yaml"): - """Legge il file di configurazione YAML usando il percorso assoluto.""" - # Trova la cartella esatta dove risiede QUESTO script (http_client.py) - script_dir = os.path.dirname(os.path.abspath(__file__)) - # Unisce la cartella al nome del file - config_path = os.path.join(script_dir, config_filename) - - if os.path.exists(config_path): - with open(config_path, "r") as file: - return yaml.safe_load(file) - else: - print(f"โš ๏ธ [ATTENZIONE] File di configurazione non trovato in: {config_path}") - print("โš ๏ธ Uso le impostazioni di default (cProfile disabilitato).") - return {} - -# Discovery functions -def get_local_network(): - """Get the local network IP range for scanning.""" - try: - # Get local IP address - s = socket.socket(socket.AF_INET, socket.SOCK_DGRAM) - s.connect(("8.8.8.8", 80)) - local_ip = s.getsockname()[0] - s.close() - - # Convert to network/24 (assuming typical home network) - network = ipaddress.IPv4Network(f"{local_ip}/24", strict=False) - return network, local_ip - except Exception as e: - print(f"Error getting local network: {e}") - return None, None - - -def check_server_at_ip(ip, port, timeout=1.0): - """Check if a server is responding at the given IP:port.""" - try: - # First try TCP connection - sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM) - sock.settimeout(timeout) - result = sock.connect_ex((str(ip), port)) - sock.close() - - if result == 0: - # Port is open, verify it's our server by trying multiple methods - # OPTIMIZATION: Use _discovery_session instead of bare requests.*() - # to reuse TCP connections across the 50-thread parallel scan. - try: - # Try 1: POST to registration endpoint (proper way) - test_url = ( - f"http://{ip}:{port}{config['http']['endpoints']['registration']}" - ) - response = _discovery_session.post( - test_url, json={"device_id": "discovery_test"}, timeout=2 - ) - if response.status_code in [ - 200, - 201, - 400, - 409, - ]: # Any reasonable server response - print( - f" Found server at {ip}:{port} (registration: {response.status_code})" - ) - return str(ip) - except requests.exceptions.RequestException: - pass - - try: - # Try 2: GET request to root or any endpoint - response = _discovery_session.get(f"http://{ip}:{port}/", timeout=2) - if response.status_code < 500: # Any response except server error - print( - f" Found server at {ip}:{port} (root: {response.status_code})" - ) - return str(ip) - except requests.exceptions.RequestException: - pass - - try: - # Try 3: Check offloading_layer endpoint - test_url = f"http://{ip}:{port}{config['http']['endpoints']['offloading_layer']}" - response = _discovery_session.get(test_url, timeout=2) - if response.status_code in [200, 400, 404, 405]: - print( - f" Found server at {ip}:{port} (offload endpoint: {response.status_code})" - ) - return str(ip) - except requests.exceptions.RequestException: - pass - - return None - except Exception as e: - return None - - -def discover_server(port): - """Scan local network to discover server.""" - print("=" * 60) - print("SERVER DISCOVERY MODE") - print("=" * 60) - - # First, try common local addresses - print(f"Checking localhost and common addresses on port {port}...\n") - common_hosts = ["127.0.0.1", "localhost", "0.0.0.0"] - - for host in common_hosts: - print(f" Trying {host}:{port}...") - result = check_server_at_ip(host, port, timeout=2.0) - if result: - print(f"\nServer found at {result}:{port}") - return result - - print("\nNo server on localhost, scanning local network...\n") - - network, local_ip = get_local_network() - if not network: - print("ERROR: Could not determine local network") - return None - - print(f"Local IP: {local_ip}") - print(f"Scanning network: {network}") - print(f"This may take 10-30 seconds...\n") - - # Scan network with threading for speed - found_servers = [] - total_hosts = network.num_addresses - 2 # Exclude network and broadcast - checked = 0 - - with ThreadPoolExecutor(max_workers=50) as executor: - futures = { - executor.submit(check_server_at_ip, ip, port): ip - for ip in network.hosts() - if str(ip) != local_ip - } - - for future in as_completed(futures): - checked += 1 - if checked % 25 == 0: - print(f"Progress: {checked}/{total_hosts} hosts checked...") - - result = future.result() - if result: - found_servers.append(result) - print(f"\nSERVER FOUND: {result}:{port}") - # Cancel remaining futures for faster completion - for f in futures: - f.cancel() - break - - print(f"\nScan complete: {checked}/{total_hosts} hosts checked") - - if found_servers: - selected = found_servers[0] - print(f"\nSelected server: {selected}:{port}") - return selected - else: - print("\nNo server found on local network") - print(" Please check:") - print(" 1. Server is running") - print(" 2. Server is on the same network") - print(f" 3. Server is listening on port {port}") - return None - - -# Run discovery if SERVER_HOST is None -if SERVER_HOST is None or SERVER_HOST == "None" or SERVER_HOST == "": - print("\nSERVER_HOST is not configured - starting discovery...\n") - discovered_host = discover_server(SERVER_PORT) - - if discovered_host: - SERVER_HOST = discovered_host - print(f"\nUsing discovered server: {SERVER_HOST}:{SERVER_PORT}\n") - else: - print("\nWARNING: No server discovered - will run in LOCAL-ONLY mode") - SERVER_HOST = "localhost" # Fallback to avoid errors - -SERVER = f"http://{SERVER_HOST}:{SERVER_PORT}" - -MODEL_CONFIG = config["model"] -INPUT_HEIGHT = MODEL_CONFIG["input_height"] -INPUT_WIDTH = MODEL_CONFIG["input_width"] -LAST_OFFLOADING_LAYER = MODEL_CONFIG["last_offloading_layer"] - -IMAGE_PATH = SCRIPT_DIR / MODEL_CONFIG["image_name"] -TFLITE_DIR = SCRIPT_DIR / MODEL_CONFIG["tflite_subdir"] -SUBMODEL_PREFIX = MODEL_CONFIG["submodel_prefix"] - -ENDPOINTS = config["http"]["endpoints"] - -# ===================================================== -# OPTIMIZATION: Configurable device input sending. -# When disabled, skips the send_image() POST entirely, saving -# ~18 KB bandwidth and one round-trip per iteration. -# ===================================================== -SEND_DEVICE_INPUT = config["http"].get("send_device_input", True) - -# ===================================================== -# OPTIMIZATION: Thread pool for overlapping the GET offloading_layer -# request with image preparation / inference setup. -# Saves one round-trip of latency per iteration. -# ===================================================== _request_pool = ThreadPoolExecutor(max_workers=1) -# ===================================================== -# OPTIMIZATION: Pre-compute the device ID header bytes once. -# The device ID never changes at runtime, so we encode it and pack -# its length prefix once instead of repeating this work every call. -# ===================================================== -_DEVICE_ID_BYTES = DEVICE_ID.encode("ascii") -_DEVICE_ID_HEADER = struct.pack("i", len(_DEVICE_ID_BYTES)) + _DEVICE_ID_BYTES - -# Initialize delay simulators -DELAY_CONFIG = config.get("delay_simulation", {}) -computation_delay = DelaySimulator(DELAY_CONFIG.get("device_computation")) -network_delay = DelaySimulator(DELAY_CONFIG.get("network")) - -if computation_delay.enabled: - print(f"Computation delay enabled: {computation_delay.get_delay_info()}") -if network_delay.enabled: - print(f"Network delay enabled: {network_delay.get_delay_info()}") - -# ===================================================== -# OPTIMIZATION: Pre-load and cache TFLite interpreters. -# Interpreters are expensive to create (file I/O + memory allocation). -# We create them once at startup and reuse across inference cycles. -# ===================================================== -_interpreter_cache = {} - - -def _get_interpreter(layer_index): - """Get a cached TFLite interpreter for the given layer index.""" - if layer_index not in _interpreter_cache: - model_path = str(TFLITE_DIR / f"{SUBMODEL_PREFIX}_{layer_index}.tflite") - interpreter = tf.lite.Interpreter(model_path=model_path) - interpreter.allocate_tensors() - _interpreter_cache[layer_index] = { - "interpreter": interpreter, - "input_details": interpreter.get_input_details(), - "output_details": interpreter.get_output_details(), - } - return _interpreter_cache[layer_index] - - -def _preload_interpreters(): - """Pre-load all TFLite interpreters at startup.""" - print(f"Pre-loading {LAST_OFFLOADING_LAYER + 1} TFLite interpreters...") - for i in range(LAST_OFFLOADING_LAYER + 1): - _get_interpreter(i) - print(f"All interpreters loaded and cached.") - - -# Function to generate a random message ID -def generate_message_id(): - return "".join(random.choices("ABCDEFGHIJKLMNOPQRSTUVWXYZ0123456789", k=4)) - - -# --------------------- -# Registration -# --------------------- -def register_device(): - url = f"{SERVER}{ENDPOINTS['registration']}" - model_hash = compute_model_hash() - payload = {"device_id": DEVICE_ID, "model_hash": model_hash} - try: - r = session.post(url, json=payload, timeout=5) - if r.status_code == 404: - print("WARNING: Server does not have this model - running LOCAL-ONLY") - return False - return r.status_code == 200 - except requests.exceptions.RequestException as e: - return False - - -# ------------------------------------------------ -# Convert image to RGB565 and send to server -# ------------------------------------------------ - -# ===================================================== -# OPTIMIZATION: Pre-compute static RGB565 data once, reuse across calls. -# Only the timestamp header needs to change per call. -# The send buffer is pre-allocated so send_image() only writes the -# 8-byte timestamp in-place instead of copying the entire image. -# ===================================================== -_cached_rgb565_buffer = None -_send_image_buffer = None - - -def _build_rgb565_buffer(): - """Build the RGB565 buffer from the image (done once at startup).""" - global _cached_rgb565_buffer, _send_image_buffer - img = Image.open(str(IMAGE_PATH)).resize((INPUT_HEIGHT, INPUT_WIDTH)).convert("RGB") - data = np.array(img) - # OPTIMIZATION: Vectorized NumPy conversion instead of nested Python loops. - # ~50-100x faster for typical image sizes (e.g. 96x96). - r = data[:, :, 0].astype(np.uint16) - g = data[:, :, 1].astype(np.uint16) - b = data[:, :, 2].astype(np.uint16) - rgb565 = ((r >> 3) << 11) | ((g >> 2) << 5) | (b >> 3) - _cached_rgb565_buffer = rgb565.astype(">u2").tobytes() - # Pre-allocate the send buffer (8 bytes timestamp + image data). - # send_image() will only overwrite the first 8 bytes each call. - _send_image_buffer = bytearray(8 + len(_cached_rgb565_buffer)) - _send_image_buffer[8:] = _cached_rgb565_buffer - - -def send_image(): - url = f"{SERVER}{ENDPOINTS['device_input']}" - - # OPTIMIZATION: Write timestamp in-place into the pre-allocated buffer. - # Avoids allocating and copying ~18 KB on every call. - struct.pack_into("d", _send_image_buffer, 0, time.time()) - - headers = {"Content-Type": "application/octet-stream"} - if COMPRESSION_ENABLED: - payload = gzip.compress(bytes(_send_image_buffer)) - headers["Content-Encoding"] = "gzip" - else: - payload = _send_image_buffer - - try: - r = session.post( - url, - data=payload, - headers=headers, - timeout=30, - ) - return True - except requests.exceptions.RequestException as e: - return False - - -# ---------------------------- -# Request Best Offloading Layer -# ---------------------------- -# _server_was_down = False - - -def get_offloading_layer(): - url = f"{SERVER}{ENDPOINTS['offloading_layer']}?device_id={DEVICE_ID}" - try: - r = session.get(url, timeout=10) - if r.status_code == 200: - best_layer = r.json().get("offloading_layer_index", LAST_OFFLOADING_LAYER) - - if best_layer > LAST_OFFLOADING_LAYER: - best_layer = LAST_OFFLOADING_LAYER - return best_layer - else: - return LAST_OFFLOADING_LAYER - - except requests.exceptions.RequestException: - return LAST_OFFLOADING_LAYER - - -# TEST: Simulate random best offloading layer change -def get_offloading_layer_random(): - url = f"{SERVER}{ENDPOINTS['offloading_layer']}" - r = session.get(url) - if r.status_code == 200: - best_layer = r.json().get("offloading_layer_index", LAST_OFFLOADING_LAYER) - print("Best layer received:", best_layer) - return random.randint(0, LAST_OFFLOADING_LAYER) - else: - print("Error requesting layer:", r.status_code) - return LAST_OFFLOADING_LAYER - - -# ---------------------- -# Send output -# --------------------- - -# ===================================================== -# OPTIMIZATION: Cache the loaded and processed image array. -# The image doesn't change between iterations, so we load it once. -# ===================================================== -_cached_image_rgb = None - - -def load_image_rgb(path): - global _cached_image_rgb - if _cached_image_rgb is None: - img = Image.open(str(path)).resize((INPUT_HEIGHT, INPUT_WIDTH)).convert("RGB") - img_np = np.asarray(img).astype(np.float32) / 255.0 - img_np = np.expand_dims(img_np, axis=0) # [1,H,W,3] - _cached_image_rgb = img_np - return _cached_image_rgb - - -def run_split_inference(image, tflite_dir, stop_layer): - input_data = image - inference_times = [] - - # Handle -1 as "run all layers until the end" - if stop_layer == -1: - stop_layer = LAST_OFFLOADING_LAYER - - for i in range(stop_layer + 1): - # OPTIMIZATION: Use cached interpreter instead of recreating each time - cached = _get_interpreter(i) - interpreter = cached["interpreter"] - input_details = cached["input_details"] - output_details = cached["output_details"] - - input_data = input_data.astype(input_details[0]["dtype"]) - interpreter.set_tensor(input_details[0]["index"], input_data) - - t0 = time.time() - - # Apply artificial computation delay - if computation_delay.enabled: - delay = computation_delay.apply_delay() - - interpreter.invoke() - t1 = time.time() - inference_times.append(t1 - t0) - input_data = interpreter.get_tensor(output_details[0]["index"]) - return input_data, inference_times - - -def send_inference_result( - output_data, inference_times, layer_index, message_id, acq_time_ms -): - # Apply network delay before sending - send_start = time.time() - if network_delay.enabled: - delay = network_delay.apply_delay() - - url = f"{SERVER}{ENDPOINTS['device_inference_result']}" - timestamp = time.time() - - # Build the binary payload. - # _DEVICE_ID_HEADER is pre-computed at module level (length prefix + encoded bytes). - # The two adjacent int/float fields (layer_index, acq_time) are packed in a single call. - output_bytes = output_data.tobytes() - times_bytes = np.array(inference_times, dtype=np.float32).tobytes() - - # ===================================================== - # OPTIMIZATION: Pre-allocate the buffer at full size and use pack_into() - # to write fields in-place, avoiding 6+ incremental bytearray copies. - # ===================================================== - msg_id_bytes = message_id.encode("ascii").ljust(4, b"\x00") - header_size = ( - 8 # timestamp (d) - + len(_DEVICE_ID_HEADER) # device id length prefix + bytes - + 4 # message_id (4 ascii bytes) - + 4 - + 4 # layer_index (i) + acq_time (f) - + 4 # output length prefix (I) - ) - total_size = header_size + len(output_bytes) + 4 + len(times_bytes) - buffer = bytearray(total_size) - - offset = 0 - struct.pack_into("d", buffer, offset, timestamp) - offset += 8 - buffer[offset : offset + len(_DEVICE_ID_HEADER)] = _DEVICE_ID_HEADER - offset += len(_DEVICE_ID_HEADER) - buffer[offset : offset + 4] = msg_id_bytes - offset += 4 - struct.pack_into("if", buffer, offset, layer_index, acq_time_ms / 1000.0) - offset += 8 - struct.pack_into("I", buffer, offset, len(output_bytes)) - offset += 4 - buffer[offset : offset + len(output_bytes)] = output_bytes - offset += len(output_bytes) - struct.pack_into("i", buffer, offset, len(times_bytes)) - offset += 4 - buffer[offset : offset + len(times_bytes)] = times_bytes - - # ===================================================== - # OPTIMIZATION: Optional gzip compression to reduce bandwidth. - # Controlled via http_config.yaml -> compression.enabled - # ===================================================== - headers = {"Content-Type": "application/octet-stream"} - if COMPRESSION_ENABLED: - payload = gzip.compress(bytes(buffer)) - headers["Content-Encoding"] = "gzip" - else: - payload = buffer - - try: - r = session.post( - url, - data=payload, - headers=headers, - timeout=5, - ) - send_end = time.time() - network_time = send_end - send_start - data_size = len(payload) - network_speed = ( - (data_size / network_time) / 1024 if network_time > 0 else 0 - ) # KB/s - return True, network_time, network_speed, data_size - except requests.exceptions.RequestException as e: - return False, 0, 0, 0 - - -# ===================================================== -# OPTIMIZATION: Cache the model hash after first computation. -# The TFLite model files never change at runtime, so we avoid -# re-reading all files from disk on every registration/reconnect. -# ===================================================== -_cached_model_hash = None - - -def compute_model_hash(): - global _cached_model_hash - if _cached_model_hash is None: - hasher = hashlib.md5() - for tflite_file in sorted(TFLITE_DIR.glob("*.tflite")): - with open(tflite_file, "rb") as f: - hasher.update(f.read()) - _cached_model_hash = hasher.hexdigest() - return _cached_model_hash - -# ----- -# MAIN -# ----- def main(): print("=" * 60) print("SCIoT Client Starting") diff --git a/src/client/python/http_clientCAMpiBench.py b/src/client/python/http_clientCAMpiBench.py index b9088ca0..ca92d04b 100644 --- a/src/client/python/http_clientCAMpiBench.py +++ b/src/client/python/http_clientCAMpiBench.py @@ -1,640 +1,63 @@ -import tensorflow as tf -import numpy as np -from PIL import Image -import struct -import gzip -import requests -from requests.adapters import HTTPAdapter -from urllib3.util.retry import Retry -import time -import random +"""SCIoT HTTP client โ€” Raspberry Pi camera benchmark/sweep variant. + +Thin wrapper around ``http_client`` (the canonical client). Adds CLI overrides +(``--id``/``--model``/``--res``) for benchmark sweeps and exports profiler +reports on shutdown; all shared logic is imported from ``http_client``. +""" + import os -import yaml -from pathlib import Path -from delay_simulator import DelaySimulator -import socket -import hashlib -import ipaddress -from concurrent.futures import ThreadPoolExecutor, as_completed -import cProfile -from profiler import AdvancedProfiler -from picamera2 import Picamera2 import sys +import time import argparse -from pathlib import Path - -# Get the directory where this script is located -SCRIPT_DIR = Path(__file__).resolve().parent -PROJECT_ROOT = SCRIPT_DIR.parent.parent.parent -CONFIG_PATH = SCRIPT_DIR / "http_config.yaml" +import numpy as np +from PIL import Image +from concurrent.futures import ThreadPoolExecutor -# Load configuration from YAML file -with open(CONFIG_PATH, "r") as f: - config = yaml.safe_load(f) -parser = argparse.ArgumentParser(description="SCIoT Edge Client") +from profiler import AdvancedProfiler -parser.add_argument("--id", type=str, help="Sovrascrive il Device ID") -parser.add_argument("--model", type=str, help="Sovrascrive la cartella TFLite") -parser.add_argument("--res", type=int, help="Sovrascrive la risoluzione di input") +# Picamera2 is provided by Raspberry Pi OS (not PyPI). Import it defensively so +# this module still loads for tooling/analysis off-device; the camera init in +# main() fails with a clear error if Picamera2 is missing. +try: + from picamera2 import Picamera2 +except ModuleNotFoundError: + Picamera2 = None -args, unknown = parser.parse_known_args() +import http_client as base +parser = argparse.ArgumentParser(description="SCIoT Edge Client") +parser.add_argument("--id", type=str, help="Override the device ID") +parser.add_argument("--model", type=str, help="Override the TFLite folder") +parser.add_argument("--res", type=int, help="Override the input resolution") +args, _unknown = parser.parse_known_args() if args.id: - config["client"]["device_id"] = args.id -if args.model: - config["model"]["tflite_subdir"] = args.model - - # โ”€โ”€โ”€ RILEVAMENTO AUTOMATICO PREFISSO E LAYER โ”€โ”€โ”€ - model_dir = Path(__file__).resolve().parent / args.model - tflite_files = list(model_dir.glob("*.tflite")) - - if tflite_files: - # 1. Rileva il nome esatto (es. "submodel") - prefix_trovato = tflite_files[0].stem.rsplit('_', 1)[0] - config["model"]["submodel_prefix"] = prefix_trovato - print(f"๐Ÿ” Prefisso rilevato automaticamente: {prefix_trovato}") - - # 2. Conta quanti file ci sono per evitare i crash! - numero_max_layer = len(tflite_files) - 1 - config["model"]["last_offloading_layer"] = numero_max_layer - print(f"๐Ÿ“Š Layer rilevati: {len(tflite_files)} (Ultimo: {numero_max_layer})") - else: - print(f"โŒ ATTENZIONE: La cartella {args.model} รจ vuota!") -if args.res: - config["model"]["input_height"] = args.res - config["model"]["input_width"] = args.res - -LAST_OFFLOADING_LAYER = config["model"]["last_offloading_layer"] - -DEVICE_ID = config["client"]["device_id"] -SERVER_HOST = config["http"]["server_host"] -SERVER_PORT = config["http"]["server_port"] - -# ===================================================== -# OPTIMIZATION: Use requests.Session for connection pooling. -# This reuses TCP connections across requests (HTTP keep-alive) -# instead of creating a new TCP connection for each request. -# ===================================================== -session = requests.Session() - -# ===================================================== -# OPTIMIZATION: Configure HTTPAdapter with retry policy and pool sizing. -# - pool_connections / pool_maxsize sized for a single-server client -# - Lightweight retry (2 attempts) with exponential backoff on transient errors -# ===================================================== -_retry_strategy = Retry( - total=2, - backoff_factor=0.1, - status_forcelist=[502, 503, 504], - allowed_methods=["GET", "POST"], -) -_adapter = HTTPAdapter( - pool_connections=1, - pool_maxsize=4, - max_retries=_retry_strategy, + base.set_device_id(args.id) +if args.model or args.res: + base.set_model_config(tflite_subdir=args.model, input_size=args.res) + +from http_client import ( + load_config, + IMAGE_PATH, + SERVER, + DEVICE_ID, + TFLITE_DIR, + INPUT_HEIGHT, + INPUT_WIDTH, + LAST_OFFLOADING_LAYER, + register_device, + generate_message_id, + get_offloading_layer, + run_split_inference, + send_inference_result, + _preload_interpreters, + _build_rgb565_buffer, + load_image_rgb, ) -session.mount("http://", _adapter) -session.mount("https://", _adapter) - -# ===================================================== -# OPTIMIZATION: Dedicated discovery session with a larger pool -# for parallel subnet scanning (up to 50 threads). -# Avoids ephemeral socket churn from bare requests.post/get calls. -# ===================================================== -_discovery_session = requests.Session() -_discovery_adapter = HTTPAdapter(pool_connections=10, pool_maxsize=50) -_discovery_session.mount("http://", _discovery_adapter) -_discovery_session.mount("https://", _discovery_adapter) - -# ===================================================== -# CONFIGURATION: Optional gzip compression for payloads. -# Saves bandwidth on constrained IoT networks at the cost of -# ~1-2 ms CPU per request. Toggle via http_config.yaml. -# ===================================================== -COMPRESSION_ENABLED = config.get("compression", {}).get("enabled", False) - - -def load_config(config_filename="http_config.yaml"): - """Legge il file di configurazione YAML usando il percorso assoluto.""" - # Trova la cartella esatta dove risiede QUESTO script (http_client.py) - script_dir = os.path.dirname(os.path.abspath(__file__)) - # Unisce la cartella al nome del file - config_path = os.path.join(script_dir, config_filename) - - if os.path.exists(config_path): - with open(config_path, "r") as file: - return yaml.safe_load(file) - else: - print(f"โš ๏ธ [ATTENZIONE] File di configurazione non trovato in: {config_path}") - print("โš ๏ธ Uso le impostazioni di default (cProfile disabilitato).") - return {} - -# Discovery functions -def get_local_network(): - """Get the local network IP range for scanning.""" - try: - # Get local IP address - s = socket.socket(socket.AF_INET, socket.SOCK_DGRAM) - s.connect(("8.8.8.8", 80)) - local_ip = s.getsockname()[0] - s.close() - - # Convert to network/24 (assuming typical home network) - network = ipaddress.IPv4Network(f"{local_ip}/24", strict=False) - return network, local_ip - except Exception as e: - print(f"Error getting local network: {e}") - return None, None - - -def check_server_at_ip(ip, port, timeout=1.0): - """Check if a server is responding at the given IP:port.""" - try: - # First try TCP connection - sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM) - sock.settimeout(timeout) - result = sock.connect_ex((str(ip), port)) - sock.close() - - if result == 0: - # Port is open, verify it's our server by trying multiple methods - # OPTIMIZATION: Use _discovery_session instead of bare requests.*() - # to reuse TCP connections across the 50-thread parallel scan. - try: - # Try 1: POST to registration endpoint (proper way) - test_url = ( - f"http://{ip}:{port}{config['http']['endpoints']['registration']}" - ) - response = _discovery_session.post( - test_url, json={"device_id": "discovery_test"}, timeout=2 - ) - if response.status_code in [ - 200, - 201, - 400, - 409, - ]: # Any reasonable server response - print( - f" Found server at {ip}:{port} (registration: {response.status_code})" - ) - return str(ip) - except requests.exceptions.RequestException: - pass - try: - # Try 2: GET request to root or any endpoint - response = _discovery_session.get(f"http://{ip}:{port}/", timeout=2) - if response.status_code < 500: # Any response except server error - print( - f" Found server at {ip}:{port} (root: {response.status_code})" - ) - return str(ip) - except requests.exceptions.RequestException: - pass - - try: - # Try 3: Check offloading_layer endpoint - test_url = f"http://{ip}:{port}{config['http']['endpoints']['offloading_layer']}" - response = _discovery_session.get(test_url, timeout=2) - if response.status_code in [200, 400, 404, 405]: - print( - f" Found server at {ip}:{port} (offload endpoint: {response.status_code})" - ) - return str(ip) - except requests.exceptions.RequestException: - pass - - return None - except Exception as e: - return None - - -def discover_server(port): - """Scan local network to discover server.""" - print("=" * 60) - print("SERVER DISCOVERY MODE") - print("=" * 60) - - # First, try common local addresses - print(f"Checking localhost and common addresses on port {port}...\n") - common_hosts = ["127.0.0.1", "localhost", "0.0.0.0"] - - for host in common_hosts: - print(f" Trying {host}:{port}...") - result = check_server_at_ip(host, port, timeout=2.0) - if result: - print(f"\nServer found at {result}:{port}") - return result - - print("\nNo server on localhost, scanning local network...\n") - - network, local_ip = get_local_network() - if not network: - print("ERROR: Could not determine local network") - return None - - print(f"Local IP: {local_ip}") - print(f"Scanning network: {network}") - print(f"This may take 10-30 seconds...\n") - - # Scan network with threading for speed - found_servers = [] - total_hosts = network.num_addresses - 2 # Exclude network and broadcast - checked = 0 - - with ThreadPoolExecutor(max_workers=50) as executor: - futures = { - executor.submit(check_server_at_ip, ip, port): ip - for ip in network.hosts() - if str(ip) != local_ip - } - - for future in as_completed(futures): - checked += 1 - if checked % 25 == 0: - print(f"Progress: {checked}/{total_hosts} hosts checked...") - - result = future.result() - if result: - found_servers.append(result) - print(f"\nSERVER FOUND: {result}:{port}") - # Cancel remaining futures for faster completion - for f in futures: - f.cancel() - break - - print(f"\nScan complete: {checked}/{total_hosts} hosts checked") - - if found_servers: - selected = found_servers[0] - print(f"\nSelected server: {selected}:{port}") - return selected - else: - print("\nNo server found on local network") - print(" Please check:") - print(" 1. Server is running") - print(" 2. Server is on the same network") - print(f" 3. Server is listening on port {port}") - return None - - -# Run discovery if SERVER_HOST is None -if SERVER_HOST is None or SERVER_HOST == "None" or SERVER_HOST == "": - print("\nSERVER_HOST is not configured - starting discovery...\n") - discovered_host = discover_server(SERVER_PORT) - - if discovered_host: - SERVER_HOST = discovered_host - print(f"\nUsing discovered server: {SERVER_HOST}:{SERVER_PORT}\n") - else: - print("\nWARNING: No server discovered - will run in LOCAL-ONLY mode") - SERVER_HOST = "localhost" # Fallback to avoid errors - -SERVER = f"http://{SERVER_HOST}:{SERVER_PORT}" - -MODEL_CONFIG = config["model"] -INPUT_HEIGHT = MODEL_CONFIG["input_height"] -INPUT_WIDTH = MODEL_CONFIG["input_width"] -LAST_OFFLOADING_LAYER = MODEL_CONFIG["last_offloading_layer"] - -IMAGE_PATH = SCRIPT_DIR / MODEL_CONFIG["image_name"] -TFLITE_DIR = SCRIPT_DIR / MODEL_CONFIG["tflite_subdir"] -SUBMODEL_PREFIX = MODEL_CONFIG["submodel_prefix"] - -ENDPOINTS = config["http"]["endpoints"] - -# ===================================================== -# OPTIMIZATION: Configurable device input sending. -# When disabled, skips the send_image() POST entirely, saving -# ~18 KB bandwidth and one round-trip per iteration. -# ===================================================== -SEND_DEVICE_INPUT = config["http"].get("send_device_input", True) - -# ===================================================== -# OPTIMIZATION: Thread pool for overlapping the GET offloading_layer -# request with image preparation / inference setup. -# Saves one round-trip of latency per iteration. -# ===================================================== _request_pool = ThreadPoolExecutor(max_workers=1) -# ===================================================== -# OPTIMIZATION: Pre-compute the device ID header bytes once. -# The device ID never changes at runtime, so we encode it and pack -# its length prefix once instead of repeating this work every call. -# ===================================================== -_DEVICE_ID_BYTES = DEVICE_ID.encode("ascii") -_DEVICE_ID_HEADER = struct.pack("i", len(_DEVICE_ID_BYTES)) + _DEVICE_ID_BYTES - -# Initialize delay simulators -DELAY_CONFIG = config.get("delay_simulation", {}) -computation_delay = DelaySimulator(DELAY_CONFIG.get("device_computation")) -network_delay = DelaySimulator(DELAY_CONFIG.get("network")) - -if computation_delay.enabled: - print(f"Computation delay enabled: {computation_delay.get_delay_info()}") -if network_delay.enabled: - print(f"Network delay enabled: {network_delay.get_delay_info()}") - -# ===================================================== -# OPTIMIZATION: Pre-load and cache TFLite interpreters. -# Interpreters are expensive to create (file I/O + memory allocation). -# We create them once at startup and reuse across inference cycles. -# ===================================================== -_interpreter_cache = {} - - -def _get_interpreter(layer_index): - """Get a cached TFLite interpreter for the given layer index.""" - if layer_index not in _interpreter_cache: - model_path = str(TFLITE_DIR / f"{SUBMODEL_PREFIX}_{layer_index}.tflite") - interpreter = tf.lite.Interpreter(model_path=model_path) - interpreter.allocate_tensors() - _interpreter_cache[layer_index] = { - "interpreter": interpreter, - "input_details": interpreter.get_input_details(), - "output_details": interpreter.get_output_details(), - } - return _interpreter_cache[layer_index] - - -def _preload_interpreters(): - """Pre-load all TFLite interpreters at startup.""" - print(f"Pre-loading {LAST_OFFLOADING_LAYER + 1} TFLite interpreters...") - for i in range(LAST_OFFLOADING_LAYER + 1): - _get_interpreter(i) - print(f"All interpreters loaded and cached.") - - -# Function to generate a random message ID -def generate_message_id(): - return "".join(random.choices("ABCDEFGHIJKLMNOPQRSTUVWXYZ0123456789", k=4)) - - -# --------------------- -# Registration -# --------------------- -def register_device(): - url = f"{SERVER}{ENDPOINTS['registration']}" - model_hash = compute_model_hash() - payload = {"device_id": DEVICE_ID, "model_hash": model_hash} - try: - r = session.post(url, json=payload, timeout=5) - if r.status_code == 404: - print("WARNING: Server does not have this model - running LOCAL-ONLY") - return False - return r.status_code == 200 - except requests.exceptions.RequestException as e: - return False - - -# ------------------------------------------------ -# Convert image to RGB565 and send to server -# ------------------------------------------------ - -# ===================================================== -# OPTIMIZATION: Pre-compute static RGB565 data once, reuse across calls. -# Only the timestamp header needs to change per call. -# The send buffer is pre-allocated so send_image() only writes the -# 8-byte timestamp in-place instead of copying the entire image. -# ===================================================== -_cached_rgb565_buffer = None -_send_image_buffer = None - - -def _build_rgb565_buffer(): - """Build the RGB565 buffer from the image (done once at startup).""" - global _cached_rgb565_buffer, _send_image_buffer - img = Image.open(str(IMAGE_PATH)).resize((INPUT_HEIGHT, INPUT_WIDTH)).convert("RGB") - data = np.array(img) - # OPTIMIZATION: Vectorized NumPy conversion instead of nested Python loops. - # ~50-100x faster for typical image sizes (e.g. 96x96). - r = data[:, :, 0].astype(np.uint16) - g = data[:, :, 1].astype(np.uint16) - b = data[:, :, 2].astype(np.uint16) - rgb565 = ((r >> 3) << 11) | ((g >> 2) << 5) | (b >> 3) - _cached_rgb565_buffer = rgb565.astype(">u2").tobytes() - # Pre-allocate the send buffer (8 bytes timestamp + image data). - # send_image() will only overwrite the first 8 bytes each call. - _send_image_buffer = bytearray(8 + len(_cached_rgb565_buffer)) - _send_image_buffer[8:] = _cached_rgb565_buffer - - -def send_image(): - url = f"{SERVER}{ENDPOINTS['device_input']}" - - # OPTIMIZATION: Write timestamp in-place into the pre-allocated buffer. - # Avoids allocating and copying ~18 KB on every call. - struct.pack_into("d", _send_image_buffer, 0, time.time()) - - headers = {"Content-Type": "application/octet-stream"} - if COMPRESSION_ENABLED: - payload = gzip.compress(bytes(_send_image_buffer)) - headers["Content-Encoding"] = "gzip" - else: - payload = _send_image_buffer - - try: - r = session.post( - url, - data=payload, - headers=headers, - timeout=30, - ) - return True - except requests.exceptions.RequestException as e: - return False - - -# ---------------------------- -# Request Best Offloading Layer -# ---------------------------- -# _server_was_down = False - - -def get_offloading_layer(): - url = f"{SERVER}{ENDPOINTS['offloading_layer']}?device_id={DEVICE_ID}" - try: - r = session.get(url, timeout=10) - if r.status_code == 200: - best_layer = r.json().get("offloading_layer_index", LAST_OFFLOADING_LAYER) - - if best_layer > LAST_OFFLOADING_LAYER: - best_layer = LAST_OFFLOADING_LAYER - return best_layer - else: - return LAST_OFFLOADING_LAYER - - except requests.exceptions.RequestException: - return LAST_OFFLOADING_LAYER - - -# TEST: Simulate random best offloading layer change -def get_offloading_layer_random(): - url = f"{SERVER}{ENDPOINTS['offloading_layer']}" - r = session.get(url) - if r.status_code == 200: - best_layer = r.json().get("offloading_layer_index", LAST_OFFLOADING_LAYER) - print("Best layer received:", best_layer) - return random.randint(0, LAST_OFFLOADING_LAYER) - else: - print("Error requesting layer:", r.status_code) - return LAST_OFFLOADING_LAYER - - -# ---------------------- -# Send output -# --------------------- - -# ===================================================== -# OPTIMIZATION: Cache the loaded and processed image array. -# The image doesn't change between iterations, so we load it once. -# ===================================================== -_cached_image_rgb = None - - -def load_image_rgb(path): - global _cached_image_rgb - if _cached_image_rgb is None: - img = Image.open(str(path)).resize((INPUT_HEIGHT, INPUT_WIDTH)).convert("RGB") - img_np = np.asarray(img).astype(np.float32) / 255.0 - img_np = np.expand_dims(img_np, axis=0) # [1,H,W,3] - _cached_image_rgb = img_np - return _cached_image_rgb - - -def run_split_inference(image, tflite_dir, stop_layer): - input_data = image - inference_times = [] - # Handle -1 as "run all layers until the end" - if stop_layer == -1: - stop_layer = LAST_OFFLOADING_LAYER - - for i in range(stop_layer + 1): - # OPTIMIZATION: Use cached interpreter instead of recreating each time - cached = _get_interpreter(i) - interpreter = cached["interpreter"] - input_details = cached["input_details"] - output_details = cached["output_details"] - - input_data = input_data.astype(input_details[0]["dtype"]) - interpreter.set_tensor(input_details[0]["index"], input_data) - - t0 = time.time() - - # Apply artificial computation delay - if computation_delay.enabled: - delay = computation_delay.apply_delay() - - interpreter.invoke() - t1 = time.time() - inference_times.append(t1 - t0) - input_data = interpreter.get_tensor(output_details[0]["index"]) - return input_data, inference_times - - -def send_inference_result( - output_data, inference_times, layer_index, message_id, acq_time_ms -): - # Apply network delay before sending - send_start = time.time() - if network_delay.enabled: - delay = network_delay.apply_delay() - - url = f"{SERVER}{ENDPOINTS['device_inference_result']}" - timestamp = time.time() - - # Build the binary payload. - # _DEVICE_ID_HEADER is pre-computed at module level (length prefix + encoded bytes). - # The two adjacent int/float fields (layer_index, acq_time) are packed in a single call. - output_bytes = output_data.tobytes() - times_bytes = np.array(inference_times, dtype=np.float32).tobytes() - - # ===================================================== - # OPTIMIZATION: Pre-allocate the buffer at full size and use pack_into() - # to write fields in-place, avoiding 6+ incremental bytearray copies. - # ===================================================== - msg_id_bytes = message_id.encode("ascii").ljust(4, b"\x00") - header_size = ( - 8 # timestamp (d) - + len(_DEVICE_ID_HEADER) # device id length prefix + bytes - + 4 # message_id (4 ascii bytes) - + 4 - + 4 # layer_index (i) + acq_time (f) - + 4 # output length prefix (I) - ) - total_size = header_size + len(output_bytes) + 4 + len(times_bytes) - buffer = bytearray(total_size) - - offset = 0 - struct.pack_into("d", buffer, offset, timestamp) - offset += 8 - buffer[offset : offset + len(_DEVICE_ID_HEADER)] = _DEVICE_ID_HEADER - offset += len(_DEVICE_ID_HEADER) - buffer[offset : offset + 4] = msg_id_bytes - offset += 4 - struct.pack_into("if", buffer, offset, layer_index, acq_time_ms / 1000.0) - offset += 8 - struct.pack_into("I", buffer, offset, len(output_bytes)) - offset += 4 - buffer[offset : offset + len(output_bytes)] = output_bytes - offset += len(output_bytes) - struct.pack_into("i", buffer, offset, len(times_bytes)) - offset += 4 - buffer[offset : offset + len(times_bytes)] = times_bytes - - # ===================================================== - # OPTIMIZATION: Optional gzip compression to reduce bandwidth. - # Controlled via http_config.yaml -> compression.enabled - # ===================================================== - headers = {"Content-Type": "application/octet-stream"} - if COMPRESSION_ENABLED: - payload = gzip.compress(bytes(buffer)) - headers["Content-Encoding"] = "gzip" - else: - payload = buffer - - try: - r = session.post( - url, - data=payload, - headers=headers, - timeout=5, - ) - send_end = time.time() - network_time = send_end - send_start - data_size = len(payload) - network_speed = ( - (data_size / network_time) / 1024 if network_time > 0 else 0 - ) # KB/s - return True, network_time, network_speed, data_size - except requests.exceptions.RequestException as e: - return False, 0, 0, 0 - - -# ===================================================== -# OPTIMIZATION: Cache the model hash after first computation. -# The TFLite model files never change at runtime, so we avoid -# re-reading all files from disk on every registration/reconnect. -# ===================================================== -_cached_model_hash = None - - -def compute_model_hash(): - global _cached_model_hash - if _cached_model_hash is None: - hasher = hashlib.md5() - for tflite_file in sorted(TFLITE_DIR.glob("*.tflite")): - with open(tflite_file, "rb") as f: - hasher.update(f.read()) - _cached_model_hash = hasher.hexdigest() - return _cached_model_hash - - -# ----- -# MAIN -# ----- def main(): print("=" * 60) print("SCIoT Client Starting") @@ -869,4 +292,4 @@ def main(): sys.exit(0) if __name__ == "__main__": - main() \ No newline at end of file + main() From d3c1d74b259f0a5965d0cb0f3ed480d4e92cdd55 Mon Sep 17 00:00:00 2001 From: ff225 Date: Fri, 10 Jul 2026 10:31:42 +0200 Subject: [PATCH 2/2] ci: pin Python to 3.13 via .python-version uv otherwise selects Python 3.14 (requires-python is >=3.13 with no upper bound), but tensorflow only ships wheels up to cp313, so 'uv sync --frozen' fails to resolve. Pinning 3.13 makes CI (and local dev) use a Python that has tensorflow wheels. Repo-wide fix; also unblocks main. --- .python-version | 1 + 1 file changed, 1 insertion(+) create mode 100644 .python-version diff --git a/.python-version b/.python-version new file mode 100644 index 00000000..24ee5b1b --- /dev/null +++ b/.python-version @@ -0,0 +1 @@ +3.13