sgn.testpoint

SGN Pipeline testpoints

This module adds "testpoint" capability to sgn applications. When the TESTPOINT environment variable is appropriately defined, various statistics will be captured from pipeline pads and displayed in a live updating table in the terminal.

If TESTPOINT=frame, the frame information passing through pads will be collected and displayed in the table.

If TESTPOINT=exec, pad execution times will be collected, and execution time statistics will be displayed in the table.

The TESTPOINT_REGEX environment variable is a space-separated list of regular expressions (regex) that control which pads are collected. The regex are matched against the pad full names, which are of the form:

{element}:{pad_type}:{pad_name}

Examples:

Display the execution time all internal pads in the pipeline:

TESTPOINT=exec TESTPOINT_REGEX=".:inl:." my_sgn_pipeline ...

Display information about frames flowing into all sink pads ("snk") of element "foo" that start with the letter "a":

TESTPOINT=frame TESTPOINT_REGEX="foo:snk:a.*" my_sgn_pipeline ...

The TESTPOINT_FIELDS environment variable is a space-separated list of the frame fields to display, in the order given. If not specified a default set of fields is displayed. The recognized field names are:

repr, type, offset, noffset, offset_end, start, start_ns, duration, duration_ns, buffers (bufs), sample_rate (rate), sample_shape (shape), samples (length, len), gap, dtype, data, latency

Names in parentheses are aliases for the preceding field. Fields that a given frame type does not provide (and unrecognized field names) are displayed as "-". TESTPOINT_FIELDS only affects the TESTPOINT=frame table; the TESTPOINT=exec table always displays the same execution time statistics.

Display just the offset and latency of frames flowing through all pads:

TESTPOINT=frame TESTPOINT_FIELDS="offset latency" my_sgn_pipeline ...

The create_testpoint() function can also be used to create the testpoint display programmatically.

TestpointDisplay

Display pad information in a "live" table in the terminal.

Parameters:

Name	Type	Description	Default
padspec	list[str] | str	List of regular expressions to determine which pads to include in the table. If not specified all pads will be included. [".*"]	'.*'
fields	Iterable[str] | None	List of fields to display in the table. The first field will always be the pad name. The iterable is consumed immediately, so one-shot iterables are acceptable. [DEFAULT_FIELDS]	None
sort	bool	Bool, whether or not to sort table. [True]	True
live	bool	Bool, whether to create a live updating table or not. [True]	True

Source code in src/sgn/testpoint.py

class TestpointDisplay:
    """Display pad information in a "live" table in the terminal.

    Args:
        padspec:
            List of regular expressions to determine which pads to
            include in the table. If not specified all pads will be
            included. [".*"]
        fields:
            List of fields to display in the table. The first field
            will always be the pad name. The iterable is consumed
            immediately, so one-shot iterables are acceptable.
            [DEFAULT_FIELDS]
        sort:
            Bool, whether or not to sort table. [True]
        live:
            Bool, whether to create a live updating table or
            not. [True]

    """

    DEFAULT_FIELDS: list[str] = []
    DEFAULT_DICT: Callable | None = None

    def __init__(
        self,
        padspec: list[str] | str = ".*",
        fields: Iterable[str] | None = None,
        *,
        sort: bool = True,
        live: bool = True,
    ):
        try:
            from rich.console import Console
            from rich.live import Live
        except ModuleNotFoundError:
            raise ModuleNotFoundError(
                "testpoint functionality requires the `rich` package."
            )

        if isinstance(padspec, str):
            self.padspec = [padspec]
        else:
            self.padspec = padspec
        # NOTE materialize fields, since they are iterated once per
        # table row in make_table()
        self.fields: list[str] = list(fields) if fields else self.DEFAULT_FIELDS
        self.sort = sort
        self.console = Console()
        self.live: Live | None = None
        if live:
            self.live = Live(console=self.console, auto_refresh=False)
        self.testpoints: dict[str, Any] = collections.defaultdict(self.DEFAULT_DICT)
        self.selected: dict[str, bool] = collections.defaultdict(bool)
        self.iteration = 0

    def select(self, pad_name: str):
        """True if pad_name matches self.padspec"""
        if pad_name not in self.selected:
            for padspec in self.padspec:
                if re.match(padspec, pad_name):
                    self.selected[pad_name] = True
                    break
        return self.selected[pad_name]

    def make_table(self):
        """Generate display table"""
        raise NotImplementedError()

    def update(self):
        """Update the table display"""
        from rich.table import Table

        if self.live and not self.live.is_started:
            self.live.start()

        table = Table(box=None)
        table.add_row(self.make_table())
        table.add_row(f"iteration: {self.iteration}")
        if self.live:
            self.live.update(table, refresh=True)
        else:
            self.console.print(table)
        self.iteration += 1

    def stop(self):
        """Stop the live display"""
        if self.live:
            self.live.stop()

make_table()

Generate display table

Source code in src/sgn/testpoint.py

def make_table(self):
    """Generate display table"""
    raise NotImplementedError()

select(pad_name)

True if pad_name matches self.padspec

Source code in src/sgn/testpoint.py

def select(self, pad_name: str):
    """True if pad_name matches self.padspec"""
    if pad_name not in self.selected:
        for padspec in self.padspec:
            if re.match(padspec, pad_name):
                self.selected[pad_name] = True
                break
    return self.selected[pad_name]

stop()

Stop the live display

Source code in src/sgn/testpoint.py

def stop(self):
    """Stop the live display"""
    if self.live:
        self.live.stop()

update()

Update the table display

Source code in src/sgn/testpoint.py

def update(self):
    """Update the table display"""
    from rich.table import Table

    if self.live and not self.live.is_started:
        self.live.start()

    table = Table(box=None)
    table.add_row(self.make_table())
    table.add_row(f"iteration: {self.iteration}")
    if self.live:
        self.live.update(table, refresh=True)
    else:
        self.console.print(table)
    self.iteration += 1

TestpointExec

Bases: TestpointDisplay

              flowchart TD
              sgn.testpoint.TestpointExec[TestpointExec]
              sgn.testpoint.TestpointDisplay[TestpointDisplay]

                              sgn.testpoint.TestpointDisplay --> sgn.testpoint.TestpointExec
                


              click sgn.testpoint.TestpointExec href "" "sgn.testpoint.TestpointExec"
              click sgn.testpoint.TestpointDisplay href "" "sgn.testpoint.TestpointDisplay"
            

Display information about pad execution times

Source code in src/sgn/testpoint.py

class TestpointExec(TestpointDisplay):
    """Display information about pad execution times"""

    DEFAULT_FIELDS = [
        r"exec time mean \[ns]",
        r"exec time stdev \[ns]",
        "# samples",
    ]

    # NOTE staticmethod is for mypy only: functools.partial is not a
    # descriptor, so this changes nothing at runtime.  It stops mypy from
    # treating the Callable-annotated class attribute as an implicitly
    # bound method where it's accessed as self.DEFAULT_DICT.
    DEFAULT_DICT: Callable = staticmethod(
        functools.partial(collections.deque, maxlen=10)
    )

    def record_exec(self, pad, dt):
        """Record pad execution time"""
        if self.select(pad.name):
            self.testpoints[pad.name].append(int(dt * 1e9))

    def make_table(self):
        """Make table of pad execution times"""
        from rich.table import Table

        table = Table(box=None)
        table.add_column("pad")
        for field in self.fields:
            table.add_column(field, justify="right")
        f_iter = [
            (
                pad_name,
                (
                    statistics.mean(samples),
                    _stdev(samples),
                    len(samples),
                ),
            )
            for pad_name, samples in self.testpoints.items()
        ]
        if self.sort:
            # sort on execution time
            f_iter = sorted(f_iter, key=lambda x: x[1][0], reverse=True)
        for pad_name, fields in f_iter:
            values = [f"{round(val):_}" for val in fields]
            table.add_row(pad_name, *values)
        return table

make_table()

Make table of pad execution times

Source code in src/sgn/testpoint.py

def make_table(self):
    """Make table of pad execution times"""
    from rich.table import Table

    table = Table(box=None)
    table.add_column("pad")
    for field in self.fields:
        table.add_column(field, justify="right")
    f_iter = [
        (
            pad_name,
            (
                statistics.mean(samples),
                _stdev(samples),
                len(samples),
            ),
        )
        for pad_name, samples in self.testpoints.items()
    ]
    if self.sort:
        # sort on execution time
        f_iter = sorted(f_iter, key=lambda x: x[1][0], reverse=True)
    for pad_name, fields in f_iter:
        values = [f"{round(val):_}" for val in fields]
        table.add_row(pad_name, *values)
    return table

record_exec(pad, dt)

Record pad execution time

Source code in src/sgn/testpoint.py

def record_exec(self, pad, dt):
    """Record pad execution time"""
    if self.select(pad.name):
        self.testpoints[pad.name].append(int(dt * 1e9))

TestpointFrame

Bases: TestpointDisplay

              flowchart TD
              sgn.testpoint.TestpointFrame[TestpointFrame]
              sgn.testpoint.TestpointDisplay[TestpointDisplay]

                              sgn.testpoint.TestpointDisplay --> sgn.testpoint.TestpointFrame
                


              click sgn.testpoint.TestpointFrame href "" "sgn.testpoint.TestpointFrame"
              click sgn.testpoint.TestpointDisplay href "" "sgn.testpoint.TestpointDisplay"
            

Display information about frames passing through pads

Source code in src/sgn/testpoint.py

class TestpointFrame(TestpointDisplay):
    """Display information about frames passing through pads"""

    DEFAULT_FIELDS = [
        "type",
        "offset",
        "noffset",
        "bufs",
        "has gap",
        "rate",
        "samples",
        "dtype",
        "data",
        "latency",
    ]

    DEFAULT_DICT = None

    def record_frame(self, pad: Pad, frame: Frame, pad_type: str = ""):
        """Record pad frame data

        `pad_type` will override the inherent pad.pad_type

        """
        pad_name = pad.name
        if pad_type:
            pad_name = pad_name.replace(f":{pad.pad_type}:", f":{pad_type}:", 1)
        if self.select(pad_name):
            self.testpoints[pad_name] = frame

    def make_table(self):
        """Make table of pad frame info"""
        from rich.table import Table

        table = Table(box=None)
        table.add_column("pad")
        for field in self.fields:
            table.add_column(_FIELD_HEADERS.get(field, field))
        f_iter = self.testpoints.items()
        if self.sort:
            f_iter = sorted(f_iter, key=lambda x: x[0])
        for pad_name, frame in f_iter:
            values = []
            for field in self.fields:
                try:
                    fstr = _frame_field_format(frame, field)
                except (AttributeError, ValueError, IndexError, TypeError):
                    # unavailable/unknown fields are displayed as "-",
                    # but log why so that a permanently empty column
                    # can be diagnosed
                    logger.debug(
                        "%s: could not format field '%s'",
                        pad_name,
                        field,
                        exc_info=True,
                    )
                    fstr = "-"
                values.append(fstr)
            table.add_row(pad_name, *values)
        return table

make_table()

Make table of pad frame info

Source code in src/sgn/testpoint.py

def make_table(self):
    """Make table of pad frame info"""
    from rich.table import Table

    table = Table(box=None)
    table.add_column("pad")
    for field in self.fields:
        table.add_column(_FIELD_HEADERS.get(field, field))
    f_iter = self.testpoints.items()
    if self.sort:
        f_iter = sorted(f_iter, key=lambda x: x[0])
    for pad_name, frame in f_iter:
        values = []
        for field in self.fields:
            try:
                fstr = _frame_field_format(frame, field)
            except (AttributeError, ValueError, IndexError, TypeError):
                # unavailable/unknown fields are displayed as "-",
                # but log why so that a permanently empty column
                # can be diagnosed
                logger.debug(
                    "%s: could not format field '%s'",
                    pad_name,
                    field,
                    exc_info=True,
                )
                fstr = "-"
            values.append(fstr)
        table.add_row(pad_name, *values)
    return table

record_frame(pad, frame, pad_type='')

Record pad frame data

pad_type will override the inherent pad.pad_type

Source code in src/sgn/testpoint.py

def record_frame(self, pad: Pad, frame: Frame, pad_type: str = ""):
    """Record pad frame data

    `pad_type` will override the inherent pad.pad_type

    """
    pad_name = pad.name
    if pad_type:
        pad_name = pad_name.replace(f":{pad.pad_type}:", f":{pad_type}:", 1)
    if self.select(pad_name):
        self.testpoints[pad_name] = frame

create_testpoint(ttype, *args, **kwargs)

Create global testpoint object

Source code in src/sgn/testpoint.py

def create_testpoint(ttype, *args, **kwargs):
    """Create global testpoint object"""
    global TESTPOINT
    if ttype == "frame":
        TESTPOINT = TestpointFrame(*args, **kwargs)
    elif ttype == "exec":
        TESTPOINT = TestpointExec(*args, **kwargs)
    else:
        msg = f"invalid testpoint type '{ttype}'"
        raise ValueError(msg)

record_exec(*args)

Record pad execution time

Source code in src/sgn/testpoint.py

def record_exec(*args):
    """Record pad execution time"""
    if not _ENV_BOOTSTRAP_TRIED:
        _create_from_env()
    if TESTPOINT and isinstance(TESTPOINT, TestpointExec):
        TESTPOINT.record_exec(*args)

record_frame(*args, **kwargs)

Record pad frame

Source code in src/sgn/testpoint.py

def record_frame(*args, **kwargs):
    """Record pad frame"""
    if not _ENV_BOOTSTRAP_TRIED:
        _create_from_env()
    if TESTPOINT and isinstance(TESTPOINT, TestpointFrame):
        TESTPOINT.record_frame(*args, **kwargs)

stop()

Stop testpoint display

Source code in src/sgn/testpoint.py

def stop():
    """Stop testpoint display"""
    if TESTPOINT:
        TESTPOINT.stop()

update()

Update testpoint display

Source code in src/sgn/testpoint.py

def update():
    """Update testpoint display"""
    if not _ENV_BOOTSTRAP_TRIED:
        _create_from_env()
    if TESTPOINT:
        TESTPOINT.update()