Skip to main content

Signal processing execution graph using DAG and data classes

Project description

SigExec - Signal Processing Chain Framework

A Python framework for building signal processing graphs with functional dataclass blocks and fluent APIs.

SigExec provides the framework - you bring the blocks! The included radar processing blocks are examples showing how to use the framework. You can easily create your own custom blocks for any signal processing application.

Quick Links

Features

  • Clean Graph Architecture: Build graphs where a single object (SignalData) flows through processing stages
  • Data Class Blocks: Type-safe, composable processing blocks using Python dataclasses
  • Extensible: Create custom blocks as simple dataclasses - no complex interfaces required
  • Functional Composition: Chain operations naturally with consistent input/output types
  • Flexible API: Multiple usage patterns from explicit chaining to graph builders
  • Example Application: Complete radar processing graph demonstrating:
    • LFM signal generation with delay and Doppler shift
    • Pulse stacking
    • Matched filtering (range compression)
    • FFT processing (Doppler compression)
    • Range-Doppler map visualization

Installation

From Source

git clone https://github.com/briday1/sigexec.git
cd sigexec
pip install -e .

Requirements

  • Python >= 3.7
  • numpy >= 1.20.0
  • scipy >= 1.7.0
  • matplotlib >= 3.3.0

Quick Start

Simplest Example - Direct Chaining

The cleanest approach where each block is a configured data class:

from sigexec.blocks import LFMGenerator, StackPulses, RangeCompress, DopplerCompress

# Configure blocks
gen = LFMGenerator(num_pulses=128, target_delay=20e-6, target_doppler=1000.0)
stack = StackPulses()
range_comp = RangeCompress()
doppler_comp = DopplerCompress(window='hann')

# Single SignalData object flows through graph
signal = gen()                    # Generate signal
signal = stack(signal)            # Stack pulses
signal = range_comp(signal)       # Range compression
signal = doppler_comp(signal)     # Doppler compression

# Result is a range-doppler map!
range_doppler_map = signal.data

Using Graph for Better Organization

from sigexec import Graph
from sigexec.blocks import LFMGenerator, StackPulses, RangeCompress, DopplerCompress

# Build graph with fluent interface
result = (Graph("Radar")
    .add(LFMGenerator(num_pulses=128, target_delay=20e-6, target_doppler=1000.0))
    .add(StackPulses())
    .add(RangeCompress())
    .add(DopplerCompress(window='hann'))
    .run(verbose=True)
)

# Access the range-doppler map
rdm = result.data

Running Examples

# Publish all demos to docs/ (for GitHub Pages)
python examples/publish_demos.py

# Or run individual demos (publishes to staticdash/)
python examples/radar_processing_demo.py
python examples/custom_blocks_demo.py
python examples/parameter_exploration_demo.py
python examples/post_processing_demo.py
python examples/input_variants_demo.py

Architecture

Core Components

SignalData

A data class that wraps signal arrays with metadata:

@dataclass
class SignalData:
    data: np.ndarray          # Signal data
    sample_rate: float        # Sampling rate
    metadata: Dict[str, Any]  # Additional information

Key Point: Every processing block takes SignalData as input and returns SignalData as output, enabling clean composition.

Data Class Blocks (Recommended)

Modern, clean blocks implemented as dataclasses:

from sigexec.blocks import LFMGenerator, StackPulses, RangeCompress, DopplerCompress

# Configure blocks with parameters
gen = LFMGenerator(num_pulses=128, target_delay=20e-6)
stack = StackPulses()
compress = RangeCompress()

# Call them directly - each returns SignalData
signal = gen()
signal = stack(signal)
signal = compress(signal)

Available data class blocks:

  • LFMGenerator - Generate LFM radar signals
  • StackPulses - Organize pulses into 2D matrix
  • RangeCompress - Matched filtering for range compression
  • DopplerCompress - FFT-based Doppler processing
  • ToMagnitudeDB - Convert to dB scale
  • Normalize - Normalize signal data

Graph

Manages execution with fluent interface:

graph = (Graph("MyPipeline")
    .add(block1)
    .add(block2)
    .add(block3)
    .run()
)

Processing Blocks

All blocks follow the pattern: SignalData → Block → SignalData

LFMGenerator

Generates LFM radar signals with configurable parameters:

  • Pulse duration and bandwidth
  • Target delay and Doppler shift
  • Noise characteristics
gen = LFMGenerator(
    num_pulses=128,
    pulse_duration=10e-6,
    bandwidth=5e6,
    target_delay=20e-6,
    target_doppler=1000.0
)
signal = gen()  # Returns SignalData

StackPulses

Organizes pulses into a 2D matrix for coherent processing.

RangeCompress

Performs range compression using matched filtering:

  • Correlates received signal with transmitted waveform
  • Improves SNR and range resolution

DopplerCompress

Performs Doppler compression using FFT:

  • FFT along pulse dimension
  • Windowing for sidelobe reduction
  • Generates Range-Doppler map

Example Output

The radar examples produce Range-Doppler maps showing:

  • 2D visualization: Range vs Doppler frequency with intensity showing target returns
  • Target detection: Clear peak at expected range (~3 km) and Doppler (~1 kHz)
  • Noise floor: Background noise visible across the map

Project Structure

sigexec/
├── sigexec/
│   ├── __init__.py
│   ├── core/
│   │   ├── __init__.py
│   │   ├── data.py          # SignalData class
│   │   └── graph.py      # Graph with fluent interface
│   └── blocks/
│       ├── __init__.py
│       └── functional.py    # Functional processing blocks
├── examples/
│   ├── radar_processing_demo.py
│   ├── custom_blocks_demo.py
│   ├── parameter_exploration_demo.py
│   ├── post_processing_demo.py
│   ├── input_variants_demo.py
│   ├── memoization_demo.py
│   └── publish_demos.py
├── tests/
│   └── test_sigexec.py
├── docs/
│   └── [Generated demo pages]
├── pyproject.toml
└── README.md

Usage Patterns

Pattern 1: Direct Chaining (Cleanest)

# Configure data class blocks
gen = LFMGenerator(num_pulses=128, target_delay=20e-6)
stack = StackPulses()
compress_range = RangeCompress()
compress_doppler = DopplerCompress()

# Single object flows through
signal = gen()
signal = stack(signal)
signal = compress_range(signal)
signal = compress_doppler(signal)

Pattern 2: Graph Builder

result = (Graph("Radar")
    .add(LFMGenerator(num_pulses=128))
    .add(StackPulses())
    .add(RangeCompress())
    .add(DopplerCompress())
    .tap(lambda sig: print(f"Shape: {sig.shape}"))  # Inspect
    .run(verbose=True)
)

Pattern 3: Functional Composition

# Compose operations functionally
process = lambda sig: DopplerCompress()(RangeCompress()(StackPulses()(sig)))
result = process(LFMGenerator()())

Creating Custom Blocks

SigExec is designed to be extended! The included radar blocks are examples - create your own blocks for any domain:

from dataclasses import dataclass
from sigexec import SignalData

@dataclass
class MyCustomBlock:
    """My custom processing block."""
    
    param1: float = 1.0
    param2: str = 'default'
    
    def __call__(self, signal_data: SignalData) -> SignalData:
        """Process the signal."""
        processed_data = your_algorithm(signal_data.data, self.param1)
        
        metadata = signal_data.metadata.copy()
        metadata['my_processing'] = True
        
        return SignalData(
            data=processed_data,
            sample_rate=signal_data.sample_rate,
            metadata=metadata
        )

# Use it with built-in or other custom blocks
my_block = MyCustomBlock(param1=2.5)
result = my_block(input_signal)

Distributing Custom Blocks

You can create and distribute your own block packages:

# Your package: my_signal_blocks
from sigexec import Graph
from my_signal_blocks import CustomFilter, CustomTransform

result = (Graph("MyPipeline")
    .add(CustomFilter(cutoff=1000))
    .add(CustomTransform(mode='advanced'))
    .run()
)

Learn more:

Documentation

  • CUSTOM_BLOCKS.md - Guide to creating and distributing custom blocks
  • examples/ - Working examples with different patterns
  • tests/ - Unit tests for all components

Design Philosophy

  1. Framework First: SigExec provides the framework; you provide the blocks
  2. Type Safety: Same type (SignalData) throughout the graph
  3. Composability: Blocks can be combined in any order
  4. Extensibility: Easy to create and distribute custom blocks
  5. Clarity: Configuration separate from execution
  6. Immutability: Each block returns new data
  7. Simplicity: Minimal API surface, maximum flexibility

Extensibility

The radar processing blocks included in sigexec.blocks are examples demonstrating the framework. The framework is designed to support:

  • Any signal processing domain: Audio, video, communications, radar, medical imaging, etc.
  • Custom block packages: Distribute your blocks as separate Python packages
  • Third-party blocks: Use blocks from other packages with full framework integration
  • Domain-specific graphs: Build specialized processing chains for your application

See CUSTOM_BLOCKS.md for a complete guide on creating and distributing custom blocks.

Contributing

Contributions are welcome! Please feel free to submit a Pull Request.

License

This project is open source and available under the MIT License.

Acknowledgments

This framework demonstrates fundamental radar signal processing concepts and serves as a foundation for building more complex signal processing graphs.

Project details


Download files

Download the file for your platform. If you're not sure which to choose, learn more about installing packages.

Source Distribution

sigexec-2026.2.tar.gz (35.4 kB view details)

Uploaded Source

Built Distribution

If you're not sure about the file name format, learn more about wheel file names.

sigexec-2026.2-py3-none-any.whl (24.3 kB view details)

Uploaded Python 3

File details

Details for the file sigexec-2026.2.tar.gz.

File metadata

  • Download URL: sigexec-2026.2.tar.gz
  • Upload date:
  • Size: 35.4 kB
  • Tags: Source
  • Uploaded using Trusted Publishing? No
  • Uploaded via: twine/6.2.0 CPython/3.11.14

File hashes

Hashes for sigexec-2026.2.tar.gz
Algorithm Hash digest
SHA256 53b8528ce0670966889e9e375c3c00fc286fb563e5fa6afd651ac9f9f26d9b34
MD5 16cdb4ffe67feceb2ee7bfc96281e37e
BLAKE2b-256 e2668533c96956fbb4eca605d67a0f06db0803dc85506e37899fd4e61f30f10b

See more details on using hashes here.

File details

Details for the file sigexec-2026.2-py3-none-any.whl.

File metadata

  • Download URL: sigexec-2026.2-py3-none-any.whl
  • Upload date:
  • Size: 24.3 kB
  • Tags: Python 3
  • Uploaded using Trusted Publishing? No
  • Uploaded via: twine/6.2.0 CPython/3.11.14

File hashes

Hashes for sigexec-2026.2-py3-none-any.whl
Algorithm Hash digest
SHA256 5f92520d20096649b5e0a853aa11880dc2b71755025938f0463c30de24981090
MD5 65edb57ac25e7c9df4c5ad5a8e08da73
BLAKE2b-256 dd6f90600872047e71d5d3f5df196ebb14b712972e2831456bb2a84b1ecb8f8d

See more details on using hashes here.

Supported by

AWS Cloud computing and Security Sponsor Datadog Monitoring Depot Continuous Integration Fastly CDN Google Download Analytics Pingdom Monitoring Sentry Error logging StatusPage Status page