core.events_file_reader Namespace Reference

Classes
class	EventLogPosition
class	EventsFileReader

Functions
Message	proto_from_json_file (str|Path file_path, Message empty_proto_message)
bool	event_has_message (Event event, type[Any] message_type)
Message	payload_to_protobuf (Event event, bytes payload)
def	playback_command (args)

Variables
	parser = argparse.ArgumentParser()
	sub_parsers = parser.add_subparsers()
	playback_parser = sub_parsers.add_parser("playback")
	func
	args = parser.parse_args()

Function Documentation

event_has_message()

bool core.events_file_reader.event_has_message	(	Event	event,
		type[Any]	message_type
	)

Check if event has a message of type msg_type.

Args:
    event: event_pb2.Event
    msg_type: protobuf message type

Returns:
    bool: True if event has a message of type msg_type

payload_to_protobuf()

Message core.events_file_reader.payload_to_protobuf	(	Event	event,
		bytes	payload
	)

Return the protobuf message from an event and payload.

Args:
    event: event_pb2.Event
    payload: bytes

Returns:
    Message: the protobuf message

playback_command()

def core.events_file_reader.playback_command

(

args

)

proto_from_json_file()

Message core.events_file_reader.proto_from_json_file	(	str | Path	file_path,
		Message	empty_proto_message
	)

Load a proto Message from a JSON file.

The ``empty_proto_message`` must be the type of the proto message to load.

Args:
    file_path (str | Path): The path to the JSON file.
    empty_proto_message (Message): The empty proto message to parse into.

Returns:
    A proto message of the type given in empty_proto_message read from the JSON file.

Variable Documentation

args

core.events_file_reader.args = parser.parse_args()

func

core.events_file_reader.func

parser

core.events_file_reader.parser = argparse.ArgumentParser()

playback_parser

core.events_file_reader.playback_parser = sub_parsers.add_parser("playback")

sub_parsers

core.events_file_reader.sub_parsers = parser.add_subparsers()