pypz.sniffer.sniffer module

class pypz.sniffer.sniffer.ChannelSniffer(input_port: ChannelInputPort, output_port: ChannelOutputPort)

Bases: object

close()

on_status_message(status_messages: list[ChannelStatusMessage])

open()

sniff()

subscribe(channel_context_name, callback: Callable[[ChannelStatusMessage], None])

class pypz.sniffer.sniffer.PipelineSniffer(pipeline: Pipeline)

Bases: object

_start()

_stop()

static get_channel_id(input_port: ChannelInputPort, output_port: ChannelOutputPort) → str

get_channel_sniffer_by_port(input_port: ChannelInputPort, output_port: ChannelOutputPort)

start() → bool

stop() → bool