Config-driven adaptive-risk offloading algorithm

docs/CONFIGURATION.md

@@ -38,12 +38,38 @@ Server validation covers:
 - HTTP/WebSocket/MQTT hostnames, ports, endpoints/topics, NTP server names, and model references;
 - model dimensions, split-layer limits, and model directories;
 - local-inference probability in the inclusive range `0.0..1.0`;
+- offloading algorithm selection under `offloading.algorithm` (`static` or `adaptive_risk`) and adaptive tuning values;
 - delay sections and delay distributions;
 - boolean flags such as `verbose`, `debug_cprofiler`, compression, and input saving;
 - `evaluation.split.max_rows` and `evaluation.split.interval_minutes` (non-negative);
 - `evaluation.outputs.inference_cycles` and `evaluation.outputs.offloading_decisions` (booleans);
 - optional Firestore telemetry publishing under `evaluation.firestore`.
 
+## Offloading algorithm
+
+The server chooses the next split point through the configured offloading
+algorithm. If the section is absent, the runtime keeps the legacy `static`
+behavior. To enable the adaptive risk-aware policy:
+
+```yaml
+offloading:
+  algorithm: adaptive_risk   # static | adaptive_risk
+  adaptive_risk:
+    network_ewma_alpha: 0.35
+    network_window_size: 10
+    uncertainty_weight: 0.50
+    hysteresis_margin_ms: 5
+    min_speed_bytes_per_second: 1
+    probe_probability: 0.05
+    device_load_weight: 0.50
+```
+
+`adaptive_risk` reuses the same candidate cost components as `static`, then
+adds penalties for unstable device/edge measurements, noisy network speed, and
+small layer switches. It also scales device-side compute estimates by the
+reported device CPU occupancy so a busy client moves toward earlier offloading
+points faster.
+
 ## Evaluation outputs
 
 The server writes inference-cycle CSV output under

docs/OFFLOADING_DECISION_EVENTS.md

@@ -23,7 +23,7 @@ structure.
 | `device_id` | string | Client/device identifier used for registration and split requests. |
 | `model_dir` | string | Model artifact directory/profile used for the decision. |
 | `request_id` | string | Correlation ID linking the client request, prediction, and decision event. |
-| `strategy` | string | Algorithm family, currently `static`. |
+| `strategy` | string | Algorithm family, such as `static` or `adaptive_risk`. |
 | `selected_layer` | integer | Selected offloading layer; `-1` means local-only fallback. |
 | `selection_reason` | string | Why the layer was selected, such as `lowest_estimated_cost`, `forced_local_inference`, or `fallback_missing_model_metadata`. |
 | `avg_speed_bytes_per_second` | number | Network speed estimate used by the algorithm. |
@@ -43,6 +43,11 @@ Each candidate includes:
 | `considered_for_selection` | boolean | Whether the candidate participates in the final choice. |
 | `selected` | boolean | Whether this candidate matches the selected layer. |
 
+Adaptive candidates may also include `adaptive_score`, `risk_penalty`,
+`switch_penalty`, `network_speed_ewma`, `network_uncertainty`,
+`device_uncertainty`, `edge_uncertainty`, `device_cpu_percent`,
+`device_load_factor`, `raw_device_compute_cost`, and `device_load_penalty`.
+
 ## Example
 
 ```csv

docs/TRANSPORTS.md

@@ -73,7 +73,7 @@ Versioned inference-result payloads start with this 8-byte header:
 | Field | Type | Value |
 | --- | --- | --- |
 | `magic` | `char[4]` | `SCIP` |
-| `version` | `uint16` | `1` |
+| `version` | `uint16` | `2` |
 | `flags` | `uint16` | `0` |
 
 The body is:
@@ -86,14 +86,13 @@ The body is:
 | `message_id` | ASCII `char[4]` | client message identifier |
 | `offloading_layer_index` | `int32` | `-1` or `0..4095` |
 | `acquisition_time` | `float32` | seconds, finite and non-negative |
+| `device_cpu_percent` | `float32` | whole-machine CPU occupancy, `0..100` |
 | `layer_output_size` | `uint32` | 1 to 32 MiB, multiple of `float32` |
 | `layer_output` | `float32[]` | split-layer tensor data |
 | `layer_times_size` | `uint32` | 0 to 64 KiB, multiple of `float32` |
 | `layer_times` | `float32[]` | per-layer device timings, seconds |
 
-Compatibility rule: payloads that do not start with the `SCIP` magic are decoded
-as legacy unversioned payloads using the same body layout. The decoder labels
-those messages as protocol version `0` internally. Payloads with an unknown
+Payloads without the `SCIP` magic/header are rejected. Payloads with an unknown
 version, non-zero flags, invalid sizes, unsupported field values, or trailing
 bytes are rejected as malformed inference payloads by HTTP, WebSocket, and MQTT.
 

docs/config/server.full.yaml

@@ -70,6 +70,17 @@ local_inference_mode:
 offloading_algo:
   ema_alpha: 0.5
 
+offloading:
+  algorithm: adaptive_risk   # static | adaptive_risk
+  adaptive_risk:
+    network_ewma_alpha: 0.35
+    network_window_size: 10
+    uncertainty_weight: 0.50
+    hysteresis_margin_ms: 5
+    min_speed_bytes_per_second: 1
+    probe_probability: 0.05
+    device_load_weight: 0.50
+
 delay_simulation:
   computation:
     enabled: false

src/client/python/http_client.py

@@ -16,10 +16,14 @@
 import ipaddress
 from concurrent.futures import ThreadPoolExecutor, as_completed
 from concurrent.futures import TimeoutError as FutureTimeoutError
-import cProfile
 
 from sciot.config import load_client_config
 
+try:
+    import psutil
+except ImportError:
+    psutil = None
+
 try:
     from .delay_simulator import DelaySimulator
     from .profiler import AdvancedProfiler
@@ -67,6 +71,15 @@
 session.mount("http://", _adapter)
 session.mount("https://", _adapter)
 
+CLIENT_TABLE_HEADER = (
+    "Offload | Acq Time (ms) | Comp Time (ms) | Net Time (ms) | "
+    "Net Speed (KB/s) | Data (bytes)"
+)
+CLIENT_TABLE_SEPARATOR = "-" * len(CLIENT_TABLE_HEADER)
+CLIENT_TABLE_HEADER_REPEAT_INTERVAL = 20
+_client_table_header_printed = False
+_client_table_rows_printed = 0
+
 # =====================================================
 # OPTIMIZATION: Dedicated discovery session with a larger pool
 # for parallel subnet scanning (up to 50 threads).
@@ -84,6 +97,18 @@
 # =====================================================
 COMPRESSION_ENABLED = config.get("compression", {}).get("enabled", False)
 
+INFERENCE_PROTOCOL_MAGIC = b"SCIP"
+INFERENCE_PROTOCOL_VERSION = 2
+INFERENCE_PROTOCOL_FLAGS = 0
+INFERENCE_PROTOCOL_HEADER_FORMAT = "<4sHH"
+INFERENCE_PROTOCOL_HEADER_BYTES = struct.calcsize(INFERENCE_PROTOCOL_HEADER_FORMAT)
+
+if psutil is not None:
+    try:
+        psutil.cpu_percent(interval=None)
+    except Exception:
+        pass
+
 
 def load_config(config_filename=None):
     """Legge il file di configurazione YAML usando il percorso assoluto."""
@@ -171,7 +196,7 @@ def check_server_at_ip(ip, port, timeout=1.0):
                 pass
 
         return None
-    except Exception as e:
+    except Exception:
         return None
 
 
@@ -201,7 +226,7 @@ def discover_server(port):
 
     print(f"Local IP: {local_ip}")
     print(f"Scanning network: {network}")
-    print(f"This may take 10-30 seconds...\n")
+    print("This may take 10-30 seconds...\n")
 
     # Scan network with threading for speed
     found_servers = []
@@ -293,6 +318,19 @@ def set_device_id(device_id):
     _DEVICE_ID_BYTES = DEVICE_ID.encode("ascii")
     _DEVICE_ID_HEADER = struct.pack("<I", len(_DEVICE_ID_BYTES)) + _DEVICE_ID_BYTES
 
+
+def get_device_cpu_percent() -> float:
+    """Return whole-machine CPU occupancy as a bounded percentage."""
+    if psutil is None:
+        return 0.0
+    try:
+        cpu_percent = float(psutil.cpu_percent(interval=None))
+    except Exception:
+        return 0.0
+    if not np.isfinite(cpu_percent):
+        return 0.0
+    return min(100.0, max(0.0, cpu_percent))
+
 # Initialize delay simulators
 DELAY_CONFIG = config.get("delay_simulation", {})
 computation_delay = DelaySimulator(DELAY_CONFIG.get("device_computation"))
@@ -341,7 +379,7 @@ def _preload_interpreters():
     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.")
+    print("All interpreters loaded and cached.")
 
 
 # Function to generate a random message ID
@@ -362,7 +400,7 @@ def register_device():
             print("WARNING: Server does not have this model - running LOCAL-ONLY")
             return False
         return r.status_code == 200
-    except requests.exceptions.RequestException as e:
+    except requests.exceptions.RequestException:
         return False
 
 
@@ -413,14 +451,14 @@ def send_image():
         payload = _send_image_buffer
 
     try:
-        r = session.post(
+        session.post(
             url,
             data=payload,
             headers=headers,
             timeout=5,
         )
         return True
-    except requests.exceptions.RequestException as e:
+    except requests.exceptions.RequestException:
         return False
 
 
@@ -691,6 +729,47 @@ def shutdown(self):
         self._shutdown_complete = True
 
 
+def print_client_table_header(*, reset_rows: bool = False):
+    global _client_table_header_printed, _client_table_rows_printed
+    if reset_rows:
+        _client_table_rows_printed = 0
+    print(f"\n{CLIENT_TABLE_HEADER}")
+    print(CLIENT_TABLE_SEPARATOR)
+    _client_table_header_printed = True
+
+
+def print_client_table_row(
+    *,
+    layer: int,
+    acq_time: float,
+    comp_time: float,
+    net_time_ms: float | None = None,
+    net_speed: float | None = None,
+    data_size: int | None = None,
+):
+    global _client_table_rows_printed
+    if (
+        not _client_table_header_printed
+        or (
+            _client_table_rows_printed > 0
+            and _client_table_rows_printed % CLIENT_TABLE_HEADER_REPEAT_INTERVAL == 0
+        )
+    ):
+        print_client_table_header()
+
+    if net_time_ms is None or net_speed is None or data_size is None:
+        print(
+            f"{layer:7d} | {acq_time:14.2f} | {comp_time:14.2f} | "
+            f"{'N/A':>12} | {'N/A':>16} | {'N/A':>12}"
+        )
+    else:
+        print(
+            f"{layer:7d} | {acq_time:14.2f} | {comp_time:14.2f} | "
+            f"{net_time_ms:12.2f} | {net_speed:16.2f} | {data_size:12d}"
+        )
+    _client_table_rows_printed += 1
+
+
 def print_upload_result(upload_result):
     success, net_time, net_speed, data_size, layer, acq_time, comp_time = upload_result
 
@@ -701,12 +780,19 @@ def print_upload_result(upload_result):
         # Se il layer eseguito è l'ultimo del modello, l'inferenza è stata
         # 100% locale. La rete non ha trasportato l'immagine.
         if layer >= LAST_OFFLOADING_LAYER:
-            print(f"{layer:7d} | {acq_time:14.2f} | {comp_time:14.2f} | {'N/A':>12} | {'N/A':>16} | {'N/A':>12}")
+            print_client_table_row(layer=layer, acq_time=acq_time, comp_time=comp_time)
         else:
             # Altrimenti c'è stato Offload (Totale a 0, o Parziale).
-            print(f"{layer:7d} | {acq_time:14.2f} | {comp_time:14.2f} | {net_time * 1000:12.2f} | {net_speed:16.2f} | {data_size:12d}")
+            print_client_table_row(
+                layer=layer,
+                acq_time=acq_time,
+                comp_time=comp_time,
+                net_time_ms=net_time * 1000,
+                net_speed=net_speed,
+                data_size=data_size,
+            )
     else:
-        print(f"{layer:7d} | {acq_time:14.2f} | {comp_time:14.2f} | {'N/A':>12} | {'N/A':>16} | {'N/A':>12}")
+        print_client_table_row(layer=layer, acq_time=acq_time, comp_time=comp_time)
 
     return success
 
@@ -767,7 +853,7 @@ def run_split_inference(image, tflite_dir, stop_layer):
 
         # Apply artificial computation delay
         if computation_delay.enabled:
-            delay = computation_delay.apply_delay()
+            computation_delay.apply_delay()
 
         interpreter.invoke()
         t1 = time.time()
@@ -784,9 +870,13 @@ def build_inference_payload(
     acq_time_ms,
     *,
     timestamp=None,
+    device_cpu_percent=None,
 ):
     """Serialize an inference result using the validated little-endian protocol."""
     timestamp = time.time() if timestamp is None else timestamp
+    if device_cpu_percent is None:
+        device_cpu_percent = get_device_cpu_percent()
+    device_cpu_percent = min(100.0, max(0.0, float(device_cpu_percent)))
     output_bytes = np.asarray(output_data, dtype="<f4").tobytes()
     times_bytes = np.asarray(inference_times, dtype="<f4").tobytes()
 
@@ -796,25 +886,43 @@ def build_inference_payload(
     # =====================================================
     msg_id_bytes = message_id.encode("ascii")[:4].ljust(4, b"\x00")
     header_size = (
-        8  # timestamp (d)
+        INFERENCE_PROTOCOL_HEADER_BYTES  # magic, version, flags
+        + 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  # layer_index (i)
+        + 4  # acq_time (f)
+        + 4  # device_cpu_percent (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(
+        INFERENCE_PROTOCOL_HEADER_FORMAT,
+        buffer,
+        offset,
+        INFERENCE_PROTOCOL_MAGIC,
+        INFERENCE_PROTOCOL_VERSION,
+        INFERENCE_PROTOCOL_FLAGS,
+    )
+    offset += INFERENCE_PROTOCOL_HEADER_BYTES
     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(
+        "<iff",
+        buffer,
+        offset,
+        layer_index,
+        acq_time_ms / 1000.0,
+        device_cpu_percent,
+    )
+    offset += 12
     struct.pack_into("<I", buffer, offset, len(output_bytes))
     offset += 4
     buffer[offset : offset + len(output_bytes)] = output_bytes
@@ -995,10 +1103,7 @@ def main(
     else:
         print("Connected to server\n")
 
-    print(
-        "Offload | Acq Time (ms) | Comp Time (ms) | Net Time (ms) | Net Speed (KB/s) | Data (bytes)"
-    )
-    print("-" * 90)
+    print_client_table_header(reset_rows=True)
 
     # imposto un "timer" per non fare richieste in continuazione
     last_reconnect_attempt = time.time()
@@ -1129,7 +1234,11 @@ def main(
                         message_id,
                     )
                 else:
-                    print(f"{best_layer:7d} | {acq_time:14.2f} | {comp_time:14.2f} | {'N/A':>12} | {'N/A':>16} | {'N/A':>12}")
+                    print_client_table_row(
+                        layer=best_layer,
+                        acq_time=acq_time,
+                        comp_time=comp_time,
+                    )
 
                 # ─── GESTIONE ESPORTAZIONI E PROFILING ───
                 cicli_completati += 1