sgnts.sinks.dump

DumpSeriesSink dataclass

Bases: TSSink

              flowchart TD
              sgnts.sinks.dump.DumpSeriesSink[DumpSeriesSink]
              sgnts.base.base.TSSink[TSSink]
              sgnts.base.base.TimeSeriesMixin[TimeSeriesMixin]

                              sgnts.base.base.TSSink --> sgnts.sinks.dump.DumpSeriesSink
                                sgnts.base.base.TimeSeriesMixin --> sgnts.base.base.TSSink
                



              click sgnts.sinks.dump.DumpSeriesSink href "" "sgnts.sinks.dump.DumpSeriesSink"
              click sgnts.base.base.TSSink href "" "sgnts.base.base.TSSink"
              click sgnts.base.base.TimeSeriesMixin href "" "sgnts.base.base.TimeSeriesMixin"
            

A sink element that dumps time series data to a txt file.

Parameters:

Name	Type	Description	Default
fname	str	str, output file name	required
verbose	bool	bool, be verbose	False

Notes

Thread safety: Marked thread_safe = True. Pad layout: exactly 1 sink pad (@validator.single_pad + @sink.single_pad). No same-element pull concurrency. internal runs alone.

``pull`` (inherited): per-pad-keyed dict writes.
``process`` (called from ``internal``, runs alone): calls
``write_to_file(buf)`` per buffer, which appends to
``self.fname`` via ``np.savetxt``. Because ``process``
runs alone, file I/O is serial per element.

Speedup from threading is negligible inside this element
(file I/O dominates); marking thread_safe lets the sink
run concurrently with other thread_safe elements without
a serialization point in the wave.

**Future editors MUST preserve thread safety**:

1. Do NOT relax the single-sink-pad decorators without
   first wrapping ``write_to_file`` in synchronization,
   or routing each pad to a distinct file path.
2. Note that two ``DumpSeriesSink`` instances pointed at
   the same path will race on file appends — that is a
   configuration error we don't try to prevent.

Source code in src/sgnts/sinks/dump.py

@dataclass(kw_only=True)
class DumpSeriesSink(TSSink):
    """A sink element that dumps time series data to a txt file.

    Args:
        fname:
            str, output file name
        verbose:
            bool, be verbose

    Notes:
        Thread safety:
            Marked ``thread_safe = True``. Pad layout: exactly 1
            sink pad (``@validator.single_pad`` + ``@sink.single_pad``).
            No same-element ``pull`` concurrency. ``internal`` runs
            alone.

            ``pull`` (inherited): per-pad-keyed dict writes.
            ``process`` (called from ``internal``, runs alone): calls
            ``write_to_file(buf)`` per buffer, which appends to
            ``self.fname`` via ``np.savetxt``. Because ``process``
            runs alone, file I/O is serial per element.

            Speedup from threading is negligible inside this element
            (file I/O dominates); marking thread_safe lets the sink
            run concurrently with other thread_safe elements without
            a serialization point in the wave.

            **Future editors MUST preserve thread safety**:

            1. Do NOT relax the single-sink-pad decorators without
               first wrapping ``write_to_file`` in synchronization,
               or routing each pad to a distinct file path.
            2. Note that two ``DumpSeriesSink`` instances pointed at
               the same path will race on file appends — that is a
               configuration error we don't try to prevent.
    """

    thread_safe = True

    fname: str
    verbose: bool = False

    def configure(self) -> None:
        # overwrite existing file
        with open(self.fname, "w"):
            pass

    @validator.single_pad
    def validate(self) -> None:
        pass

    def write_to_file(self, buf) -> None:
        """Write time series data to txt file.

        Args:
            buf:
                SeriesBuffer, the buffer with time series data to write out
        """
        start = buf.start
        duration = buf.duration
        data = buf.data
        # FIXME: How to write multi-dimensional data?
        data = data.reshape(-1, data.shape[-1])
        ts = np.linspace(
            start,
            start + duration,
            data.shape[-1],
            endpoint=False,
        )
        out = np.vstack([ts, data]).T
        with open(self.fname, "ab") as f:
            np.savetxt(f, out)

    @sink.single_pad
    def process(self, input_frame: TSFrame) -> None:
        """Write out time-series data."""
        if input_frame.EOS:
            self.mark_eos(self.sink_pads[0])
        if self.verbose is True:
            print(input_frame)
        for buf in input_frame:
            if not buf.is_gap:
                self.write_to_file(buf)

process(input_frame)

Write out time-series data.

Source code in src/sgnts/sinks/dump.py

@sink.single_pad
def process(self, input_frame: TSFrame) -> None:
    """Write out time-series data."""
    if input_frame.EOS:
        self.mark_eos(self.sink_pads[0])
    if self.verbose is True:
        print(input_frame)
    for buf in input_frame:
        if not buf.is_gap:
            self.write_to_file(buf)

write_to_file(buf)

Write time series data to txt file.

Parameters:

Name	Type	Description	Default
buf		SeriesBuffer, the buffer with time series data to write out	required

Source code in src/sgnts/sinks/dump.py

def write_to_file(self, buf) -> None:
    """Write time series data to txt file.

    Args:
        buf:
            SeriesBuffer, the buffer with time series data to write out
    """
    start = buf.start
    duration = buf.duration
    data = buf.data
    # FIXME: How to write multi-dimensional data?
    data = data.reshape(-1, data.shape[-1])
    ts = np.linspace(
        start,
        start + duration,
        data.shape[-1],
        endpoint=False,
    )
    out = np.vstack([ts, data]).T
    with open(self.fname, "ab") as f:
        np.savetxt(f, out)