Fix UDP drops due to NAT timeouts and state table limits on specific routers.

[wangjushi930531-arch]

Issue: The native UDP mode suffers from severe throughput degradation (dropping to 5–10 pps) in specific router environments due to NAT timeouts and firewall restrictions.

Modification: Refactored the underlying logic of traffic_generator.py to implement support for persistent TCP connections.

Result: Throughput has recovered to a stable range of 90–130 pps, with significantly enhanced transmission stability.

Note: Due to the lack of extensive compatibility testing across diverse router brands, this code has not been committed to the main repository. It is provided solely as an architectural reference to address low-throughput bottlenecks.

"""
Micro-ESPectre - WiFi Traffic Generator

Generates UDP/TCP traffic to ensure continuous CSI data flow.
Essential for maintaining stable CSI packet reception on ESP32-C6.

Author: Francesco Pace <francesco.pace@gmail.com>
License: GPLv3
"""
import socket
import time
import _thread
import network

# Note: No thread lock needed for simple integer operations on MicroPython/ESP32
# Integer reads/writes are atomic on 32-bit systems

TRAFFIC_RATE_MIN = 0          # Minimum rate (0=disabled)
TRAFFIC_RATE_MAX = 1000       # Maximum rate (packets per second)
METRICS_INTERVAL = 500        # Metrics update interval (packets, ~5s at 100pps)

class TrafficGenerator:
    """WiFi traffic generator using lightweight TCP connections"""
    
    def __init__(self):
        """Initialize traffic generator"""
        self.running = False
        self.rate_pps = 0
        self.packet_count = 0
        self.error_count = 0
        self.gateway_ip = None
        self.start_time = 0  # Time when generator started (ticks_ms)
        self.actual_pps = 0  # Actual packets per second (moving window)
        
    def _get_gateway_ip(self):
        """Get gateway IP address from network interface"""
        try:
            wlan = network.WLAN(network.STA_IF)
            if not wlan.isconnected():
                return None
            
            # ifconfig returns: (ip, netmask, gateway, dns)
            ip_info = wlan.ifconfig()
            if len(ip_info) >= 3:
                return ip_info[2]  # Gateway IP
            return None
        except Exception as e:
            print(f"Error getting gateway IP: {e}")
            return None
    
    def _traffic_task(self): 
        """Background task that sends lightweight TCP packets (runs in thread)
        
        Instead of DNS, we open a quick TCP connection to port 80 on the gateway.
        This triggers a reply (SYN+ACK), generating incoming traffic → CSI.
        The socket is immediately closed to minimize resource usage.
        """
        
        # Pre-resolve destination address
        dest_port = 80  # Use HTTP port – most gateways respond to SYN on 80
        
        # Microsecond timing with fractional accumulator (aligned with C++ implementation)
        # This compensates for integer division error (e.g., 1000000/100 = 10000µs exact)
        interval_us = 1000000 // self.rate_pps
        remainder_us = 1000000 % self.rate_pps
        accumulator = 0
        
        next_send_time = time.ticks_us()
        
        # For actual PPS calculation (moving window)
        window_start_time = time.ticks_us()
        window_packet_count = 0
        
        while self.running:
            try:
                # Open and immediately close a TCP connection to gateway:80
                # This generates bidirectional traffic (SYN → SYN+ACK) → triggers CSI
                try:
                    sock = socket.socket()
                    sock.settimeout(0.05)
                    sock.connect((self.gateway_ip, dest_port))
                    sock.close()
                    self.packet_count += 1
                    window_packet_count += 1
                        
                except OSError as e:
                    # Socket error (e.g., network unavailable, ENOMEM)
                    self.error_count += 1
                    if self.error_count % 100 == 1:
                        print(f"Socket error: {e}")
                
                # Calculate next send time with fractional accumulator for precise rate
                accumulator += remainder_us
                extra_us = accumulator // self.rate_pps
                accumulator %= self.rate_pps
                
                next_send_time += interval_us + extra_us
                
                # Periodic metrics update (no GC needed - no allocations in loop)
                if window_packet_count >= METRICS_INTERVAL:
                    # Update actual pps (moving window)
                    window_elapsed = time.ticks_diff(time.ticks_us(), window_start_time)
                    if window_elapsed > 0:
                        self.actual_pps = (window_packet_count * 1000000) / window_elapsed
                    
                    window_start_time = time.ticks_us()
                    window_packet_count = 0
                
                # Sleep until next send time
                now = time.ticks_us()
                sleep_us = time.ticks_diff(next_send_time, now)
                
                if sleep_us > 100:
                    # Convert to ms for sleep (minimum 1ms to yield to other threads)
                    sleep_ms = sleep_us // 1000
                    if sleep_ms > 0:
                        time.sleep_ms(sleep_ms)
                    else:
                        time.sleep_us(sleep_us)
                elif sleep_us < -100000:
                    # We're more than 100ms behind, reset timing
                    next_send_time = time.ticks_us()
                else:
                    # Small sleep to yield to other threads
                    time.sleep_us(100)
                
            except Exception as e:
                self.error_count += 1
                
                # Log occasional errors
                if self.error_count % 10 == 1:
                    print(f"Traffic generator error: {e}")
                
                # Fallback sleep to avoid tight loop on error
                time.sleep_ms(max(1, interval_us // 1000))
    
    def start(self, rate_pps, max_retries=3, retry_delay=2):
        """
        Start traffic generator
        
        Args:
            rate_pps: Packets per second (0-1000, recommended: 100)
            max_retries: Number of retries to get gateway IP (default: 3)
            retry_delay: Seconds between retries (default: 2)
            
        Returns:
            bool: True if started successfully
        """
        if self.running:
            print("Traffic generator already running")
            return False
        
        if rate_pps < TRAFFIC_RATE_MIN or rate_pps > TRAFFIC_RATE_MAX:
            print(f"Invalid rate: {rate_pps} (must be {TRAFFIC_RATE_MIN}-{TRAFFIC_RATE_MAX} packets/sec)")
            return False
        
        # Get gateway IP with retries
        for attempt in range(1, max_retries + 1):
            self.gateway_ip = self._get_gateway_ip()
            if self.gateway_ip:
                break
            print(f"Failed to get gateway IP (attempt {attempt}/{max_retries})")
            if attempt < max_retries:
                time.sleep(retry_delay)
        
        if not self.gateway_ip:
            print(f"ERROR: Could not get gateway IP after {max_retries} attempts")
            return False
        
        # Reset counters
        self.packet_count = 0
        self.error_count = 0
        self.rate_pps = rate_pps
        self.start_time = time.ticks_ms()
        self.running = True
        
        # Start background task
        try:
            _thread.start_new_thread(self._traffic_task, ())
            return True
        except Exception as e:
            print(f"Failed to start traffic generator: {e}")
            self.running = False
            return False
    
    def stop(self):
        """Stop traffic generator"""
        if not self.running:
            return
        
        self.running = False
        time.sleep(0.5)  # Give thread time to stop
        
        self.rate_pps = 0
    
    def is_running(self):
        """Check if traffic generator is running"""
        return self.running
    
    def get_packet_count(self):
        """Get number of packets sent"""
        return self.packet_count
    
    def get_rate(self):
        """Get current rate in packets per second"""
        return self.rate_pps
    
    def get_actual_pps(self):
        """Get actual packets per second (moving window)"""
        return round(self.actual_pps, 1)
    
    def get_error_count(self):
        """Get number of errors"""
        return self.error_count

[francescopace]

Thanks for sharing this, @wangjushi930531-arch. These are very useful findings.

Could you also share some details about your test environment? That would help with reproducing the behavior:

• router brand/model and firmware version
• ISP-provided modem/router vs. standalone router
• NAT/firewall configuration (especially UDP timeout or session settings)
• any DNS security or filtering features enabled

Could you confirm whether you tested both modes available in the official ESPectre traffic generator script: broadcast mode and specific target device list mode?

One important note regarding the code you posted: the TCP connections do not appear to be persistent yet. At the moment, the loop performs socket() → connect() → close() on every iteration, which is closer to repeated short-lived TCP probes than to a persistent connection.

It would be helpful if you could open a PR (a draft is fine) so we can collaborate on this and move toward a design that:

• opens a single TCP socket
• keeps the connection alive
• sends minimal periodic payloads / heartbeats
• reconnects only on failure

If we move toward TCP transport, we will likely need coordinated changes on the device side as well to support TCP end-to-end, along with corresponding updates to configuration, documentation, and tests.

As an additional experiment, it might also be worth testing ICMP-based traffic generation (e.g., periodic pings to the device). Some routers treat ICMP differently from UDP, which might avoid part of the router-specific filtering behavior.

Another option could be to let users choose the traffic mode (UDP, TCP, or ICMP), or implement a fallback mechanism when the observed performance does not match the requested PPS.

[wangjushi930531-arch]

"Thanks for the suggestion. To clarify, I am currently testing on a TL-XDR3010 (易展版, Firmware v1.0.27). As this is a consumer-grade router, it does not allow manual configuration of NAT or firewall rules, so I cannot adjust those settings to test UDP behavior directly.

Regarding my current implementation: I am using short-lived TCP connections specifically to trigger a single data burst for CSI acquisition, rather than attempting to maintain a persistent long-lived connection. I haven't verified if a persistent connection would yield better results in this specific environment yet.

Moving forward, I plan to:

1. Experiment with ICMP-based traffic generation as an alternative.
2. Implement your suggested fallback mechanism: allowing users to select the traffic mode (UDP, TCP, or ICMP) or automatically switching protocols when the observed throughput falls below the requested PPS.

I will work on these modifications and submit a Pull Request (PR) once the changes are ready. Thanks again for the insightful advice!"

[wangjushi930531-arch]

I have discovered that short-lived TCP connections are unsuitable for this application. The inherent overhead of the TCP handshake and teardown processes introduces artifacts into the CSI data. Additionally, MicroPython’s limited support for ICMP complicates connectivity management.by bypassing the gateway IP (e.g., targeting 8.8.8.8 directly), the system operates normally without router-specific restrictions, provided the network infrastructure supports UDP traffic.
dest_addr = (“8.8.8.8”, 53)
Regarding fall detection, I need guidance on the optimal data collection strategy for training. Given a 30-second recording containing a single fall event and a window_size of 75, the dataset would be overwhelmingly dominated by non-fall samples, leading to severe class imbalance. Furthermore, are the CSI features distinct enough to reliably differentiate between a fall and normal walking patterns?

[francescopace]

Yes, that aligns with what I expected: short-lived TCP introduces significant overhead (handshake/teardown), which is exactly why I did not use it for the traffic generator.

In any case, if you found a stable workaround in your environment, that is great.

About fall detection: the current model is not suitable for that task. It was explicitly designed for binary motion detection (motion/no motion), not multi-class gesture/fall classification.

For gesture/fall classification, we need:

• a model with multi-class outputs (one class per gesture/event),
• a wider inference window (e.g., around 2 seconds), so the full movement fits inside the window.

If you want to experiment, there is an open branch I am working on: feature/gesture_detection.
You can start from there.

Current direction (still work in progress in that branch):

• collect many short samples per gesture (about 2s each), matching the inference window length,
• run inference every N ms (e.g., every 25 ms),
• confirm a gesture only after 2-3 consecutive positive predictions.