multigp_toolkit.fpvscoresapi#

FPVScores Connections

Attributes#

logger

Module logger

BASE_API_URL

FPVScores API base URL

FPVS_API_VERSION

FPVScores Sync API version

LEGACY_HEADERS

Headers for FPVScores MultiGP requests

P

Generic for typing

R

Generic for typing

Classes#

FPVScoresAPI

The primary class used to interact with the FPVScores API

AlchemyEncoder

JSON encoder for SQLAlchemy objects

Functions#

standard_plugin_not_installed()

Check if the full FPVScores plugin is not installed

_assemble_pilots_complete(rhapi)

Gets the database pilots and adds their MultiGP id

_assemble_heatnodes_complete(rhapi)

Assembles heatnode data for FPVScores push

register_handlers(args)

Register export handlers

Module Contents#

multigp_toolkit.fpvscoresapi.logger#

Module logger

multigp_toolkit.fpvscoresapi.BASE_API_URL = 'https://api.fpvscores.com'#

FPVScores API base URL

multigp_toolkit.fpvscoresapi.FPVS_API_VERSION = '0.1.0'#

FPVScores Sync API version

multigp_toolkit.fpvscoresapi.LEGACY_HEADERS#

Headers for FPVScores MultiGP requests

multigp_toolkit.fpvscoresapi.P#

Generic for typing

multigp_toolkit.fpvscoresapi.R#

Generic for typing

multigp_toolkit.fpvscoresapi.standard_plugin_not_installed()#

Check if the full FPVScores plugin is not installed

Returns:

status of the install

Return type:

bool

class multigp_toolkit.fpvscoresapi.FPVScoresAPI(rhapi)#

Bases: multigp_toolkit.abstracts._APIManager

The primary class used to interact with the FPVScores API

Parameters:

rhapi (RHAPI.RHAPI)

_linked_org: bool | None = None#

Whether the current MultiGP chapter is linked to FPVScores or not

sync_ran: bool = False#

Status if a full sync was ran or not

_rhapi#

A stored instance of RHAPI

sync_guard#
_register_listeners()#

Registers the FPVScores event listeners

Return type:

None

connection_check()#

Checks for a connection to FPVScores

Returns:

The connection status

Return type:

bool

generate_fpvsconditions()#

Lazy loads and runs checks for fpvscores api actions

Yield:

Check statuses

Return type:

collections.abc.Generator[bool, None, None]

_generate_listener_conditions()#

Lazy loads and runs checks for event listeners

Yield:

Check statuses

Return type:

collections.abc.Generator[bool, None, None]

_check_listener_conditions()#

Decorator to run a series of checks before running event callback

Return type:

collections.abc.Callable[Concatenate[Self, P], R]

_parse_server_response(data)#

Attempts to parse the incoming data from the FPVScores server.

Parameters:

data (str) – The returned FPVScores data

Return type:

None

_process_response(greenlet)#

Wait for the response greenlet to finish. Attempt to parse the incoming data when completed.

Parameters:

greenlet (gevent.Greenlet) – The greenlet to wait for

Return type:

None

_class_listener_request(payload)#

Submits a request to the class update enpoint asynchronously using gevent

Parameters:

payload (dict) – The payload to send

Return type:

None

add_raceclass_listener(args)#

Sync the individual race class creation data to FPVScores

Parameters:

args (Union[dict, None]) – Default callback arguments

Return type:

None

alter_raceclass_listener(args)#

Sync the individual race class modification data to FPVScores

Parameters:

args (Union[dict, None]) – Default callback arguments

Return type:

None

get_race_channels()#

Gets the channel list in the FPVScores format

Returns:

The list of channels

Return type:

list[str]

class_delete(args)#

Deletes a single race class from FPVScores.

Parameters:

args (Union[dict, None]) – Default callback arguments

Return type:

None

heat_listener(args)#

Sync the individual heat data to FPVScores

Parameters:

args (Union[dict, None]) – Default callback arguments

Return type:

None

heat_delete(args)#

Deletes a single heat from FPVScores.

Parameters:

args (Union[dict, None]) – Default callback arguments

pilot_listener(args)#

Sync the individual pilot data to FPVScores

Parameters:

args (Union[dict, None]) – Default callback arguments

Return type:

None

generate_rank_payload(raceclass)#

Generate the rankings payload for FPVScores

Parameters:

raceclass (Database.RaceClass) – The raceclass to generate the payload from

Returns:

A generated list of pilot data

Return type:

list[dict]

generate_results_payload(raceclass)#

Generate the results payload for FPVScores

Parameters:

raceclass (Database.RaceClass) – The raceclass to generate the payload from

Returns:

A generated list of pilot data

Return type:

list[dict]

results_listener(args)#

Sync the individual class results to FPVScores

Parameters:

args (Union[dict, None]) – Default callback arguments

Return type:

None

run_full_sync(_args=None)#

Syncs the FPVScores event to the current RotorHazard state

Parameters:

_args (Union[dict, None]) – Default callback arguments

Return type:

None

get_event_url()#

Get the FPVScores event url for the active race

Returns:

The event url

Return type:

Union[str, None]

check_linked_org()#

Checks if the MultiGP API timer key in the system is linked to an FPVScores organization

Returns:

Whether the key is linked or not

Return type:

bool

multigp_toolkit.fpvscoresapi._assemble_pilots_complete(rhapi)#

Gets the database pilots and adds their MultiGP id to the upload data.

Returns:

The list of pilots

Parameters:

rhapi (RHAPI.RHAPI)

Return type:

list[Database.Pilot]

multigp_toolkit.fpvscoresapi._assemble_heatnodes_complete(rhapi)#

Assembles heatnode data for FPVScores push

Parameters:

rhapi (RHAPI.RHAPI) – An instance of RHAPI

Returns:

The formated payload

Return type:

dict

multigp_toolkit.fpvscoresapi.register_handlers(args)#

Register export handlers

Parameters:

args (Union[dict, None]) – Default callback arguments

Return type:

None

class multigp_toolkit.fpvscoresapi.AlchemyEncoder(*, skipkeys=False, ensure_ascii=True, check_circular=True, allow_nan=True, sort_keys=False, indent=None, separators=None, default=None)#

Bases: json.JSONEncoder

JSON encoder for SQLAlchemy objects

default(o)#

Implement this method in a subclass such that it returns a serializable object for o, or calls the base implementation (to raise a TypeError).

For example, to support arbitrary iterators, you could implement default like this:

def default(self, o):
    try:
        iterable = iter(o)
    except TypeError:
        pass
    else:
        return list(iterable)
    # Let the base class default method raise the TypeError
    return super().default(o)
Parameters:

o (object)