diff --git a/src/SignalChannels.jl b/src/SignalChannels.jl index 136cfa8..4f241d0 100644 --- a/src/SignalChannels.jl +++ b/src/SignalChannels.jl @@ -18,6 +18,8 @@ export SignalChannel, read_from_file, spawn_signal_channel_thread, membuffer, + mux, + add, stream_data, SDRChannelConfig, PeriodogramData, @@ -30,6 +32,7 @@ export SignalChannel, include("signal_channel.jl") include("rechunk.jl") include("channel_utilities.jl") +include("channel_combine.jl") include("stream_utilities.jl") include("periodogram.jl") diff --git a/src/channel_combine.jl b/src/channel_combine.jl new file mode 100644 index 0000000..22023ec --- /dev/null +++ b/src/channel_combine.jl @@ -0,0 +1,184 @@ +using FixedSizeArrays: FixedSizeMatrixDefault + +""" + mux(ch1::SignalChannel{T,N}, ch2::SignalChannel{T,N}; + channel_size::Integer=10, sync::Bool=true) where {T,N} + +Sequential channel multiplexer that forwards all chunks from `ch1` until it closes, +then forwards all chunks from `ch2`. + +When `sync=true` (the default), while forwarding chunks from `ch1`, this function also +takes (and discards) chunks from `ch2` to keep both channels synchronized. This is useful +when both channels represent the same signal timeline and you need continuous phase when +switching from `ch1` to `ch2`. + +When `sync=false`, chunks from `ch2` are not consumed during the `ch1` phase. All chunks +from `ch2` are forwarded after `ch1` closes. + +# Arguments +- `ch1`: First input channel (forwarded until closed) +- `ch2`: Second input channel (forwarded after ch1 closes) +- `channel_size`: Output channel buffer size (default: 10) +- `sync`: If true, consume and discard ch2 chunks while forwarding ch1 (default: true) + +# Returns +- `SignalChannel{T,N}`: Output channel receiving chunks from ch1, then ch2 + +# Throws +- `ErrorException` if channels have different `num_samples` + +# Examples +```julia +ch1 = SignalChannel{ComplexF32,1}(1024, 10) +ch2 = SignalChannel{ComplexF32,1}(1024, 10) + +# With sync (default): ch2 is drained in parallel with ch1 +out = mux(ch1, ch2) + +# Without sync: ch2 is forwarded as-is after ch1 closes +out = mux(ch1, ch2; sync=false) +``` +""" +function mux( + ch1::SignalChannel{T,N}, + ch2::SignalChannel{T,N}; + channel_size::Integer = 10, + sync::Bool = true, +) where {T,N} + if ch1.num_samples != ch2.num_samples + error( + "mux requires both channels to have the same num_samples. Got $(ch1.num_samples) and $(ch2.num_samples)", + ) + end + num_samples = ch1.num_samples + out = SignalChannel{T,N}(num_samples, channel_size) + + # Pre-allocate buffer slots to avoid allocation in the hot loop. + # channel_size + 2 ensures we never overwrite a buffer still in flight + # in the output channel. + num_slots = channel_size + 2 + buffer_slots = [FixedSizeMatrixDefault{T}(undef, num_samples, N) for _ = 1:num_slots] + + task = + let ch1 = ch1, + ch2 = ch2, + out = out, + buffer_slots = buffer_slots, + num_slots = num_slots + + Threads.@spawn begin + slot_idx = 1 + + # Phase 1: Forward chunks from ch1, optionally consuming from ch2 + for chunk in ch1 + buffer = buffer_slots[slot_idx] + copyto!(buffer, chunk) + put!(out, buffer) + slot_idx = mod1(slot_idx + 1, num_slots) + if sync && (isopen(ch2) || isready(ch2)) + take!(ch2) + end + end + + # Phase 2: Forward all remaining chunks from ch2 + for chunk in ch2 + buffer = buffer_slots[slot_idx] + copyto!(buffer, chunk) + put!(out, buffer) + slot_idx = mod1(slot_idx + 1, num_slots) + end + + close(out) + end + end + + bind(out, task) + bind(ch1, task) + bind(ch2, task) + return out +end + +""" + add(channels::SignalChannel{T,N}...; channel_size::Integer=10) where {T,N} + +Element-wise addition of chunks from multiple SignalChannels. + +Takes chunks from all input channels simultaneously, adds them element-wise, +and puts the summed result into the output channel. Closes when the first input +channel closes. All inputs must produce data at the same rate. + +# Arguments +- `channels...`: Two or more input SignalChannels (all must have same T, N, and num_samples) +- `channel_size`: Output channel buffer size (default: 10) + +# Returns +- `SignalChannel{T,N}`: Output channel receiving the element-wise sum of input chunks + +# Throws +- `ArgumentError` if fewer than 2 channels are provided +- `ErrorException` if channels have different `num_samples` + +# Examples +```julia +ch1 = SignalChannel{ComplexF32,1}(1024, 10) +ch2 = SignalChannel{ComplexF32,1}(1024, 10) + +# Add two channels +out = add(ch1, ch2) + +# Add three or more channels +ch3 = SignalChannel{ComplexF32,1}(1024, 10) +out = add(ch1, ch2, ch3) +``` +""" +function add(channels::SignalChannel{T,N}...; channel_size::Integer = 10) where {T,N} + length(channels) >= 2 || + throw(ArgumentError("add requires at least 2 channels, got $(length(channels))")) + + num_samples = first(channels).num_samples + for (i, ch) in enumerate(channels) + if ch.num_samples != num_samples + error( + "add requires all channels to have the same num_samples. " * + "Channel 1 has $(num_samples), channel $i has $(ch.num_samples)", + ) + end + end + + out = SignalChannel{T,N}(num_samples, channel_size) + rest = channels[2:end] + + # Pre-allocate buffer slots to avoid allocation in the hot loop. + # channel_size + 2 ensures we never overwrite a buffer still in flight + # in the output channel. + num_slots = channel_size + 2 + buffer_slots = [FixedSizeMatrixDefault{T}(undef, num_samples, N) for _ = 1:num_slots] + + task = + let channels = channels, + rest = rest, + out = out, + buffer_slots = buffer_slots, + num_slots = num_slots + + Threads.@spawn begin + slot_idx = 1 + for chunk in first(channels) + result = buffer_slots[slot_idx] + copyto!(result, chunk) + for ch in rest + result .+= take!(ch) + end + put!(out, result) + slot_idx = mod1(slot_idx + 1, num_slots) + end + close(out) + end + end + + bind(out, task) + for ch in channels + bind(ch, task) + end + return out +end diff --git a/test/channel_combine.jl b/test/channel_combine.jl new file mode 100644 index 0000000..334be01 --- /dev/null +++ b/test/channel_combine.jl @@ -0,0 +1,386 @@ +module ChannelCombineTest + +using Test: @test, @testset, @test_throws +using SignalChannels: SignalChannel, mux, add +using FixedSizeArrays: FixedSizeMatrixDefault + +@testset "Channel Combine" begin + @testset "mux" begin + @testset "Basic sequential forwarding with sync" begin + chunk_size = 100 + ch1 = SignalChannel{ComplexF32,1}(chunk_size, 10) + ch2 = SignalChannel{ComplexF32,1}(chunk_size, 10) + + task1 = Threads.@spawn begin + for _ = 1:3 + chunk = FixedSizeMatrixDefault{ComplexF32}( + fill(ComplexF32(1.0), chunk_size, 1), + ) + put!(ch1, chunk) + end + close(ch1) + end + + task2 = Threads.@spawn begin + for i = 1:5 + chunk = FixedSizeMatrixDefault{ComplexF32}( + fill(ComplexF32(Float32(i)), chunk_size, 1), + ) + put!(ch2, chunk) + end + close(ch2) + end + + muxed = mux(ch1, ch2) + chunks = collect(muxed) + + @test length(chunks) == 5 + @test real(chunks[1][1]) == 1.0f0 + @test real(chunks[2][1]) == 1.0f0 + @test real(chunks[3][1]) == 1.0f0 + @test real(chunks[4][1]) == 4.0f0 + @test real(chunks[5][1]) == 5.0f0 + end + + @testset "Empty ch1 - immediately forwards ch2" begin + chunk_size = 100 + ch1 = SignalChannel{ComplexF32,1}(chunk_size, 10) + ch2 = SignalChannel{ComplexF32,1}(chunk_size, 10) + + close(ch1) + + task2 = Threads.@spawn begin + for _ = 1:3 + chunk = FixedSizeMatrixDefault{ComplexF32}( + fill(ComplexF32(2.0), chunk_size, 1), + ) + put!(ch2, chunk) + end + close(ch2) + end + + muxed = mux(ch1, ch2) + chunks = collect(muxed) + + @test length(chunks) == 3 + @test all(real(c[1]) == 2.0f0 for c in chunks) + end + + @testset "Empty ch2 - forwards ch1 then closes" begin + chunk_size = 100 + ch1 = SignalChannel{ComplexF32,1}(chunk_size, 10) + ch2 = SignalChannel{ComplexF32,1}(chunk_size, 10) + + task1 = Threads.@spawn begin + for _ = 1:3 + chunk = FixedSizeMatrixDefault{ComplexF32}( + fill(ComplexF32(1.0), chunk_size, 1), + ) + put!(ch1, chunk) + end + close(ch1) + end + + close(ch2) + + muxed = mux(ch1, ch2) + chunks = collect(muxed) + + @test length(chunks) == 3 + @test all(real(c[1]) == 1.0f0 for c in chunks) + end + + @testset "Order preserved" begin + chunk_size = 10 + ch1 = SignalChannel{ComplexF32,1}(chunk_size, 10) + ch2 = SignalChannel{ComplexF32,1}(chunk_size, 10) + + task1 = Threads.@spawn begin + for i = 1:3 + chunk = FixedSizeMatrixDefault{ComplexF32}( + fill(ComplexF32(Float32(i)), chunk_size, 1), + ) + put!(ch1, chunk) + end + close(ch1) + end + + task2 = Threads.@spawn begin + for i = 1:5 + chunk = FixedSizeMatrixDefault{ComplexF32}( + fill(ComplexF32(Float32(10 + i)), chunk_size, 1), + ) + put!(ch2, chunk) + end + close(ch2) + end + + muxed = mux(ch1, ch2) + chunks = collect(muxed) + + @test length(chunks) == 5 + @test real(chunks[1][1]) == 1.0f0 + @test real(chunks[2][1]) == 2.0f0 + @test real(chunks[3][1]) == 3.0f0 + @test real(chunks[4][1]) == 14.0f0 + @test real(chunks[5][1]) == 15.0f0 + end + + @testset "Works with Complex{Int16}" begin + chunk_size = 50 + ch1 = SignalChannel{Complex{Int16},1}(chunk_size, 10) + ch2 = SignalChannel{Complex{Int16},1}(chunk_size, 10) + + task1 = Threads.@spawn begin + chunk = FixedSizeMatrixDefault{Complex{Int16}}( + fill(Complex{Int16}(100, 0), chunk_size, 1), + ) + put!(ch1, chunk) + close(ch1) + end + + task2 = Threads.@spawn begin + for i = 1:2 + chunk = FixedSizeMatrixDefault{Complex{Int16}}( + fill(Complex{Int16}(100 * (i + 1), 0), chunk_size, 1), + ) + put!(ch2, chunk) + end + close(ch2) + end + + muxed = mux(ch1, ch2) + chunks = collect(muxed) + + @test length(chunks) == 2 + @test real(chunks[1][1]) == Int16(100) + @test real(chunks[2][1]) == Int16(300) + end + + @testset "Errors on mismatched num_samples" begin + ch1 = SignalChannel{ComplexF32,1}(100, 10) + ch2 = SignalChannel{ComplexF32,1}(200, 10) + + @test_throws ErrorException mux(ch1, ch2) + + close(ch1) + close(ch2) + end + + @testset "sync=false does not consume ch2 during ch1" begin + chunk_size = 100 + ch1 = SignalChannel{ComplexF32,1}(chunk_size, 10) + ch2 = SignalChannel{ComplexF32,1}(chunk_size, 10) + + task1 = Threads.@spawn begin + for _ = 1:3 + chunk = FixedSizeMatrixDefault{ComplexF32}( + fill(ComplexF32(1.0), chunk_size, 1), + ) + put!(ch1, chunk) + end + close(ch1) + end + + task2 = Threads.@spawn begin + for i = 1:5 + chunk = FixedSizeMatrixDefault{ComplexF32}( + fill(ComplexF32(Float32(i)), chunk_size, 1), + ) + put!(ch2, chunk) + end + close(ch2) + end + + muxed = mux(ch1, ch2; sync = false) + chunks = collect(muxed) + + # Without sync: 3 from ch1 + all 5 from ch2 = 8 total + @test length(chunks) == 8 + # First 3 from ch1 + @test all(real(chunks[i][1]) == 1.0f0 for i = 1:3) + # Then all 5 from ch2 (none were discarded) + @test real(chunks[4][1]) == 1.0f0 + @test real(chunks[5][1]) == 2.0f0 + @test real(chunks[6][1]) == 3.0f0 + @test real(chunks[7][1]) == 4.0f0 + @test real(chunks[8][1]) == 5.0f0 + end + end + + @testset "add" begin + @testset "Basic 2-channel addition" begin + chunk_size = 100 + ch1 = SignalChannel{ComplexF32,1}(chunk_size, 10) + ch2 = SignalChannel{ComplexF32,1}(chunk_size, 10) + + task1 = Threads.@spawn begin + for _ = 1:3 + put!( + ch1, + FixedSizeMatrixDefault{ComplexF32}( + fill(ComplexF32(1.0), chunk_size, 1), + ), + ) + end + close(ch1) + end + + task2 = Threads.@spawn begin + for _ = 1:3 + put!( + ch2, + FixedSizeMatrixDefault{ComplexF32}( + fill(ComplexF32(2.0), chunk_size, 1), + ), + ) + end + close(ch2) + end + + added = add(ch1, ch2) + chunks = collect(added) + + @test length(chunks) == 3 + @test all(real(c[1]) == 3.0f0 for c in chunks) + end + + @testset "3-channel addition" begin + chunk_size = 50 + ch1 = SignalChannel{ComplexF32,1}(chunk_size, 10) + ch2 = SignalChannel{ComplexF32,1}(chunk_size, 10) + ch3 = SignalChannel{ComplexF32,1}(chunk_size, 10) + + task1 = Threads.@spawn begin + for _ = 1:2 + put!( + ch1, + FixedSizeMatrixDefault{ComplexF32}( + fill(ComplexF32(1.0), chunk_size, 1), + ), + ) + end + close(ch1) + end + + task2 = Threads.@spawn begin + for _ = 1:2 + put!( + ch2, + FixedSizeMatrixDefault{ComplexF32}( + fill(ComplexF32(2.0), chunk_size, 1), + ), + ) + end + close(ch2) + end + + task3 = Threads.@spawn begin + for _ = 1:2 + put!( + ch3, + FixedSizeMatrixDefault{ComplexF32}( + fill(ComplexF32(3.0), chunk_size, 1), + ), + ) + end + close(ch3) + end + + added = add(ch1, ch2, ch3) + chunks = collect(added) + + @test length(chunks) == 2 + @test all(real(c[1]) == 6.0f0 for c in chunks) + end + + @testset "Multi-antenna channels (N=2)" begin + chunk_size = 50 + ch1 = SignalChannel{ComplexF32,2}(chunk_size, 10) + ch2 = SignalChannel{ComplexF32,2}(chunk_size, 10) + + task1 = Threads.@spawn begin + for _ = 1:2 + put!( + ch1, + FixedSizeMatrixDefault{ComplexF32}( + fill(ComplexF32(1.0, 0.0), chunk_size, 2), + ), + ) + end + close(ch1) + end + + task2 = Threads.@spawn begin + for _ = 1:2 + put!( + ch2, + FixedSizeMatrixDefault{ComplexF32}( + fill(ComplexF32(0.0, 1.0), chunk_size, 2), + ), + ) + end + close(ch2) + end + + added = add(ch1, ch2) + chunks = collect(added) + + @test length(chunks) == 2 + @test all(c[1, 1] == ComplexF32(1.0, 1.0) for c in chunks) + @test all(c[1, 2] == ComplexF32(1.0, 1.0) for c in chunks) + end + + @testset "Works with Complex{Int16}" begin + chunk_size = 50 + ch1 = SignalChannel{Complex{Int16},1}(chunk_size, 10) + ch2 = SignalChannel{Complex{Int16},1}(chunk_size, 10) + + task1 = Threads.@spawn begin + put!( + ch1, + FixedSizeMatrixDefault{Complex{Int16}}( + fill(Complex{Int16}(100, 0), chunk_size, 1), + ), + ) + close(ch1) + end + + task2 = Threads.@spawn begin + put!( + ch2, + FixedSizeMatrixDefault{Complex{Int16}}( + fill(Complex{Int16}(200, 50), chunk_size, 1), + ), + ) + close(ch2) + end + + added = add(ch1, ch2) + chunks = collect(added) + + @test length(chunks) == 1 + @test chunks[1][1] == Complex{Int16}(300, 50) + end + + @testset "Errors on mismatched num_samples" begin + ch1 = SignalChannel{ComplexF32,1}(100, 10) + ch2 = SignalChannel{ComplexF32,1}(200, 10) + + @test_throws ErrorException add(ch1, ch2) + + close(ch1) + close(ch2) + end + + @testset "Errors on fewer than 2 channels" begin + ch1 = SignalChannel{ComplexF32,1}(100, 10) + + @test_throws ArgumentError add(ch1) + + close(ch1) + end + end +end + +end # module ChannelCombineTest diff --git a/test/runtests.jl b/test/runtests.jl index cc6d804..9f7e149 100644 --- a/test/runtests.jl +++ b/test/runtests.jl @@ -1,6 +1,7 @@ include("signal_channel.jl") include("rechunk.jl") include("channel_utilities.jl") +include("channel_combine.jl") include("stream_utilities.jl") include("periodogram.jl") include("soapysdr_ext.jl")