Source code for scrutiny.sdk.listeners.text_stream_listener
# text_stream_listener.py
# Simple listener useful for debug. Prints all updates in a text stream
#
# - License : MIT - See LICENSE file.
# - Project : Scrutiny Debugger (github.com/scrutinydebugger/scrutiny-python)
#
# Copyright (c) 2021 Scrutiny Debugger
__all__ = ['TextStreamListener']
from scrutiny.sdk.listeners import ValueUpdate
from . import BaseListener
import sys
import time
from typing import List, TextIO, Any
[docs]
class TextStreamListener(BaseListener):
_stream:TextIO
_start_time:float
[docs]
def __init__(self, stream:TextIO=sys.stdout, *args:Any, **kwargs:Any):
"""
Create a listener that writes every value update it receive into a text stream,
by formatting the update into a single-line string representation of the form:
``<time>ms\\t(<type>/<datatype>) <path>: <value>``.
Where
- <time> is the relative time in millisecond since the listener has been started
- <type> is the watchable type : variable, alias or rpv
- <datatype> is the value :class:`datatype<scrutiny.core.basic_types.EmbeddedDataType>`, such as : sint8, float32, uint16, etc.
- <path> is the tree path used to identify the watchable at the server level
- <value> Value converted to text
:param stream: The text stream to write to. Defaults to ``stdout``
:param args: Passed to :class:`BaseListener<scrutiny.sdk.listeners.BaseListener>`
:param kwargs: Passed to :class:`BaseListener<scrutiny.sdk.listeners.BaseListener>`
"""
BaseListener.__init__(self, *args, **kwargs)
self._stream = stream
self._start_time = 0
def setup(self) -> None:
self._start_time = time.perf_counter()
def receive(self, updates: List[ValueUpdate]) -> None:
update_time = (time.perf_counter() - self._start_time)*1e3
for update in updates:
self._stream.write(f'{update_time:0.2f}ms\t ({update.watchable_type.name}/{update.datatype.name}) {update.display_path}: {update.value}\n')
def teardown(self) -> None:
self._stream.flush()