Skip to content

AEG / AEG-Comm implementation overview

[!NOTE] For the latest implementation status, please refer to Functional Implementation Status (Remaining Functionality).

Target audience: Engineers implementing/reviewing energy control gating and distributed communications. Useful code: control.py and aeg_comm.py.

This document explains the implementation background and components of AEG and AEG-Comm in the order of concept, processing flow, data structure, and integration procedure. The role and reading of each diagram are clearly stated to help you follow the design intent.

1. Concept

  • AEG (Activity-driven Energy Gating): Energy is retained for each neuron and consumed by spikes. If the energy is below the threshold, gate it and support replenishment with rewards and power saving adaptation.
  • AEG-Comm: Extends AEG to inter-node messaging and implements 3-layer safety (energy gate, critical priority, timestamp guarantee) plus buffer/batch sending.
  • Transport: Zenoh Pub/Sub (compression, optional encryption, retries, performance statistics) by ZenohCommunicator.

Overview: AEG manages local computational load using energy concepts and suppresses excessive firing. AEG-Comm, on the other hand, uses a three-layer safety system to screen transmissions to achieve both communication redundancy and safety, and only sends them to Zenoh when necessary.

2. Role of components

  • AEG (Local Gate): Energy management, load-responsive consumption, arbitrary adaptive thresholds, reward replenishment, power statistics.
  • AEGCommGate: Wraps AEG and determines whether transmission is possible. Adds modality-specific safety, buffering, batch, and telemetry.
  • ZenohCommunicator: Publish spike packet. Supports compression, encryption hook (MT25-EV015), retry and ACK.

Detailed explanation: - AEG calculates the consumption per spike in update() and masks neurons below the threshold. If enable_power_optimization is enabled, the consumption rate will be automatically adjusted according to the load, and if adaptive_threshold is enabled, the threshold will be dynamically updated from the load history. - AEGCommGate receives locally gated spikes and uses 3-layer safety to decide whether to send or not. Prioritize critical conditions and timestamp constraints, and if it cannot be sent, store it in a local buffer and send it as a batch later. - ZenohCommunicator is a Pub/Sub layer with compression, encryption, retransmission, ACK, and reconnection. Send spikes and metadata to topics with publish_spikes and track performance statistics internally.

3. High-level architecture```mermaid

graph LR subgraph ローカルノード S[spike tensor] I[Importance
Meta context] S -->|spikes| G[AEG.update] I --> G G -->|gated spikes| C{AEG-Comm
3層セーフティ} C -->|allow| TX[Zenoh publish_spikes] C -->|block| B[Local buffer
Batch flush] end TX --> Net[Zenoh topics] Net --> Remote[remote module/node] 

**Figure description:** At the local node on the left, spikes and severity enter `AEG.update` and gated spikes proceed to 3-tier safety determination. Those that are approved are published to the network via Zenoh, and those that are rejected are accumulated in a local buffer and followed up by batch transmission. This shows a design that reliably sends important data while suppressing the amount of communication.

## 4. AEG-Comm 3-layer safety (operation)```mermaid
flowchart TD
    P[SpikePacket
modality / data / metadata / ts] --> L1{Layer 1:
エネルギーゲート}
    L1 -->|energy ok| L2{Layer 2:
クリティカル優先?}
    L1 -->|low energy| BUF[buffer]
    L2 -->|critical| SEND[Immediate sending]
    L2 -->|normal| L3{Layer 3:
タイムスタンプ期限超過?}
    L3 -->|> max_interval| SEND
    L3 -->|within window| BUF
    BUF --> FLUSH[Regular batch flush]
    SEND --> PUB[Zenoh publish_spikes]
```- **Layer 1:** Consumption estimate = `consumption_rate * sum(spikes) * importance`; Sent only when average energy exceeds `threshold + estimated_consumption`.
- **Layer 2 (critical priority):** Always sent with modality `critical_modalities` (default `force` and `safety`), force change above `force_change_threshold`, emergency keyword in the text, or one of the `emergency` / `safety_interrupt` flags.
- **Layer 3 (Timestamp Guaranteed):** Forced transmission if `max_interval_ms` (default 50ms) has passed since the last transmission for each modality.
- **Buffer:** Cache unsent packets (default 1000). Send batches periodically with `spikes_batch/{target}` (default 1 second).

**Figure explanation:** Packets first undergo energy determination (Layer 1), and if insufficient, are sent to a buffer. Immediate transmission if critical conditions (Layer 2) are met. Normally, packets are forced to be sent at regular intervals due to timestamp guarantee (Layer 3), and if they are within the interval, they are kept in a buffer. Buffers are periodically transmitted in batches to optimize both transmission delay and bandwidth.

## 5. Data structures and defaults
- `SpikePacket`: `{timestamp_ns, modality, data (tensor), metadata}`.
- `LocalSpikeBuffer(max_size=1000)`: FIFO cache. Batch flush to `spikes_batch/{target}`.
- `AEG` Default: `initial_energy=255.0`, `threshold=10.0`, `consumption_rate=1.0`, `supply_rate=5.0`, `adaptive_threshold=False`, `enable_power_optimization` Optional.
- `AEGCommGate` Default: `critical_modalities=["force","safety"]`, `emergency_keywords=["stop","pain","danger","stop","stop","pain","danger","halt"]`, `force_change_threshold=10.0`, `max_interval_ms=50.0`, `batch_interval_s=1.0`.

## 6. Sequence (transmission route)```mermaid
sequenceDiagram
    participant Src as ローカルモジュール
    participant AEG as AEG.update()
    participant Gate as AEGCommGate
    participant Zen as ZenohCommunicator
    participant Rmt as リモートノード

    Src->>AEG: spikes, importance
    AEG-->>Src: gated_spikes
    Src->>Gate: gated_spikes, modality, metadata
    Gate->>Gate: _should_send (energy, criticality, timestamp)
    alt should_send
        Gate->>Zen: publish_spikes(target, data, metadata)
    else buffer
        Gate->>Gate: cache(packet)
    end
    Zen-->>Rmt: spikes/{target}

Figure description: Local module passes spikes and importance to AEG.update and feeds gated spikes to AEGCommGate. AEGCommGate performs a three-layer safety judgment, and if it is possible to send, it requests Publish to ZenohCommunicator, and if not, it sends a buffer. Zenoh delivers spikes to remote nodes, and upstream/downstream module coordination is established.

7. Integration points

  • Model side gate: Call AEG.update(spikes, importance) during forward and mask with energy > threshold.
  • Communication side gate: Wrap the output with AEGCommGate(..., communicator=ZenohCommunicator(...)) and pass modality/metadata to apply priority logic.
  • Reward Supply: Replenish energy with AEG.supply(reward) or AEGCommGate.supply(reward).
  • Statistics: Get transmit/buffer/critical rate and reduction rate with AEGCommGate.get_stats(), energy/efficiency metrics with AEG.get_power_statistics().

8. Safety and Reliability Notes

  • Critical routes are bypassed while ensuring freshness using energy gates (Layer 2) and timestamp guards (Layer 3).
  • Local buffer prevents loss at low energy times and batch transmission reduces bandwidth.
  • The Zenoh layer has compression, arbitrary encryption, retry, ACK, and reconnection logic to ensure delivery reliability.

9. Quick start (connection example)

1) Communicator generation: comm = ZenohCommunicator(node_id="pfc-0", config=ZenohConfig(...)). 2) Wrap: gate = AEGCommGate(num_neurons=..., communicator=comm, target_node="pfc"). 3) In forward:python importance = ... # shape broadcastable to spikes spikes_out = gate(spikes, importance, modality="vision", metadata={"text": "stop"})4) Optionally call gate.supply(reward) from the reinforcement learning loop, etc.

10. Traceability

  • Core gate processing: control.py
  • Communication safety wrapper: aeg_comm.py
  • Transport/Runtime: zenoh_comm.py
  • Test: test_aeg_comm.py, test_control.py