From eccbe9f187489bbb637132b5e21883d0a9e6f686 Mon Sep 17 00:00:00 2001 From: madrigal Date: Fri, 12 Dec 2025 16:46:21 -0500 Subject: [PATCH] Removed unincluded channel models from use in cli --- .../ria_toolkit_oss/generate.py | 306 +++--------------- .../ria_toolkit_oss/transform.py | 103 ------ 2 files changed, 38 insertions(+), 371 deletions(-) diff --git a/src/ria_toolkit_oss/ria_toolkit_oss_cli/ria_toolkit_oss/generate.py b/src/ria_toolkit_oss/ria_toolkit_oss_cli/ria_toolkit_oss/generate.py index b4d183b..716332f 100644 --- a/src/ria_toolkit_oss/ria_toolkit_oss_cli/ria_toolkit_oss/generate.py +++ b/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, channel_type: str, channel_params: dict, verbose: bool + recording: Recording, frequency_shift: float, add_noise: str, channel_params: dict, verbose: bool ) -> Recording: """Apply frequency shift and channel models to a recording.""" @@ -200,57 +200,7 @@ def apply_post_processing( processed = fs_block.get_samples(num) recording = Recording(data=processed, metadata=recording.metadata) - # 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 + # 2. IQ Imbalance amp = channel_params.get("iq_amp_imbalance") phase = channel_params.get("iq_phase_imbalance") dc = channel_params.get("iq_dc_offset") @@ -265,61 +215,15 @@ def apply_post_processing( dc_offset=dc if dc is not None else 0, ) - # Phase Noise - pn = channel_params.get("phase_noise") - if pn: - echo_verbose(f"Applying Phase Noise (Var={pn})", verbose) - recording = add_phase_noise(recording, phase_variance=pn) - - # Gain Fluctuation - gf = channel_params.get("gain_fluctuation") - if gf: - 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? + if add_noise == "awgn": + npow = channel_params.get("noise_power", 0.1) + echo_verbose(f"Applying AWGN (Power={npow})", verbose) - 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) + # 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 @@ -346,25 +250,18 @@ def common_options(f): f ) f = click.option("--center-frequency", "-fc", type=float, help="Metadata center frequency (Hz)")(f) - f = click.option( - "--channel-type", type=click.Choice(["none", "awgn", "rayleigh"]), default="none", help="Channel model" - )(f) + f = click.option("--add-noise", is_flag=True, help="Add noise to signal")(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", @@ -413,7 +310,7 @@ def tone( duration, frequency_shift, center_frequency, - channel_type, + add_noise, noise_power, path_gain, output, @@ -444,7 +341,7 @@ def tone( # Post processing chan_params = {"noise_power": noise_power, "path_gain": path_gain} - recording = apply_post_processing(recording, frequency_shift, channel_type, chan_params, verbose) + recording = apply_post_processing(recording, frequency_shift, add_noise, chan_params, verbose) # User metadata metadata = apply_user_config_metadata(metadata) @@ -466,7 +363,7 @@ def noise( duration, frequency_shift, center_frequency, - channel_type, + add_noise, noise_power, path_gain, output, @@ -504,7 +401,7 @@ def noise( # Post processing chan_params = {"noise_power": noise_power, "path_gain": path_gain} - recording = apply_post_processing(recording, frequency_shift, channel_type, chan_params, verbose) + recording = apply_post_processing(recording, frequency_shift, add_noise, chan_params, verbose) for key, value in apply_user_config_metadata(metadata).items(): recording.update_metadata(key, value) @@ -523,7 +420,7 @@ def chirp( duration, frequency_shift, center_frequency, - channel_type, + add_noise, noise_power, path_gain, output, @@ -557,7 +454,7 @@ def chirp( # Post processing chan_params = {"noise_power": noise_power, "path_gain": path_gain} - recording = apply_post_processing(recording, frequency_shift, channel_type, chan_params, verbose) + recording = apply_post_processing(recording, frequency_shift, add_noise, chan_params, verbose) for key, value in apply_user_config_metadata(metadata).items(): recording.update_metadata(key, value) @@ -577,7 +474,7 @@ def square( duration, frequency_shift, center_frequency, - channel_type, + add_noise, noise_power, path_gain, output, @@ -609,7 +506,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, channel_type, chan_params, verbose) + recording = apply_post_processing(recording, frequency_shift, add_noise, chan_params, verbose) for key, value in apply_user_config_metadata(metadata).items(): recording.update_metadata(key, value) @@ -628,7 +525,7 @@ def sawtooth( duration, frequency_shift, center_frequency, - channel_type, + add_noise, noise_power, path_gain, output, @@ -657,7 +554,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, channel_type, chan_params, verbose) + recording = apply_post_processing(recording, frequency_shift, add_noise, chan_params, verbose) for key, value in apply_user_config_metadata(metadata).items(): recording.update_metadata(key, value) @@ -718,7 +615,7 @@ def _run_mod_gen( message_content, frequency_shift, center_frequency, - channel_type, + add_noise, noise_power, path_gain, output, @@ -802,7 +699,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, channel_type, chan_params, verbose) + final_recording = apply_post_processing(base_recording, frequency_shift, add_noise, 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) @@ -842,7 +739,7 @@ def qam( duration, frequency_shift, center_frequency, - channel_type, + add_noise, noise_power, path_gain, output, @@ -851,16 +748,11 @@ 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, @@ -894,7 +786,7 @@ def qam( message_content, frequency_shift, center_frequency, - channel_type, + add_noise, noise_power, path_gain, output, @@ -930,7 +822,7 @@ def apsk( duration, frequency_shift, center_frequency, - channel_type, + add_noise, noise_power, path_gain, output, @@ -939,16 +831,11 @@ 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, @@ -975,7 +862,7 @@ def apsk( message_content, frequency_shift, center_frequency, - channel_type, + add_noise, noise_power, path_gain, output, @@ -1011,7 +898,7 @@ def pam( duration, frequency_shift, center_frequency, - channel_type, + add_noise, noise_power, path_gain, output, @@ -1020,16 +907,11 @@ 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, @@ -1056,7 +938,7 @@ def pam( message_content, frequency_shift, center_frequency, - channel_type, + add_noise, noise_power, path_gain, output, @@ -1085,7 +967,7 @@ def fsk( duration, frequency_shift, center_frequency, - channel_type, + add_noise, noise_power, path_gain, output, @@ -1094,16 +976,11 @@ 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, @@ -1172,19 +1049,14 @@ 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, channel_type, chan_params, verbose) + recording = apply_post_processing(recording, frequency_shift, add_noise, chan_params, verbose) for key, value in apply_user_config_metadata(metadata).items(): recording.update_metadata(key, value) @@ -1205,7 +1077,7 @@ def ook( duration, frequency_shift, center_frequency, - channel_type, + add_noise, noise_power, path_gain, output, @@ -1214,16 +1086,11 @@ 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, @@ -1251,19 +1118,14 @@ 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, channel_type, chan_params, verbose) + recording = apply_post_processing(recording, frequency_shift, add_noise, chan_params, verbose) for key, value in apply_user_config_metadata(metadata).items(): recording.update_metadata(key, value) @@ -1284,7 +1146,7 @@ def oqpsk( duration, frequency_shift, center_frequency, - channel_type, + add_noise, noise_power, path_gain, output, @@ -1293,16 +1155,11 @@ 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, @@ -1330,19 +1187,14 @@ 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, channel_type, chan_params, verbose) + recording = apply_post_processing(recording, frequency_shift, add_noise, chan_params, verbose) for key, value in apply_user_config_metadata(metadata).items(): recording.update_metadata(key, value) @@ -1364,7 +1216,7 @@ def gmsk( duration, frequency_shift, center_frequency, - channel_type, + add_noise, noise_power, path_gain, output, @@ -1373,16 +1225,11 @@ 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, @@ -1412,19 +1259,14 @@ 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, channel_type, chan_params, verbose) + recording = apply_post_processing(recording, frequency_shift, add_noise, chan_params, verbose) for key, value in apply_user_config_metadata(metadata).items(): recording.update_metadata(key, value) @@ -1456,7 +1298,7 @@ def psk( duration, frequency_shift, center_frequency, - channel_type, + add_noise, noise_power, path_gain, output, @@ -1491,7 +1333,7 @@ def psk( message_content, frequency_shift, center_frequency, - channel_type, + add_noise, noise_power, path_gain, output, @@ -1501,75 +1343,3 @@ 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) diff --git a/src/ria_toolkit_oss/ria_toolkit_oss_cli/ria_toolkit_oss/transform.py b/src/ria_toolkit_oss/ria_toolkit_oss_cli/ria_toolkit_oss/transform.py index 361d67c..9cf73d5 100644 --- a/src/ria_toolkit_oss/ria_toolkit_oss_cli/ria_toolkit_oss/transform.py +++ b/src/ria_toolkit_oss/ria_toolkit_oss_cli/ria_toolkit_oss/transform.py @@ -492,109 +492,6 @@ 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)