Skip to content

sgnts.sinks

FakeSeriesSink dataclass

Bases: TSSink

A fake series sink element.

Parameters:

Name Type Description Default
verbose bool

bool, be verbose

 False
Source code in sgnts/sinks/__init__.py 
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
@dataclass
class FakeSeriesSink(TSSink):
    """A fake series sink element.

    Args:
        verbose:
            bool, be verbose
    """

    verbose: bool = False

    def internal(self) -> None:
        """Print frames if verbose."""
        super().internal()
        for sink_pad in self.sink_pads:
            frame = self.preparedframes[sink_pad]
            if frame.EOS:
                self.mark_eos(sink_pad)
            if self.verbose is True:
                print(frame)
                latency = gpsnow() - Offset.tosec(
                    frame.offset + Offset.SAMPLE_STRIDE_AT_MAX_RATE
                )
                print(f"latency: {latency} s")

internal()

Print frames if verbose.

Source code in sgnts/sinks/__init__.py 
20
21
22
23
24
25
26
27
28
29
30
31
32
def internal(self) -> None:
    """Print frames if verbose."""
    super().internal()
    for sink_pad in self.sink_pads:
        frame = self.preparedframes[sink_pad]
        if frame.EOS:
            self.mark_eos(sink_pad)
        if self.verbose is True:
            print(frame)
            latency = gpsnow() - Offset.tosec(
                frame.offset + Offset.SAMPLE_STRIDE_AT_MAX_RATE
            )
            print(f"latency: {latency} s")

DumpSeriesSink dataclass

Bases: TSSink

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

Parameters:

Name Type Description Default
fname str

str, output file name

 'out.txt'
verbose bool

bool, be verbose

 False
Source code in sgnts/sinks/__init__.py 
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
@dataclass
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
    """

    fname: str = "out.txt"
    verbose: bool = False

    def __post_init__(self):
        super().__post_init__()
        if len(self.sink_pads) != 1:
            # FIXME: When will we use multiple sink pads here?
            # Do we want to support writing multiple frames into the same file?
            raise ValueError("Only supports one sink pad.")

        self.sink_pad = self.sink_pads[0]

        # overwrite existing file
        with open(self.fname, "w"):
            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
        """
        t0 = buf.t0
        duration = buf.duration
        data = buf.data
        # FIXME: How to write multi-dimensional data?
        data = data.reshape(-1, data.shape[-1])
        ts = np.linspace(
            t0 / Time.SECONDS,
            (t0 + duration) / Time.SECONDS,
            data.shape[-1],
            endpoint=False,
        )
        out = np.vstack([ts, data]).T
        with open(self.fname, "a") as f:
            np.savetxt(f, out)

    def internal(self) -> None:
        """Write out time-series data."""
        super().internal()
        sink_pad = self.sink_pad
        frame = self.preparedframes[sink_pad]
        if frame.EOS:
            self.mark_eos(sink_pad)
        if self.verbose is True:
            print(frame)
        for buf in frame:
            if not buf.is_gap:
                self.write_to_file(buf)

internal()

Write out time-series data.

Source code in sgnts/sinks/__init__.py 
84
85
86
87
88
89
90
91
92
93
94
95
def internal(self) -> None:
    """Write out time-series data."""
    super().internal()
    sink_pad = self.sink_pad
    frame = self.preparedframes[sink_pad]
    if frame.EOS:
        self.mark_eos(sink_pad)
    if self.verbose is True:
        print(frame)
    for buf in 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 sgnts/sinks/__init__.py 
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
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
    """
    t0 = buf.t0
    duration = buf.duration
    data = buf.data
    # FIXME: How to write multi-dimensional data?
    data = data.reshape(-1, data.shape[-1])
    ts = np.linspace(
        t0 / Time.SECONDS,
        (t0 + duration) / Time.SECONDS,
        data.shape[-1],
        endpoint=False,
    )
    out = np.vstack([ts, data]).T
    with open(self.fname, "a") as f:
        np.savetxt(f, out)