Utils - livef1.utils

Helper

Functions

build_session_endpoint(session_path)

Constructs a full endpoint URL for accessing session data.

json_parser_for_objects(data)

Converts the keys of a dictionary to lowercase.

get_data(path, stream)

Fetches data from a specified endpoint.

get_car_data_stream(path)

Fetches car data from a specified endpoint and returns it as a dictionary.

parse(text[, zipped])

Parses a given text input and decompresses it if necessary.

parse_hash(hash_code)

Parses a hashed string and decompresses it.

parse_helper_for_nested_dict(info, record[, ...])

Recursively parses a nested dictionary and flattens it into a single-level dictionary.

find_most_similar_vectorized(df, target)

Find the most similar string in a Pandas DataFrame using Jaccard and Jaro-Winkler similarity.

livef1.utils.helper.build_session_endpoint(session_path)[source]

Constructs a full endpoint URL for accessing session data.

Parameters:
session_pathstr

The path for the specific session data.

Returns:
str

The complete URL for the session endpoint.

livef1.utils.helper.find_most_similar_vectorized(df, target)[source]

Find the most similar string in a Pandas DataFrame using Jaccard and Jaro-Winkler similarity.

Parameters:
dfpd.DataFrame

The DataFrame to search in.

targetstr

The string to search for.

Returns:
dict
A dictionary containing:
  • “isFound” (int): 1 if a match is found, 0 otherwise.

  • “how” (str): The method used for matching (“jaccard” or “jaro”).

  • “value” (str): The most similar value found.

  • “similarity” (float): The similarity score of the match.

  • “row” (int): The row index of the match.

  • “column” (str): The column name of the match.

Raises:
livef1Exception

If no match is found and suggestions are provided.

livef1.utils.helper.get_car_data_stream(path)[source]

Fetches car data from a specified endpoint and returns it as a dictionary.

Parameters:
pathstr

The endpoint to retrieve car data from.

Returns:
dict

A dictionary where keys are the first 12 characters of each record and values are the remaining data.

livef1.utils.helper.get_data(path, stream)[source]

Fetches data from a specified endpoint.

Parameters:
pathstr

The endpoint to retrieve data from.

streambool

Indicates whether to return a stream of records or a single response.

Returns:
Union[dict, str]

A dictionary of records if stream is True, else a string response.

livef1.utils.helper.identifer_text_format(text)[source]

Formats text for comparison by splitting into words and removing stopwords.

Parameters:
textstr

The input text to format.

Returns:
list

A list of words from the input text with stopwords removed.

livef1.utils.helper.json_parser_for_objects(data: Dict) Dict[source]

Converts the keys of a dictionary to lowercase.

Parameters:
dataDict

The original dictionary with keys.

Returns:
Dict

A new dictionary with all keys converted to lowercase.

livef1.utils.helper.parse(text: str, zipped: bool = False) str | dict[source]

Parses a given text input and decompresses it if necessary.

Parameters:
textstr

The input text to be parsed.

zippedbool, optional

Indicates if the input is a zipped string, by default False.

Returns:
Union[str, dict]

The parsed output as a dictionary if input is JSON, otherwise as a string.

livef1.utils.helper.parse_hash(hash_code)[source]

Parses a hashed string and decompresses it.

Parameters:
hash_codestr

The hash string to be parsed.

Returns:
dict

The decompressed and parsed data as a dictionary.

livef1.utils.helper.parse_helper_for_nested_dict(info, record, prefix='')[source]

Recursively parses a nested dictionary and flattens it into a single-level dictionary.

Parameters:
infodict

The nested dictionary to parse.

recorddict

The record to which parsed information will be added.

prefixstr, optional

A prefix for keys in the flattened dictionary, by default “”.

Returns:
dict

The updated record with flattened keys from the nested dictionary.

livef1.utils.helper.print_found_model(df, key, cols)[source]
livef1.utils.helper.to_datetime(var)[source]

Exceptions

Functions

exception livef1.utils.exceptions.AdapterError(message)[source]

Base exception for adapter-related issues.

add_note()

Exception.add_note(note) – add a note to the exception

args
with_traceback()

Exception.with_traceback(tb) – set self.__traceback__ to tb and return self.

exception livef1.utils.exceptions.ArgumentError(message)[source]

Exception for arguments of methods related errors

add_note()

Exception.add_note(note) – add a note to the exception

args
with_traceback()

Exception.with_traceback(tb) – set self.__traceback__ to tb and return self.

exception livef1.utils.exceptions.DataDecodingError(message)[source]

Raised when decoding the response fails.

add_note()

Exception.add_note(note) – add a note to the exception

args
with_traceback()

Exception.with_traceback(tb) – set self.__traceback__ to tb and return self.

exception livef1.utils.exceptions.DataFormatError(message)[source]

Exception for unexpected data formats.

add_note()

Exception.add_note(note) – add a note to the exception

args
with_traceback()

Exception.with_traceback(tb) – set self.__traceback__ to tb and return self.

exception livef1.utils.exceptions.DataProcessingError(message)[source]

Raised when data processing / parsing related error occurs.

add_note()

Exception.add_note(note) – add a note to the exception

args
with_traceback()

Exception.with_traceback(tb) – set self.__traceback__ to tb and return self.

exception livef1.utils.exceptions.ETLError(message)[source]

Exception for ETL-specific issues.

add_note()

Exception.add_note(note) – add a note to the exception

args
with_traceback()

Exception.with_traceback(tb) – set self.__traceback__ to tb and return self.

exception livef1.utils.exceptions.InvalidEndpointError(message)[source]

Raised when an invalid endpoint is accessed.

add_note()

Exception.add_note(note) – add a note to the exception

args
with_traceback()

Exception.with_traceback(tb) – set self.__traceback__ to tb and return self.

exception livef1.utils.exceptions.InvalidResponseError(message)[source]

Exception for invalid API responses.

add_note()

Exception.add_note(note) – add a note to the exception

args
with_traceback()

Exception.with_traceback(tb) – set self.__traceback__ to tb and return self.

exception livef1.utils.exceptions.LiveF1Error(message)[source]

Base class for all LiveF1 module exceptions.

add_note()

Exception.add_note(note) – add a note to the exception

args
with_traceback()

Exception.with_traceback(tb) – set self.__traceback__ to tb and return self.

exception livef1.utils.exceptions.MissingFunctionError(message)[source]

Raised when ETL functions does not include a function

add_note()

Exception.add_note(note) – add a note to the exception

args
with_traceback()

Exception.with_traceback(tb) – set self.__traceback__ to tb and return self.

exception livef1.utils.exceptions.ParsingError(message)[source]

Raised when parsing the data fails.

add_note()

Exception.add_note(note) – add a note to the exception

args
with_traceback()

Exception.with_traceback(tb) – set self.__traceback__ to tb and return self.

exception livef1.utils.exceptions.RealF1Error(message)[source]

Exception for RealF1Client related errors

add_note()

Exception.add_note(note) – add a note to the exception

args
with_traceback()

Exception.with_traceback(tb) – set self.__traceback__ to tb and return self.

exception livef1.utils.exceptions.SubscriptionError(message)[source]

Exception for subscription errors.

add_note()

Exception.add_note(note) – add a note to the exception

args
with_traceback()

Exception.with_traceback(tb) – set self.__traceback__ to tb and return self.

exception livef1.utils.exceptions.livef1Exception[source]
add_note()

Exception.add_note(note) – add a note to the exception

args
with_traceback()

Exception.with_traceback(tb) – set self.__traceback__ to tb and return self.

Logger

Functions

livef1.utils.logger.set_log_level(level)[source]

Set the logging level for the livef1 logger.

Parameters:
levelUnion[str, int]

The logging level to set. Can be either a string (‘DEBUG’, ‘INFO’, ‘WARNING’, ‘ERROR’, ‘CRITICAL’) or the corresponding integer value.

Examples

>>> set_log_level('DEBUG')  # Set to debug level
>>> set_log_level(logging.INFO)  # Set to info level