serial::Hdf5OutputSink::Impl Struct Reference

Classes
struct	StreamState

Public Member Functions
	Impl (std::string out_dir, std::shared_ptr< core::OutputMetadataCollector > collector)
void	initializeRun (const core::OutputConfig &config, const std::string &)
std::uint32_t	registerStream (const core::ReceiverStreamDescriptor &stream)
void	openStream (const std::uint32_t stream_id, const RealType)
void	submitBlock (const core::ReceiverSampleBlock &block)
void	closeStream (const std::uint32_t stream_id)
core::OutputStats	finalize ()
std::string	outputPath (const std::string &receiver_name) const
StreamState &	stateFor (const std::uint32_t stream_id)
void	writePulsedBlock (StreamState &state, const core::ReceiverSampleBlock &block)
void	appendStreamingBlock (StreamState &state, const core::ReceiverSampleBlock &block)
void	closePulsedStream (StreamState &state)
void	closeStreamingStream (StreamState &state)

Static Public Member Functions
static bool	isPulsed (const StreamState &state)
static std::size_t	expectedStreamingSamples (const RealType sample_rate)
static std::string	streamKey (const core::ReceiverStreamDescriptor &stream)

Public Attributes
std::string	output_dir
std::shared_ptr< core::OutputMetadataCollector >	metadata_collector
std::recursive_mutex	mutex
std::unordered_map< std::uint32_t, StreamState >	streams
std::unordered_map< std::string, std::uint32_t >	stream_ids
std::uint32_t	next_stream_id = 1
bool	finalized = false

Detailed Description

Definition at line 60 of file hdf5_output_sink.cpp.

Constructor & Destructor Documentation

Impl()

serial::Hdf5OutputSink::Impl::Impl	(	std::string	out_dir,
		std::shared_ptr< core::OutputMetadataCollector >	collector
	)

Definition at line 74 of file hdf5_output_sink.cpp.

                                                                                      :
            output_dir(std::move(out_dir)), metadata_collector(std::move(collector))
        {
        }

Member Function Documentation

appendStreamingBlock()

void serial::Hdf5OutputSink::Impl::appendStreamingBlock	(	StreamState &	state,
		const core::ReceiverSampleBlock &	block
	)

Definition at line 260 of file hdf5_output_sink.cpp.

        {
            const auto sample_start = static_cast<std::size_t>(block.sample_start);
            const auto sample_end = sample_start + block.samples.size();
            if (state.streaming_buffer.size() < sample_end)
            {
                state.streaming_buffer.resize(sample_end);
            }
            std::ranges::copy(block.samples,
                              std::next(state.streaming_buffer.begin(),
                                        static_cast<std::vector<ComplexType>::difference_type>(sample_start)));
        }

References max.

Referenced by submitBlock().

Here is the caller graph for this function:

closePulsedStream()

void serial::Hdf5OutputSink::Impl::closePulsedStream

(

StreamState &

state

)

Definition at line 273 of file hdf5_output_sink.cpp.

        {
            if (!state.pulsed_file)
            {
                return;
            }
            finalizePulsedMetadata(state.metadata);
            {
                std::scoped_lock const hdf5_lock(hdf5_global_mutex);
                writeOutputFileMetadataAttributes(*state.pulsed_file, state.metadata);
                state.pulsed_file.reset();
            }
            if (metadata_collector)
            {
                metadata_collector->addFile(state.metadata);
            }
        }

References serial::hdf5_global_mutex, max, metadata_collector, and serial::writeOutputFileMetadataAttributes().

Referenced by closeStream().

Here is the call graph for this function:

Here is the caller graph for this function:

closeStream()

void serial::Hdf5OutputSink::Impl::closeStream

(

const std::uint32_t

stream_id

)

Definition at line 160 of file hdf5_output_sink.cpp.

        {
            std::scoped_lock const lock(mutex);
            auto& state = stateFor(stream_id);
            if (state.closed)
            {
                return;
            }
 
            if (isPulsed(state))
            {
                closePulsedStream(state);
            }
            else
            {
                closeStreamingStream(state);
            }
            state.closed = true;
        }

References closePulsedStream(), closeStreamingStream(), isPulsed(), max, mutex, and stateFor().

Referenced by finalize().

Here is the call graph for this function:

Here is the caller graph for this function:

closeStreamingStream()

void serial::Hdf5OutputSink::Impl::closeStreamingStream

(

StreamState &

state

)

Definition at line 291 of file hdf5_output_sink.cpp.

        {
            if (state.streaming_buffer.empty())
            {
                return;
            }
 
            auto metadata = state.streaming_metadata.value_or(
                core::OutputFileMetadata{.receiver_id = state.descriptor.receiver_id,
                                         .receiver_name = state.descriptor.receiver_name,
                                         .mode = state.descriptor.mode,
                                         .path = state.metadata.path,
                                         .sampling_rate = state.descriptor.sample_rate});
            metadata.path = state.metadata.path;
            metadata.sampling_rate = state.descriptor.sample_rate;
            metadata.total_samples = static_cast<std::uint64_t>(state.streaming_buffer.size());
            metadata.sample_start = 0;
            metadata.sample_end_exclusive = static_cast<std::uint64_t>(state.streaming_buffer.size());
 
            const RealType fullscale = processing::quantizeAndScaleWindow(state.streaming_buffer);
            processing::pipeline::exportStreamingToHdf5(state.metadata.path, state.streaming_buffer, fullscale,
                                                        state.descriptor.reference_frequency, &metadata,
                                                        state.descriptor.sample_rate);
            if (metadata_collector)
            {
                metadata_collector->addFile(std::move(metadata));
            }
        }

References processing::pipeline::exportStreamingToHdf5(), max, metadata_collector, core::OutputFileMetadata::path, processing::quantizeAndScaleWindow(), core::OutputFileMetadata::receiver_id, core::OutputFileMetadata::sample_end_exclusive, core::OutputFileMetadata::sample_start, core::OutputFileMetadata::sampling_rate, and core::OutputFileMetadata::total_samples.

Referenced by closeStream().

Here is the call graph for this function:

Here is the caller graph for this function:

expectedStreamingSamples()

static std::size_t serial::Hdf5OutputSink::Impl::expectedStreamingSamples ( const RealType  sample_rate )

static

Definition at line 205 of file hdf5_output_sink.cpp.

        {
            if (sample_rate <= 0.0)
            {
                return 0u;
            }
            const RealType duration = std::max<RealType>(0.0, params::endTime() - params::startTime());
            return static_cast<std::size_t>(std::ceil(duration * sample_rate));
        }

References params::endTime(), max, and params::startTime().

Referenced by openStream().

Here is the call graph for this function:

Here is the caller graph for this function:

finalize()

core::OutputStats serial::Hdf5OutputSink::Impl::finalize

(

)

Definition at line 180 of file hdf5_output_sink.cpp.

        {
            std::scoped_lock const lock(mutex);
            if (!finalized)
            {
                std::vector<std::uint32_t> ids;
                ids.reserve(streams.size());
                for (const auto& [stream_id, state] : streams)
                {
                    if (state.opened && !state.closed)
                    {
                        ids.push_back(stream_id);
                    }
                }
                for (const auto stream_id : ids)
                {
                    closeStream(stream_id);
                }
                finalized = true;
            }
            return core::OutputStats{.mode = core::OutputMode::Hdf5, .streams = {}};
        }

References closeStream(), finalized, core::Hdf5, max, core::OutputStats::mode, mutex, and streams.

Here is the call graph for this function:

initializeRun()

void serial::Hdf5OutputSink::Impl::initializeRun	(	const core::OutputConfig &	config,
		const std::string &
	)

Definition at line 79 of file hdf5_output_sink.cpp.

        {
            if (config.mode != core::OutputMode::Hdf5)
            {
                throw std::invalid_argument("Hdf5OutputSink requires HDF5 output mode");
            }
            std::scoped_lock const lock(mutex);
            std::filesystem::create_directories(output_dir);
            finalized = false;
        }

References finalized, core::Hdf5, max, mutex, and output_dir.

isPulsed()

static bool serial::Hdf5OutputSink::Impl::isPulsed ( const StreamState &  state )

static

Definition at line 203 of file hdf5_output_sink.cpp.

{ return state.descriptor.mode == "pulsed"; }

References max.

Referenced by closeStream(), openStream(), and submitBlock().

Here is the caller graph for this function:

openStream()

void serial::Hdf5OutputSink::Impl::openStream	(	const std::uint32_t	stream_id,
		const RealType
	)

Definition at line 112 of file hdf5_output_sink.cpp.

        {
            std::scoped_lock const lock(mutex);
            auto& state = stateFor(stream_id);
            if (state.opened)
            {
                return;
            }
            std::filesystem::create_directories(output_dir);
            if (isPulsed(state))
            {
                std::scoped_lock const hdf5_lock(hdf5_global_mutex);
                state.pulsed_file = std::make_unique<HighFive::File>(state.metadata.path, HighFive::File::Truncate);
            }
            else
            {
                const auto expected_samples = expectedStreamingSamples(state.descriptor.sample_rate);
                if (expected_samples > 0u)
                {
                    state.streaming_buffer.resize(expected_samples);
                }
            }
            state.opened = true;
            state.closed = false;
        }

References expectedStreamingSamples(), serial::hdf5_global_mutex, isPulsed(), max, mutex, output_dir, and stateFor().

Referenced by submitBlock().

Here is the call graph for this function:

Here is the caller graph for this function:

outputPath()

std::string serial::Hdf5OutputSink::Impl::outputPath

(

const std::string &

receiver_name

)

const

Definition at line 220 of file hdf5_output_sink.cpp.

        {
            const std::filesystem::path out_path(output_dir);
            return (out_path / std::format("{}_results.h5", receiver_name)).string();
        }

References max, and output_dir.

Referenced by registerStream().

Here is the caller graph for this function:

registerStream()

std::uint32_t serial::Hdf5OutputSink::Impl::registerStream

(

const core::ReceiverStreamDescriptor &

stream

)

Definition at line 90 of file hdf5_output_sink.cpp.

        {
            std::scoped_lock const lock(mutex);
            const auto key = streamKey(stream);
            if (const auto found = stream_ids.find(key); found != stream_ids.end())
            {
                return found->second;
            }
 
            const auto stream_id = next_stream_id++;
            StreamState state;
            state.descriptor = stream;
            state.metadata = core::OutputFileMetadata{.receiver_id = stream.receiver_id,
                                                      .receiver_name = stream.receiver_name,
                                                      .mode = stream.mode,
                                                      .path = outputPath(stream.receiver_name),
                                                      .sampling_rate = stream.sample_rate};
            stream_ids.emplace(key, stream_id);
            streams.emplace(stream_id, std::move(state));
            return stream_id;
        }

References max, core::ReceiverStreamDescriptor::mode, mutex, next_stream_id, outputPath(), core::OutputFileMetadata::receiver_id, core::ReceiverStreamDescriptor::receiver_id, core::ReceiverStreamDescriptor::receiver_name, core::ReceiverStreamDescriptor::sample_rate, stream_ids, streamKey(), and streams.

Referenced by submitBlock().

Here is the call graph for this function:

Here is the caller graph for this function:

stateFor()

StreamState & serial::Hdf5OutputSink::Impl::stateFor

(

const std::uint32_t

stream_id

)

Definition at line 226 of file hdf5_output_sink.cpp.

        {
            const auto found = streams.find(stream_id);
            if (found == streams.end())
            {
                throw std::out_of_range("Unknown HDF5 output stream ID");
            }
            return found->second;
        }

References max, and streams.

Referenced by closeStream(), openStream(), and submitBlock().

Here is the caller graph for this function:

streamKey()

static std::string serial::Hdf5OutputSink::Impl::streamKey ( const core::ReceiverStreamDescriptor &  stream )

static

Definition at line 215 of file hdf5_output_sink.cpp.

        {
            return std::format("{}:{}:{}", stream.receiver_id, stream.receiver_name, stream.mode);
        }

References core::ReceiverStreamDescriptor::mode, core::ReceiverStreamDescriptor::receiver_id, and core::ReceiverStreamDescriptor::receiver_name.

Referenced by registerStream().

Here is the caller graph for this function:

submitBlock()

void serial::Hdf5OutputSink::Impl::submitBlock

(

const core::ReceiverSampleBlock &

block

)

Definition at line 138 of file hdf5_output_sink.cpp.

        {
            std::scoped_lock const lock(mutex);
            const auto stream_id = registerStream(block.stream);
            auto& state = stateFor(stream_id);
            if (!state.opened)
            {
                openStream(stream_id, block.first_sample_time);
            }
            if (block.file_metadata)
            {
                state.streaming_metadata = *block.file_metadata;
            }
 
            if (isPulsed(state))
            {
                writePulsedBlock(state, block);
                return;
            }
            appendStreamingBlock(state, block);
        }

References appendStreamingBlock(), isPulsed(), max, mutex, openStream(), registerStream(), stateFor(), and writePulsedBlock().

Here is the call graph for this function:

writePulsedBlock()

void serial::Hdf5OutputSink::Impl::writePulsedBlock	(	StreamState &	state,
		const core::ReceiverSampleBlock &	block
	)

Definition at line 236 of file hdf5_output_sink.cpp.

        {
            if (!state.pulsed_file)
            {
                throw std::logic_error("HDF5 pulsed stream is not open");
            }
 
            std::vector<ComplexType> chunk(block.samples.begin(), block.samples.end());
            const RealType fullscale = processing::quantizeAndScaleWindow(chunk);
            const auto chunk_index = state.next_chunk_index++;
            const auto sample_start = state.metadata.total_samples;
            core::PulseChunkMetadata chunk_metadata{.chunk_index = chunk_index,
                                                    .i_dataset = std::format("chunk_{:06}_I", chunk_index),
                                                    .q_dataset = std::format("chunk_{:06}_Q", chunk_index),
                                                    .start_time = block.first_sample_time,
                                                    .sample_count = static_cast<std::uint64_t>(chunk.size()),
                                                    .sample_start = sample_start,
                                                    .sample_end_exclusive =
                                                        sample_start + static_cast<std::uint64_t>(chunk.size())};
            addChunkToFile(*state.pulsed_file, chunk, block.first_sample_time, fullscale, chunk_index, &chunk_metadata);
            state.metadata.chunks.push_back(std::move(chunk_metadata));
            state.metadata.total_samples = state.metadata.chunks.back().sample_end_exclusive;
        }

References serial::addChunkToFile(), max, and processing::quantizeAndScaleWindow().

Referenced by submitBlock().

Here is the call graph for this function:

Here is the caller graph for this function:

Member Data Documentation

finalized

bool serial::Hdf5OutputSink::Impl::finalized = false

Definition at line 326 of file hdf5_output_sink.cpp.

Referenced by finalize(), and initializeRun().

metadata_collector

std::shared_ptr<core::OutputMetadataCollector> serial::Hdf5OutputSink::Impl::metadata_collector

Definition at line 321 of file hdf5_output_sink.cpp.

Referenced by closePulsedStream(), and closeStreamingStream().

mutex

std::recursive_mutex serial::Hdf5OutputSink::Impl::mutex

Definition at line 322 of file hdf5_output_sink.cpp.

Referenced by closeStream(), finalize(), initializeRun(), openStream(), registerStream(), and submitBlock().

next_stream_id

std::uint32_t serial::Hdf5OutputSink::Impl::next_stream_id = 1

Definition at line 325 of file hdf5_output_sink.cpp.

Referenced by registerStream().

output_dir

std::string serial::Hdf5OutputSink::Impl::output_dir

Definition at line 320 of file hdf5_output_sink.cpp.

Referenced by initializeRun(), openStream(), and outputPath().

stream_ids

std::unordered_map<std::string, std::uint32_t> serial::Hdf5OutputSink::Impl::stream_ids

Definition at line 324 of file hdf5_output_sink.cpp.

Referenced by registerStream().

streams

std::unordered_map<std::uint32_t, StreamState> serial::Hdf5OutputSink::Impl::streams

Definition at line 323 of file hdf5_output_sink.cpp.

Referenced by finalize(), registerStream(), and stateFor().

---

The documentation for this struct was generated from the following file:

• packages/libfers/src/serial/hdf5_output_sink.cpp