Real-Time Client - livef1.adapters.realtime_client

Classes

RealF1Client(topics[, log_file_name, ...])	A client for managing real-time Formula 1 data streaming.
MessageHandlerTemplate(func)	A template for handling incoming SignalR messages.

class livef1.adapters.realtime_client.MessageHandlerTemplate(func)

A template for handling incoming SignalR messages.

This class serves as a message handler for SignalR streams, where incoming messages are processed and passed to a user-defined function.

Parameters:

funccallable

A user-defined asynchronous function that processes the parsed records from incoming SignalR messages. The function must accept the processed records as its input.

Methods

get(msg)

Process incoming messages and invoke the handler function.

async get(msg)

Process incoming messages and invoke the handler function.

The method handles two types of incoming message formats:

• Messages in the “R” key: Representing topics with associated data.

• Messages in the “M” key: Representing method calls with data payloads.

For each message, the method uses a function_map to parse the data and sends the resulting records to the user-defined handler function.

Parameters:

msgdict

The incoming message in SignalR format. It is expected to contain either: - “R”: A dictionary where the keys represent topic names and the values

are the associated data for that topic.

• “M”: A list of dictionaries, where each dictionary contains: - “M”: The method name. - “A”: A list of message components including topic name, data, and timestamp.

class livef1.adapters.realtime_client.RealF1Client(topics, log_file_name=None, log_file_mode='w')

A client for managing real-time Formula 1 data streaming.

Parameters:

topicsstr or list

Topic(s) to subscribe to for live updates.

log_file_namestr, optional

Name of the log file (default is None).

log_file_modestr, optional

Mode for opening the log file (default is “w”).

Attributes:

topicslist

List of topics to subscribe to for receiving live data.

headersdict

HTTP headers used for the connection.

_connection_urlstr

URL for the SignalR connection.

_log_file_namestr

Path to the log file.

_log_file_modestr

Mode for opening the log file.

_testbool

Indicates if the client is in test mode.

_log_filefile object

Log file object for writing messages (used in test mode).

_handlersdict

Mapping of methods to their respective handlers.

Methods

callback(method)	Decorator to register a callback function for a specific method.
on_message(method, handler)	Register a handler for a specific method.
run()	Start the client in asynchronous mode.

callback(method)

Decorator to register a callback function for a specific method.

This decorator allows you to associate a callback function with a particular method. The function being registered must have arguments matching the required parameters defined in REALTIME_CALLBACK_DEFAULT_PARAMETERS.

Raises:

TypeError

If the provided callback function does not have the required arguments.

Notes

• The REALTIME_CALLBACK_DEFAULT_PARAMETERS is a predefined list of parameter names that the callback function must include.

• Once the function is successfully registered, it will handle messages for the specified method.

Examples

Registering a callback for a method:

from livef1.adapters.realtime_client import RealF1Client

client = RealF1Client(
    topics = ["CarData.z", "SessionInfo"],
    log_file_name="./output.json"
)

@client.callback("new one")
async def print_callback(records): # records argument have to be set
    print(records) # or you can do whatever you want with incoming data

client.run()

on_message(method, handler)

Register a handler for a specific method.

Parameters:

methodstr

The method to handle.

handlercallable

The function to handle the method.

run()

Start the client in asynchronous mode.