src/ria_toolkit_oss/io/recording.py

@@ -267,13 +267,6 @@ def to_sigmf(
         SigMFFile.DATATYPE_KEY: get_data_type_str(samples),
         SigMFFile.VERSION_KEY: sigmf.__version__,
         SigMFFile.RECORDER_KEY: "RIA",
-        SigMFFile.EXTENSIONS_KEY: [
-            {
-                "name": "ria",
-                "version": "0.1.4",
-                "optional": True,
-            }
-        ],
     }
 
     converted_metadata = {

src/ria_toolkit_oss/ria_toolkit_oss_cli/ria_toolkit_oss/generate.py

@@ -183,7 +183,7 @@ class FileSourceBlock(SourceBlock):
 
 
 def apply_post_processing(
-    recording: Recording, frequency_shift: float, add_noise: str, channel_params: dict, verbose: bool
+    recording: Recording, frequency_shift: float, channel_type: str, channel_params: dict, verbose: bool
 ) -> Recording:
     """Apply frequency shift and channel models to a recording."""
 
@@ -200,7 +200,57 @@ def apply_post_processing(
         processed = fs_block.get_samples(num)
         recording = Recording(data=processed, metadata=recording.metadata)
 
-    # 2.  IQ Imbalance
+    # 2. Dynamic Impairments (Transforms)
+
+    # Rician / Rayleigh
+    if channel_type == "rayleigh":
+        # Use improved complex multipath if available
+        echo_verbose("Applying Multipath Rayleigh Channel", verbose)
+        recording = complex_multipath_rayleigh_channel(
+            recording,
+            num_paths=channel_params.get("multipath_paths") or 3,
+            max_delay=channel_params.get("multipath_max_delay") or 2.6e-6,
+            sample_rate=recording.sample_rate,
+            snr_db=None,  # We handle noise separately
+        )
+
+    elif channel_type == "rician":
+        echo_verbose(f"Applying Rician Channel (K={channel_params.get('rician_k', 2.0)})", verbose)
+        recording = rician_fading_channel(
+            recording,
+            k_factor=channel_params.get("rician_k", 2.0),
+            num_paths=channel_params.get("multipath_paths") or 3,
+            max_delay=channel_params.get("multipath_max_delay") or 1.2e-6,
+            sample_rate=recording.sample_rate,
+            snr_db=None,
+        )
+
+    # Doppler
+    doppler_freq = channel_params.get("doppler_freq")
+    if doppler_freq:
+        echo_verbose(f"Applying Doppler (Shift={doppler_freq} Hz)", verbose)
+        # add_doppler expects velocity. Convert freq to velocity assuming 1GHz carrier or pass freq directly?
+        # dynamic_channel wrapper handles this conversion.
+        # Or use add_doppler directly if we have velocity.
+        # User supplied doppler_freq.
+        # Let's use a simple transform or dynamic_channel
+        # We need to reuse dynamic_channel logic for freq->velocity conversion or assume carrier.
+        # Or create add_doppler_freq(signal, freq_shift)
+        # add_doppler takes satellite_velocity etc.
+        # dynamic_channel takes doppler_hz.
+        # We use dynamic_channel logic here but just for Doppler part
+        c_light = 299792458
+        f_carrier = 1e9  # Assumption for conversion
+        velocity = doppler_freq * c_light / f_carrier
+        recording = add_doppler(
+            recording,
+            satellite_velocity=velocity,
+            satellite_initial_distance=1000,
+            frequency=f_carrier,
+            sample_rate=recording.sample_rate,
+        )
+
+    # IQ Imbalance
     amp = channel_params.get("iq_amp_imbalance")
     phase = channel_params.get("iq_phase_imbalance")
     dc = channel_params.get("iq_dc_offset")
@@ -215,15 +265,61 @@ def apply_post_processing(
             dc_offset=dc if dc is not None else 0,
         )
 
-    # 3. AWGN (Final stage usually)
+    # Phase Noise
-    if add_noise == "awgn":
+    pn = channel_params.get("phase_noise")
-        npow = channel_params.get("noise_power", 0.1)
+    if pn:
-        echo_verbose(f"Applying AWGN (Power={npow})", verbose)
+        echo_verbose(f"Applying Phase Noise (Var={pn})", verbose)
+        recording = add_phase_noise(recording, phase_variance=pn)
 
-        # Use AWGNChannel block logic directly
+    # Gain Fluctuation
-        noise_std = np.sqrt(npow / 2)
+    gf = channel_params.get("gain_fluctuation")
-        noise = noise_std * (np.random.randn(*recording.data.shape) + 1j * np.random.randn(*recording.data.shape))
+    if gf:
-        recording = Recording(data=recording.data + noise, metadata=recording.metadata)
+        echo_verbose(f"Applying Gain Fluctuation (Var={gf})", verbose)
+        recording = add_gain_fluctuation(recording, amplitude_variance=gf)
+
+    # Compression
+    comp = channel_params.get("compression")
+    if comp:
+        echo_verbose(f"Applying Compression (Gain={comp})", verbose)
+        recording = add_compression(recording, compression_gain=comp)
+
+    # 3. AWGN (Final stage usually)
+    if channel_type == "awgn" or channel_params.get("noise_power"):
+        # If 'awgn' selected OR noise_power explicitly set (default is 0.1, so always set?)
+        # If channel_type is NOT awgn/rayleigh/rician, and noise_power is default 0.1?
+        # If user didn't specify noise_power, but did specify channel_type=none, do we add noise?
+        # Default noise_power is 0.1.
+        # If channel_type == 'none', we probably shouldn't add noise unless user asked for it.
+        # But noise_power has default.
+        # Let's check if channel_type is 'awgn'.
+        # Or if user provided --noise-power?
+        # (We can't distinguish default vs user provided easily with click unless we use ctx)
+        # For now: only add noise if channel_type is set to something, or if noise_power > 0 and user intended it.
+        # Simpler: If channel_type == 'awgn', definitely add.
+        # If rayleigh/rician, they might want noise too.
+        # If 'none', skip noise?
+
+        should_add_noise = False
+        if channel_type in ["awgn", "rayleigh", "rician"]:
+            should_add_noise = True
+
+        if should_add_noise:
+            npow = channel_params.get("noise_power", 0.1)
+            echo_verbose(f"Applying AWGN (Power={npow})", verbose)
+            # Convert Power (variance) to SNR?
+            # add_awgn_to_signal takes SNR.
+            # AWGNChannel block takes Variance.
+            # Use AWGNChannel block logic (additive noise with variance)
+            # or utils.transforms.iq_channel_models.awgn_channel which takes SNR.
+            # The user CLI says --noise-power (variance).
+            # We should use a simple additive noise function with variance.
+            # transforms.iq_augmentations.generate_awgn uses SNR.
+            # Let's implement simple additive noise here or use AWGNChannel block.
+
+            # Use AWGNChannel block logic directly
+            noise_std = np.sqrt(npow / 2)
+            noise = noise_std * (np.random.randn(*recording.data.shape) + 1j * np.random.randn(*recording.data.shape))
+            recording = Recording(data=recording.data + noise, metadata=recording.metadata)
 
     return recording
 
@@ -250,18 +346,25 @@ def common_options(f):
         f
     )
     f = click.option("--center-frequency", "-fc", type=float, help="Metadata center frequency (Hz)")(f)
-    f = click.option("--add-noise", is_flag=True, help="Add noise to signal")(f)
+    f = click.option(
+        "--channel-type", type=click.Choice(["none", "awgn", "rayleigh"]), default="none", help="Channel model"
+    )(f)
     f = click.option("--noise-power", type=float, default=0.1, help="Noise power (variance) for AWGN")(f)
     f = click.option("--path-gain", type=float, default=0.0, help="Path gain (dB) for Rayleigh")(f)
     f = click.option("--output", "-o", required=True, help="Output filename")(f)
     f = click.option("--format", "-F", type=click.Choice(["npy", "sigmf", "wav", "blue"]), help="Output format")(f)
 
     # Impairment options
+    f = click.option("--rician-k", type=float, help="Rician K-factor")(f)
     f = click.option("--multipath-paths", type=int, help="Multipath: Number of paths")(f)
     f = click.option("--multipath-max-delay", type=float, help="Multipath: Max delay (s)")(f)
+    f = click.option("--doppler-freq", type=float, help="Doppler: Frequency shift (Hz)")(f)
     f = click.option("--iq-amp-imbalance", type=float, help="IQ Imbalance: Amplitude (dB)")(f)
     f = click.option("--iq-phase-imbalance", type=float, help="IQ Imbalance: Phase (rad)")(f)
     f = click.option("--iq-dc-offset", type=float, help="IQ Imbalance: DC Offset")(f)
+    f = click.option("--phase-noise", type=float, help="Phase Noise: Variance")(f)
+    f = click.option("--gain-fluctuation", type=float, help="Gain Fluctuation: Variance")(f)
+    f = click.option("--compression", type=float, help="Compression: Gain")(f)
 
     f = click.option(
         "--config",
@@ -310,7 +413,7 @@ def tone(
     duration,
     frequency_shift,
     center_frequency,
-    add_noise,
+    channel_type,
     noise_power,
     path_gain,
     output,
@@ -341,7 +444,7 @@ def tone(
 
     # Post processing
     chan_params = {"noise_power": noise_power, "path_gain": path_gain}
-    recording = apply_post_processing(recording, frequency_shift, add_noise, chan_params, verbose)
+    recording = apply_post_processing(recording, frequency_shift, channel_type, chan_params, verbose)
 
     # User metadata
     metadata = apply_user_config_metadata(metadata)
@@ -363,7 +466,7 @@ def noise(
     duration,
     frequency_shift,
     center_frequency,
-    add_noise,
+    channel_type,
     noise_power,
     path_gain,
     output,
@@ -401,7 +504,7 @@ def noise(
 
     # Post processing
     chan_params = {"noise_power": noise_power, "path_gain": path_gain}
-    recording = apply_post_processing(recording, frequency_shift, add_noise, chan_params, verbose)
+    recording = apply_post_processing(recording, frequency_shift, channel_type, chan_params, verbose)
 
     for key, value in apply_user_config_metadata(metadata).items():
         recording.update_metadata(key, value)
@@ -420,7 +523,7 @@ def chirp(
     duration,
     frequency_shift,
     center_frequency,
-    add_noise,
+    channel_type,
     noise_power,
     path_gain,
     output,
@@ -454,7 +557,7 @@ def chirp(
 
     # Post processing
     chan_params = {"noise_power": noise_power, "path_gain": path_gain}
-    recording = apply_post_processing(recording, frequency_shift, add_noise, chan_params, verbose)
+    recording = apply_post_processing(recording, frequency_shift, channel_type, chan_params, verbose)
 
     for key, value in apply_user_config_metadata(metadata).items():
         recording.update_metadata(key, value)
@@ -474,7 +577,7 @@ def square(
     duration,
     frequency_shift,
     center_frequency,
-    add_noise,
+    channel_type,
     noise_power,
     path_gain,
     output,
@@ -506,7 +609,7 @@ def square(
         recording._metadata["center_frequency"] = center_frequency
 
     chan_params = {"noise_power": noise_power, "path_gain": path_gain}
-    recording = apply_post_processing(recording, frequency_shift, add_noise, chan_params, verbose)
+    recording = apply_post_processing(recording, frequency_shift, channel_type, chan_params, verbose)
 
     for key, value in apply_user_config_metadata(metadata).items():
         recording.update_metadata(key, value)
@@ -525,7 +628,7 @@ def sawtooth(
     duration,
     frequency_shift,
     center_frequency,
-    add_noise,
+    channel_type,
     noise_power,
     path_gain,
     output,
@@ -554,7 +657,7 @@ def sawtooth(
         recording._metadata["center_frequency"] = center_frequency
 
     chan_params = {"noise_power": noise_power, "path_gain": path_gain}
-    recording = apply_post_processing(recording, frequency_shift, add_noise, chan_params, verbose)
+    recording = apply_post_processing(recording, frequency_shift, channel_type, chan_params, verbose)
 
     for key, value in apply_user_config_metadata(metadata).items():
         recording.update_metadata(key, value)
@@ -615,7 +718,7 @@ def _run_mod_gen(
     message_content,
     frequency_shift,
     center_frequency,
-    add_noise,
+    channel_type,
     noise_power,
     path_gain,
     output,
@@ -699,7 +802,7 @@ def _run_mod_gen(
 
     # Post Processing
     chan_params = {"noise_power": noise_power, "path_gain": path_gain}
-    final_recording = apply_post_processing(base_recording, frequency_shift, add_noise, chan_params, verbose)
+    final_recording = apply_post_processing(base_recording, frequency_shift, channel_type, chan_params, verbose)
 
     # Trim if explicit num_samples was requested and we generated more (due to symbol alignment)
     target_ns = resolve_length(sample_rate, num_samples, duration)
@@ -739,7 +842,7 @@ def qam(
     duration,
     frequency_shift,
     center_frequency,
-    add_noise,
+    channel_type,
     noise_power,
     path_gain,
     output,
@@ -748,11 +851,16 @@ def qam(
     metadata,
     verbose,
     quiet,
+    rician_k,
     multipath_paths,
     multipath_max_delay,
+    doppler_freq,
     iq_amp_imbalance,
     iq_phase_imbalance,
     iq_dc_offset,
+    phase_noise,
+    gain_fluctuation,
+    compression,
     symbols,
     order,
     symbol_rate,
@@ -786,7 +894,7 @@ def qam(
         message_content,
         frequency_shift,
         center_frequency,
-        add_noise,
+        channel_type,
         noise_power,
         path_gain,
         output,
@@ -822,7 +930,7 @@ def apsk(
     duration,
     frequency_shift,
     center_frequency,
-    add_noise,
+    channel_type,
     noise_power,
     path_gain,
     output,
@@ -831,11 +939,16 @@ def apsk(
     metadata,
     verbose,
     quiet,
+    rician_k,
     multipath_paths,
     multipath_max_delay,
+    doppler_freq,
     iq_amp_imbalance,
     iq_phase_imbalance,
     iq_dc_offset,
+    phase_noise,
+    gain_fluctuation,
+    compression,
     symbols,
     order,
     symbol_rate,
@@ -862,7 +975,7 @@ def apsk(
         message_content,
         frequency_shift,
         center_frequency,
-        add_noise,
+        channel_type,
         noise_power,
         path_gain,
         output,
@@ -898,7 +1011,7 @@ def pam(
     duration,
     frequency_shift,
     center_frequency,
-    add_noise,
+    channel_type,
     noise_power,
     path_gain,
     output,
@@ -907,11 +1020,16 @@ def pam(
     metadata,
     verbose,
     quiet,
+    rician_k,
     multipath_paths,
     multipath_max_delay,
+    doppler_freq,
     iq_amp_imbalance,
     iq_phase_imbalance,
     iq_dc_offset,
+    phase_noise,
+    gain_fluctuation,
+    compression,
     symbols,
     order,
     symbol_rate,
@@ -938,7 +1056,7 @@ def pam(
         message_content,
         frequency_shift,
         center_frequency,
-        add_noise,
+        channel_type,
         noise_power,
         path_gain,
         output,
@@ -967,7 +1085,7 @@ def fsk(
     duration,
     frequency_shift,
     center_frequency,
-    add_noise,
+    channel_type,
     noise_power,
     path_gain,
     output,
@@ -976,11 +1094,16 @@ def fsk(
     metadata,
     verbose,
     quiet,
+    rician_k,
     multipath_paths,
     multipath_max_delay,
+    doppler_freq,
     iq_amp_imbalance,
     iq_phase_imbalance,
     iq_dc_offset,
+    phase_noise,
+    gain_fluctuation,
+    compression,
     symbols,
     order,
     symbol_rate,
@@ -1049,14 +1172,19 @@ def fsk(
     chan_params = {
         "noise_power": noise_power,
         "path_gain": path_gain,
+        "rician_k": rician_k,
         "multipath_paths": multipath_paths,
         "multipath_max_delay": multipath_max_delay,
+        "doppler_freq": doppler_freq,
         "iq_amp_imbalance": iq_amp_imbalance,
         "iq_phase_imbalance": iq_phase_imbalance,
         "iq_dc_offset": iq_dc_offset,
+        "phase_noise": phase_noise,
+        "gain_fluctuation": gain_fluctuation,
+        "compression": compression,
     }
 
-    recording = apply_post_processing(recording, frequency_shift, add_noise, chan_params, verbose)
+    recording = apply_post_processing(recording, frequency_shift, channel_type, chan_params, verbose)
 
     for key, value in apply_user_config_metadata(metadata).items():
         recording.update_metadata(key, value)
@@ -1077,7 +1205,7 @@ def ook(
     duration,
     frequency_shift,
     center_frequency,
-    add_noise,
+    channel_type,
     noise_power,
     path_gain,
     output,
@@ -1086,11 +1214,16 @@ def ook(
     metadata,
     verbose,
     quiet,
+    rician_k,
     multipath_paths,
     multipath_max_delay,
+    doppler_freq,
     iq_amp_imbalance,
     iq_phase_imbalance,
     iq_dc_offset,
+    phase_noise,
+    gain_fluctuation,
+    compression,
     symbol_rate,
     message_source,
     message_content,
@@ -1118,14 +1251,19 @@ def ook(
     chan_params = {
         "noise_power": noise_power,
         "path_gain": path_gain,
+        "rician_k": rician_k,
         "multipath_paths": multipath_paths,
         "multipath_max_delay": multipath_max_delay,
+        "doppler_freq": doppler_freq,
         "iq_amp_imbalance": iq_amp_imbalance,
         "iq_phase_imbalance": iq_phase_imbalance,
         "iq_dc_offset": iq_dc_offset,
+        "phase_noise": phase_noise,
+        "gain_fluctuation": gain_fluctuation,
+        "compression": compression,
     }
 
-    recording = apply_post_processing(recording, frequency_shift, add_noise, chan_params, verbose)
+    recording = apply_post_processing(recording, frequency_shift, channel_type, chan_params, verbose)
 
     for key, value in apply_user_config_metadata(metadata).items():
         recording.update_metadata(key, value)
@@ -1146,7 +1284,7 @@ def oqpsk(
     duration,
     frequency_shift,
     center_frequency,
-    add_noise,
+    channel_type,
     noise_power,
     path_gain,
     output,
@@ -1155,11 +1293,16 @@ def oqpsk(
     metadata,
     verbose,
     quiet,
+    rician_k,
     multipath_paths,
     multipath_max_delay,
+    doppler_freq,
     iq_amp_imbalance,
     iq_phase_imbalance,
     iq_dc_offset,
+    phase_noise,
+    gain_fluctuation,
+    compression,
     symbol_rate,
     message_source,
     message_content,
@@ -1187,14 +1330,19 @@ def oqpsk(
     chan_params = {
         "noise_power": noise_power,
         "path_gain": path_gain,
+        "rician_k": rician_k,
         "multipath_paths": multipath_paths,
         "multipath_max_delay": multipath_max_delay,
+        "doppler_freq": doppler_freq,
         "iq_amp_imbalance": iq_amp_imbalance,
         "iq_phase_imbalance": iq_phase_imbalance,
         "iq_dc_offset": iq_dc_offset,
+        "phase_noise": phase_noise,
+        "gain_fluctuation": gain_fluctuation,
+        "compression": compression,
     }
 
-    recording = apply_post_processing(recording, frequency_shift, add_noise, chan_params, verbose)
+    recording = apply_post_processing(recording, frequency_shift, channel_type, chan_params, verbose)
 
     for key, value in apply_user_config_metadata(metadata).items():
         recording.update_metadata(key, value)
@@ -1216,7 +1364,7 @@ def gmsk(
     duration,
     frequency_shift,
     center_frequency,
-    add_noise,
+    channel_type,
     noise_power,
     path_gain,
     output,
@@ -1225,11 +1373,16 @@ def gmsk(
     metadata,
     verbose,
     quiet,
+    rician_k,
     multipath_paths,
     multipath_max_delay,
+    doppler_freq,
     iq_amp_imbalance,
     iq_phase_imbalance,
     iq_dc_offset,
+    phase_noise,
+    gain_fluctuation,
+    compression,
     symbol_rate,
     bt,
     message_source,
@@ -1259,14 +1412,19 @@ def gmsk(
     chan_params = {
         "noise_power": noise_power,
         "path_gain": path_gain,
+        "rician_k": rician_k,
         "multipath_paths": multipath_paths,
         "multipath_max_delay": multipath_max_delay,
+        "doppler_freq": doppler_freq,
         "iq_amp_imbalance": iq_amp_imbalance,
         "iq_phase_imbalance": iq_phase_imbalance,
         "iq_dc_offset": iq_dc_offset,
+        "phase_noise": phase_noise,
+        "gain_fluctuation": gain_fluctuation,
+        "compression": compression,
     }
 
-    recording = apply_post_processing(recording, frequency_shift, add_noise, chan_params, verbose)
+    recording = apply_post_processing(recording, frequency_shift, channel_type, chan_params, verbose)
 
     for key, value in apply_user_config_metadata(metadata).items():
         recording.update_metadata(key, value)
@@ -1298,7 +1456,7 @@ def psk(
     duration,
     frequency_shift,
     center_frequency,
-    add_noise,
+    channel_type,
     noise_power,
     path_gain,
     output,
@@ -1333,7 +1491,7 @@ def psk(
         message_content,
         frequency_shift,
         center_frequency,
-        add_noise,
+        channel_type,
         noise_power,
         path_gain,
         output,
@@ -1343,3 +1501,75 @@ def psk(
         verbose,
         quiet,
     )
+
+
+@generate.command()
+@click.option("--bandwidth", "-b", type=int, required=True, help="Bandwidth in MHz (e.g. 10, 20)")
+@click.option("--mu", "-u", type=int, default=1, help="Numerology (0-3)")
+@click.option("--frames", type=int, default=1, help="Number of 10ms frames")
+@click.option("--ssb/--no-ssb", default=True, help="Enable SSB")
+@common_options
+def nr5g(
+    sample_rate,
+    frequency_shift,
+    center_frequency,
+    channel_type,
+    noise_power,
+    path_gain,
+    output,
+    format,
+    overwrite,
+    metadata,
+    verbose,
+    quiet,
+    bandwidth,
+    mu,
+    frames,
+    ssb,
+    **kwargs,
+):
+    """Generate 5G NR frame."""
+
+    if not HAS_NR5G:
+        raise click.ClickException("5G NR Generator not available (missing dependencies or module)")
+
+    echo_progress(f"Generating 5G NR ({bandwidth} MHz, mu={mu}, {frames} frames)...", quiet)
+
+    # NR5GGenerator parameters
+    # It determines sample rate based on bandwidth/mu/fr?
+    # nr_ofdm_params(bandwidth_mhz, mu, fr) returns fs.
+    # We should verify if user supplied sample_rate matches or we should ignore user sample_rate?
+    # Or we resample?
+    # The generator has fixed fs for a given BW/mu config usually.
+    # Let's instantiate it and see its fs.
+
+    gen = NR5GGenerator(bandwidth_mhz=bandwidth, mu=mu, frames_per_recording=frames, ssb=ssb)
+
+    native_fs = gen.fs
+    if sample_rate and abs(sample_rate - native_fs) > 1.0:
+        echo_progress(
+            message=(
+                f"Warning: Requested sample rate {format_sample_rate(sample_rate)} "
+                f"differs from native NR rate {format_sample_rate(native_fs)}."
+            ),
+            quiet=quiet,
+        )
+        echo_progress("Output will be at native rate.", quiet)
+        # If we really wanted to support arbitrary rate, we'd need resampling.
+        # For now, just warn and use native.
+
+    recording = gen.record(batch_size=1)
+
+    recording._metadata["signal_type"] = "nr5g"
+
+    if center_frequency:
+        recording._metadata["center_frequency"] = center_frequency
+
+    # Post processing
+    chan_params = {"noise_power": noise_power, "path_gain": path_gain}
+    recording = apply_post_processing(recording, frequency_shift, channel_type, chan_params, verbose)
+
+    for key, value in apply_user_config_metadata(metadata).items():
+        recording.update_metadata(key, value)
+    fmt = get_output_format(output, format)
+    save_recording(recording, output, fmt, overwrite, verbose)

src/ria_toolkit_oss/ria_toolkit_oss_cli/ria_toolkit_oss/transform.py

@@ -492,6 +492,109 @@ def impair(impairment, input, output, list_transforms, help_transform, params, v
         quick_view_transform(result, output, title=f"{impairment.replace('_', ' ').title()} - {Path(output).name}")
 
 
+@transform.command(name="apply_channel")
+@click.argument("channel_model", required=False)
+@click.argument("input", type=click.Path(exists=True), required=False)
+@click.argument("output", type=click.Path(), required=False)
+@click.option("--list", "list_transforms", is_flag=True, help="List available channel models")
+@click.option("--help-transform", is_flag=True, help="Show parameters for this channel model")
+@click.option("--params", multiple=True, help="Transform parameters as KEY=VALUE (can be repeated)")
+@click.option("--view", is_flag=True, help="Save visualization PNG with constellation plot")
+@click.option("--overwrite", is_flag=True, help="Overwrite output if it exists")
+@click.option("--verbose", "-v", is_flag=True, help="Verbose output")
+@click.option("--quiet", "-q", is_flag=True, help="Suppress output")
+def apply_channel(
+    channel_model, input, output, list_transforms, help_transform, params, view, overwrite, verbose, quiet
+):
+    """Apply channel models to recordings.
+
+    Channel models simulate RF propagation effects like fading, Doppler shift,
+    and multipath reflections.
+
+    Use --list to see available channel models and their parameters.
+
+    \b
+    Examples:
+        ria_toolkit_oss transform apply_channel rayleigh_fading_channel input.npy --params num_paths=3 snr_db=15
+
+    \b
+        ria_toolkit_oss transform apply_channel doppler_channel recordings/input.npy \\
+        --params satellite_velocity=7500 \\
+        --params satellite_initial_distance=400000 \\
+        --params frequency=1e9 \\
+        --params sample_rate=2e6
+    """
+    available = get_available_transforms(iq_channel_models)
+
+    if list_transforms:
+        click.echo("Available channel models:")
+        for name in sorted(available.keys()):
+            func = available[name]
+            docstring = (func.__doc__ or "").split("\n")[0].strip()
+            click.echo(f"  {name:30} {docstring}")
+        return
+
+    if help_transform:
+        check_input_errors("channel_model", channel_model, available, input, help_transform)
+        show_transform_help(channel_model, available[channel_model])
+        return
+
+    check_input_errors("channel_model", channel_model, available, input, help_transform)
+
+    # Generate output filename if not provided
+    if output is None:
+        input_path = Path(input)
+        input_stem = input_path.stem
+        ext = input_path.suffix
+        suffix = generate_transform_suffix(channel_model, parse_transform_params(params))
+        output = str(input_path.parent / f"{input_stem}_{suffix}{ext}")
+        echo_verbose(f"Auto-generated output: {output}", verbose)
+
+    # Check if output exists
+    if not overwrite and Path(output).exists():
+        raise click.ClickException(f"Output file '{output}' already exists\n" f"Use --overwrite to replace")
+
+    echo_progress(f"Applying channel: {os.path.basename(input)} → {os.path.basename(output)}", quiet)
+    echo_verbose(f"Channel model: {channel_model}", verbose)
+
+    # Load input
+    recording = load_input(input, verbose)
+
+    # Parse and apply transform
+    try:
+        transform_func = available[channel_model]
+        transform_params = parse_transform_params(params)
+        echo_verbose(f"Parameters: {transform_params}", verbose)
+
+        result = transform_func(recording, **transform_params)
+    except Exception as e:
+        raise click.ClickException(f"Transform failed: {e}")
+
+    # Track transform in metadata (Recording.metadata is a property that returns a copy)
+    updated_metadata = result.metadata.copy()
+    if "transforms_applied" not in updated_metadata:
+        updated_metadata["transforms_applied"] = []
+
+    updated_metadata["transforms_applied"].append(
+        {"type": "channel", "name": channel_model, "params": parse_transform_params(params)}
+    )
+
+    # Create new recording with updated metadata
+    result = Recording(data=result.data, metadata=updated_metadata, annotations=result.annotations)
+
+    # Save output
+    try:
+        save_recording(result, output, overwrite=overwrite, verbose=verbose)
+        echo_progress(f"Saved to: {output}", quiet)
+    except Exception as e:
+        raise click.ClickException(f"Failed to save output: {e}")
+
+    # Optional: Create visualization
+    if view:
+        echo_verbose("Creating visualization...", verbose)
+        quick_view_transform(result, output, title=f"{channel_model.replace('_', ' ').title()} - {Path(output).name}")
+
+
 @transform.command(name="custom")
 @click.argument("transform_name", required=False)
 @click.argument("input", type=click.Path(exists=True), required=False)