Enhance data handling and visualization in channel comparison and MAC integration examples; improve comments and remove unnecessary print statements in FEC tutorials.

Author: selimfirat

examples/channels/plot_channel_comparison.py

@@ -136,8 +136,20 @@ def apply_channels(data, channels, channel_names):
 fig, axes = plt.subplots(2, 3, figsize=(18, 10))
 axes = axes.flatten()
 
-# Add an "Original" plot
-axes[0].scatter(np.arange(100), continuous_data[:100], color=colors[0], alpha=0.7, s=40)
+
+# Add an "Original" plot with safe conversion
+def safe_data_conversion(data, slice_obj=None):
+    """Safely convert data to numpy array for plotting."""
+    if slice_obj is not None:
+        data = data[slice_obj]
+    data_np = data.numpy() if isinstance(data, torch.Tensor) else data
+    if np.iscomplexobj(data_np):
+        data_np = data_np.real
+    return data_np
+
+
+original_data_safe = safe_data_conversion(continuous_data, slice(100))
+axes[0].scatter(np.arange(100), original_data_safe, color=colors[0], alpha=0.7, s=40)
 axes[0].set_title("Original Signal", fontsize=14)
 axes[0].set_xlabel("Sample Index", fontsize=12)
 axes[0].set_ylabel("Signal Value", fontsize=12)
@@ -437,14 +449,45 @@ def complex_to_real(x):
 awgn_15db = AWGNChannel(snr_db=15)
 fading_10db = FlatFadingChannel(fading_type="rayleigh", coherence_time=1, snr_db=10)
 
+
+# Helper function to safely convert channel output to real-valued numpy array
+def safe_numpy_conversion(tensor_data):
+    """Convert tensor to numpy array, handling complex values properly.
+
+    This function ensures that any complex-valued outputs from channels
+    are properly converted to real values before being used in matplotlib
+    plots, preventing ComplexWarning messages.
+
+    Parameters
+    ----------
+    tensor_data : torch.Tensor
+        The tensor data to convert, which may contain complex values.
+
+    Returns
+    -------
+    numpy.ndarray
+        Real-valued numpy array suitable for plotting.
+    """
+    numpy_data = tensor_data.numpy()
+    if np.iscomplexobj(numpy_data):
+        # If complex, take the real part (constellation plots expect real I/Q components)
+        # Note: For constellation diagrams, the complex data should already be in I/Q format
+        # where real/imag parts are stored as separate columns, so this is a fallback
+        numpy_data = numpy_data.real
+        # Only warn if the imaginary part contains significant values
+        if tensor_data.dtype in [torch.complex64, torch.complex128]:
+            print(f"Info: Complex tensor converted to real part for plotting. " f"Shape: {numpy_data.shape}")
+    return numpy_data
+
+
 # Process data through channels
-qpsk_awgn_5db = awgn_5db(qpsk_data).numpy()
-qpsk_awgn_15db = awgn_15db(qpsk_data).numpy()
-qpsk_fading = fading_10db(qpsk_data).numpy()
+qpsk_awgn_5db = safe_numpy_conversion(awgn_5db(qpsk_data))
+qpsk_awgn_15db = safe_numpy_conversion(awgn_15db(qpsk_data))
+qpsk_fading = safe_numpy_conversion(fading_10db(qpsk_data))
 
-qam_awgn_5db = awgn_5db(qam_data).numpy()
-qam_awgn_15db = awgn_15db(qam_data).numpy()
-qam_fading = fading_10db(qam_data).numpy()
+qam_awgn_5db = safe_numpy_conversion(awgn_5db(qam_data))
+qam_awgn_15db = safe_numpy_conversion(awgn_15db(qam_data))
+qam_fading = safe_numpy_conversion(fading_10db(qam_data))
 
 # Create constellation plots
 fig, axes = plt.subplots(2, 4, figsize=(20, 10))

examples/constraints/plot_basic_constraints.py

@@ -336,7 +336,6 @@
 avg_power_signals = {name: torch.tensor(data).reshape(1, -1) for name, data in avg_power_results.items()}
 
 # Compare constraints side by side
-print("\n=== Signal Properties Comparison ===")
 fig, axes = plt.subplots(2, 2, figsize=(15, 10), constrained_layout=True)
 fig.suptitle("Signal Properties Analysis - All Constraints", fontsize=16, fontweight="bold")
 

examples/models/plot_afmodule.py

@@ -314,19 +314,22 @@ def forward(self, x, snr):
 
 # %%
 # Conclusion
-# ---------------
+# ----------
+#
 # In this example, we explored the Attention-Feature Module (AFModule), a component
 # designed to help neural networks adapt to varying channel conditions in wireless
 # communication systems.
 #
-# Key points:
-# - AFModule recalibrates feature maps based on channel state information
-# - It can work with different input tensor dimensions (2D, 3D, 4D)
-# - It helps maintain performance across different channel conditions (like varying SNRs)
-# - The module can adapt to different feature sizes dynamically
+# **Key Points:**
+#
+# * AFModule recalibrates feature maps based on channel state information
+# * It can work with different input tensor dimensions (2D, 3D, 4D)
+# * It helps maintain performance across different channel conditions (like varying SNRs)
+# * The module can adapt to different feature sizes dynamically
 #
 # The AFModule is particularly useful in deep learning-based communication systems
 # that need to operate reliably in varying channel conditions.
 #
-# References:
-# - :cite:`xu2021wireless`
+# **References:**
+#
+# * :cite:`xu2021wireless`

examples/models/plot_channel_aware_base_model.py

@@ -81,10 +81,10 @@ def forward(self, x: torch.Tensor, csi: torch.Tensor, *args, **kwargs) -> torch.
 csi_dim = 2
 model = SimpleChannelAwareEncoder(input_dim, output_dim, csi_dim)
 
-print("Created SimpleChannelAwareEncoder:")
-print(f"  Input dimension: {input_dim}")
-print(f"  Output dimension: {output_dim}")
-print(f"  CSI dimension: {csi_dim}")
+# Created SimpleChannelAwareEncoder:
+# Input dimension: {input_dim}
+# Output dimension: {output_dim}
+# CSI dimension: {csi_dim}
 
 # %%
 # Demonstrating CSI Validation and Normalization
@@ -96,81 +96,79 @@ def forward(self, x: torch.Tensor, csi: torch.Tensor, *args, **kwargs) -> torch.
 x = torch.randn(batch_size, input_dim)
 
 # Different CSI examples
-print("\n=== CSI Validation Examples ===")
-
 # Valid CSI
 csi_valid = torch.tensor([[10.0, 0.5], [15.0, 0.8], [5.0, 0.3], [20.0, 0.9], [12.0, 0.6], [8.0, 0.4], [18.0, 0.7], [14.0, 0.65]])
-print(f"Valid CSI shape: {csi_valid.shape}")
+# Valid CSI shape: {csi_valid.shape}
 
 try:
     validated_csi = model.validate_csi(csi_valid)
     print("✓ CSI validation passed")
+except ValueError as e:
+    # ✗ CSI validation failed due to value error
+    print("✗ CSI validation failed: " + str(e))
+    validated_csi = None
 except Exception as e:
-    print(f"✗ CSI validation failed: {e}")
+    # ✗ CSI validation failed due to unexpected error
+    print("✗ Unexpected error during CSI validation: " + str(e))
+    validated_csi = None
 
 # Test normalization
-print("\n=== CSI Normalization Examples ===")
-
 # MinMax normalization
 normalized_minmax = model.normalize_csi(csi_valid, method="minmax", target_range=(0, 1))
-print(f"Original CSI range: [{csi_valid.min():.2f}, {csi_valid.max():.2f}]")
-print(f"MinMax normalized range: [{normalized_minmax.min():.2f}, {normalized_minmax.max():.2f}]")
+# Original CSI range: [{csi_valid.min():.2f}, {csi_valid.max():.2f}]
+# MinMax normalized range: [{normalized_minmax.min():.2f}, {normalized_minmax.max():.2f}]
 
 # Z-score normalization
 normalized_zscore = model.normalize_csi(csi_valid, method="zscore")
-print(f"Z-score normalized mean: {normalized_zscore.mean():.4f}, std: {normalized_zscore.std():.4f}")
+# Z-score normalized mean: {normalized_zscore.mean():.4f}, std: {normalized_zscore.std():.4f}
 
 # %%
 # Working with the AFModule
 # -------------------------
 # The AFModule has been updated to use ChannelAwareBaseModel. Let's demonstrate its usage.
 
-print("\n=== AFModule (Channel-Aware) Example ===")
-
 # Create AFModule
 N = 64  # Number of feature channels
 csi_length = 1  # CSI vector length
 af_module = AFModule(N=N, csi_length=csi_length)
 
 # Create feature maps (4D tensor for image-like data)
 feature_maps = torch.randn(batch_size, N, 8, 8)
-print(f"Feature maps shape: {feature_maps.shape}")
+# Feature maps shape: {feature_maps.shape}
 
 # Create CSI for AFModule (SNR values in dB)
 snr_values = torch.tensor([10.0, 15.0, 5.0, 20.0, 12.0, 8.0, 18.0, 14.0])
 csi_af = snr_values.unsqueeze(1)  # Shape: [batch_size, 1]
-print(f"CSI shape for AFModule: {csi_af.shape}")
+# CSI shape for AFModule: {csi_af.shape}
 
 # Apply AFModule
 with torch.no_grad():
     modulated_features = af_module(feature_maps, csi=csi_af)
 
-print(f"Modulated features shape: {modulated_features.shape}")
-print(f"Feature modulation factor range: [{(modulated_features/feature_maps).min():.3f}, {(modulated_features/feature_maps).max():.3f}]")
+# Modulated features shape: {modulated_features.shape}
+# Feature modulation factor range: [{(modulated_features/feature_maps).min():.3f}, {(modulated_features/feature_maps).max():.3f}]
 
 # %%
 # CSI Feature Extraction
 # ----------------------
 # The base class provides methods to extract useful features from CSI.
 
-print("\n=== CSI Feature Extraction ===")
-
 csi_features = model.extract_csi_features(csi_valid)
-print("Extracted CSI features:")
+# Extracted CSI features:
 for feature_name, feature_value in csi_features.items():
     if isinstance(feature_value, torch.Tensor):
         if feature_value.numel() == 1:
-            print(f"  {feature_name}: {feature_value.item():.4f}")
+            # Single-value feature: {feature_name}: {feature_value.item():.4f}
+            pass
         else:
-            print(f"  {feature_name}: {feature_value.tolist()}")
+            # Multi-value feature: {feature_name}: {feature_value.tolist()}
+            pass
 
 # %%
 # Visualization of CSI Effects
 # ----------------------------
 # Let's visualize how different CSI values affect model outputs.
 
-print("\n=== Visualizing CSI Effects ===")
-
 # Generate a range of CSI values
 snr_range = torch.linspace(-5, 25, 31)  # SNR from -5 to 25 dB
 quality_factor = torch.ones_like(snr_range) * 0.5  # Fixed quality factor
@@ -204,7 +202,7 @@ def forward(self, x: torch.Tensor, csi: torch.Tensor, *args, **kwargs) -> torch.
 # Plot 2: First few output dimensions vs SNR
 actual_output_dim = outputs.shape[1] if outputs.ndim > 1 else 1
 for i in range(min(4, actual_output_dim)):
-    axes[0, 1].plot(snr_range.numpy(), outputs[:, i], label=f"Dim {i+1}")
+    axes[0, 1].plot(snr_range.numpy(), outputs[:, i], label="Dim " + str(i + 1))
 axes[0, 1].set_xlabel("SNR (dB)")
 axes[0, 1].set_ylabel("Output Value")
 axes[0, 1].set_title("Output Dimensions vs CSI (SNR)")
@@ -260,8 +258,6 @@ def forward(self, x: torch.Tensor, csi: torch.Tensor, *args, **kwargs) -> torch.
 # ---------------------------------
 # Demonstrate how to extract and use CSI from channel outputs.
 
-print("\n=== Integration with Channels ===")
-
 # Create channels that might provide CSI
 awgn_channel = AWGNChannel(snr_db=15.0)
 fading_channel = FlatFadingChannel(fading_type="rayleigh", coherence_time=50, snr_db=10.0)
@@ -273,24 +269,24 @@ def forward(self, x: torch.Tensor, csi: torch.Tensor, *args, **kwargs) -> torch.
 awgn_output = awgn_channel(test_signal)
 fading_output = fading_channel(test_signal)
 
-print(f"AWGN channel output shape: {awgn_output.shape}")
-print(f"Fading channel output shape: {fading_output.shape}")
+# AWGN channel output shape: {awgn_output.shape}
+# Fading channel output shape: {fading_output.shape}
 
 # Try to extract CSI (channels might not provide it directly)
 awgn_csi = model.extract_csi_from_channel_output(awgn_output)
 fading_csi = model.extract_csi_from_channel_output(fading_output)
 
-print(f"Extracted CSI from AWGN: {awgn_csi}")
-print(f"Extracted CSI from Fading: {fading_csi}")
+# Extracted CSI from AWGN: {awgn_csi}
+# Extracted CSI from Fading: {fading_csi}
 
 # Create CSI manually based on channel properties
 if hasattr(awgn_channel, "snr_db"):
     manual_awgn_csi = torch.full((batch_size, 1), awgn_channel.snr_db)
-    print(f"Manual AWGN CSI: {manual_awgn_csi.mean().item():.1f} dB")
+    print("Manual AWGN CSI: " + str(manual_awgn_csi.mean().item()) + " dB")
 
 if hasattr(fading_channel, "snr_db"):
     manual_fading_csi = torch.full((batch_size, 1), fading_channel.snr_db)
-    print(f"Manual Fading CSI: {manual_fading_csi.mean().item():.1f} dB")
+    print("Manual Fading CSI: " + str(manual_fading_csi.mean().item()) + " dB")
 
 # %%
 # Best Practices Summary

examples/models/plot_uplink_mac_integration.py

@@ -87,7 +87,7 @@ def forward(self, user_messages, **kwargs):
 # %%
 # Setup Parameters
 # --------------------------------------------------------------------------
-print("=== UplinkMACChannel Integration Example ===\n")
+# === UplinkMACChannel Integration Example ===
 
 # System parameters
 num_users = 3
@@ -100,19 +100,19 @@ def forward(self, user_messages, **kwargs):
 avg_noise_power = 0.1
 snr_db = 12.0
 
-print("System Configuration:")
-print(f"- Number of users: {num_users}")
-print(f"- Message dimension: {message_dim}")
-print(f"- Code dimension: {code_dim}")
-print(f"- Batch size: {batch_size}")
-print(f"- SNR: {snr_db} dB")
-print(f"- Coherence time: {coherence_time}")
-print(f"- Average noise power: {avg_noise_power}\n")
+# System Configuration:
+# - Number of users: {num_users}
+# - Message dimension: {message_dim}
+# - Code dimension: {code_dim}
+# - Batch size: {batch_size}
+# - SNR: {snr_db} dB
+# - Coherence time: {coherence_time}
+# - Average noise power: {avg_noise_power}
 
 # %%
 # Scenario 1: Shared Channel Configuration
 # --------------------------------------------------------------------------
-print("--- Scenario 1: Shared Channel Configuration ---")
+# --- Scenario 1: Shared Channel Configuration ---
 
 # Create a shared Rayleigh fading channel for all users
 shared_channel = RayleighFadingChannel(coherence_time=coherence_time, avg_noise_power=avg_noise_power)
@@ -147,15 +147,15 @@ def forward(self, user_messages, **kwargs):
 for i in range(num_users):
     user_mse = mse_loss(reconstructed_messages[i], user_messages[i]).item()
     total_mse_shared += user_mse
-    print(f"User {i+1} MSE (Shared Channel): {user_mse:.6f}")
+    # User {i+1} MSE (Shared Channel): {user_mse:.6f}
 
 avg_mse_shared = total_mse_shared / num_users
-print(f"Average MSE (Shared Channel): {avg_mse_shared:.6f}\n")
+# Average MSE (Shared Channel): {avg_mse_shared:.6f}
 
 # %%
 # Scenario 2: Per-User Channel Configuration
 # --------------------------------------------------------------------------
-print("--- Scenario 2: Per-User Channel Configuration ---")
+# --- Scenario 2: Per-User Channel Configuration ---
 
 # Create individual channels for each user with different characteristics
 per_user_channels = [
@@ -187,21 +187,21 @@ def forward(self, user_messages, **kwargs):
 for i in range(num_users):
     user_mse = mse_loss(reconstructed_messages_per_user[i], user_messages[i]).item()
     total_mse_per_user += user_mse
-    print(f"User {i+1} MSE (Per-User Channel): {user_mse:.6f}")
+    # User {i+1} MSE (Per-User Channel): {user_mse:.6f}
 
 avg_mse_per_user = total_mse_per_user / num_users
-print(f"Average MSE (Per-User Channel): {avg_mse_per_user:.6f}\n")
+# Average MSE (Per-User Channel): {avg_mse_per_user:.6f}
 
 # %%
 # Scenario 3: Combining Methods Comparison
 # --------------------------------------------------------------------------
-print("--- Scenario 3: Combining Methods Comparison ---")
+# --- Scenario 3: Combining Methods Comparison ---
 
 combining_methods = ["sum", "weighted_sum"]
 combine_results = {}
 
 for method in combining_methods:
-    print(f"Testing combining method: {method}")
+    # Testing combining method: {method}
 
     # Create channels for this test
     test_channels = [FlatFadingChannel(fading_type="rayleigh", coherence_time=coherence_time, avg_noise_power=avg_noise_power) for _ in range(num_users)]
@@ -230,12 +230,12 @@ def forward(self, user_messages, **kwargs):
 
     avg_mse_test = total_mse_test / num_users
     combine_results[method] = avg_mse_test
-    print(f"  Average MSE with {method}: {avg_mse_test:.6f}")
+    # Average MSE with {method}: {avg_mse_test:.6f}
 
 # %%
 # Scenario 4: Dynamic Parameter Updates
 # --------------------------------------------------------------------------
-print("\n--- Scenario 4: Dynamic Parameter Updates ---")
+# --- Scenario 4: Dynamic Parameter Updates ---
 
 # Create UplinkMACChannel for dynamic updates
 dynamic_channels = [FlatFadingChannel(fading_type="rayleigh", coherence_time=coherence_time, avg_noise_power=avg_noise_power) for _ in range(num_users)]
@@ -254,7 +254,7 @@ def forward(self, user_messages, **kwargs):
 mse_results_dynamic = []
 
 for snr in snr_values:
-    print(f"Testing at SNR = {snr} dB...")
+    # Testing at SNR = {snr} dB...
 
     # Update individual channel parameters dynamically
     new_noise_power = 0.1 * (10 ** (-snr / 10))  # Adjust noise based on SNR
@@ -282,7 +282,7 @@ def forward(self, user_messages, **kwargs):
 
     avg_mse_dynamic = total_mse_dynamic / num_users
     mse_results_dynamic.append(avg_mse_dynamic)
-    print(f"  Average MSE: {avg_mse_dynamic:.6f}")
+    # Average MSE: {avg_mse_dynamic:.6f}
 
 # %%
 # Plotting Results