# Released under the MIT License. See LICENSE for details.
#
"""Functionality related to cloud functionality."""
from __future__ import annotations
import datetime
from enum import Enum
from dataclasses import dataclass, field
from typing import TYPE_CHECKING, Annotated, override
from efro.message import Message, Response
from efro.dataclassio import ioprepped, IOAttrs
from bacommon.transfer import DirectoryManifest
from bacommon.login import LoginType
if TYPE_CHECKING:
pass
[docs]
class WebLocation(Enum):
"""Set of places we can be directed on ballistica.net."""
ACCOUNT_EDITOR = 'e'
ACCOUNT_DELETE_SECTION = 'd'
[docs]
@ioprepped
@dataclass
class LoginProxyRequestMessage(Message):
"""Request send to the cloud to ask for a login-proxy."""
[docs]
@override
@classmethod
def get_response_types(cls) -> list[type[Response] | None]:
return [LoginProxyRequestResponse]
[docs]
@ioprepped
@dataclass
class LoginProxyRequestResponse(Response):
"""Response to a request for a login proxy."""
# URL to direct the user to for sign in.
url: Annotated[str, IOAttrs('u')]
# URL to use for overlay-web-browser sign ins.
url_overlay: Annotated[str, IOAttrs('uo')]
# Proxy-Login id for querying results.
proxyid: Annotated[str, IOAttrs('p')]
# Proxy-Login key for querying results.
proxykey: Annotated[str, IOAttrs('k')]
[docs]
@ioprepped
@dataclass
class LoginProxyStateQueryMessage(Message):
"""Soo.. how is that login proxy going?"""
proxyid: Annotated[str, IOAttrs('p')]
proxykey: Annotated[str, IOAttrs('k')]
[docs]
@override
@classmethod
def get_response_types(cls) -> list[type[Response] | None]:
return [LoginProxyStateQueryResponse]
[docs]
@ioprepped
@dataclass
class LoginProxyStateQueryResponse(Response):
"""Here's the info on that login-proxy you asked about, boss."""
[docs]
class State(Enum):
"""States a login-proxy can be in."""
WAITING = 'waiting'
SUCCESS = 'success'
FAIL = 'fail'
state: Annotated[State, IOAttrs('s')]
# On success, these will be filled out.
credentials: Annotated[str | None, IOAttrs('tk')]
[docs]
@ioprepped
@dataclass
class LoginProxyCompleteMessage(Message):
"""Just so you know, we're done with this proxy."""
proxyid: Annotated[str, IOAttrs('p')]
[docs]
@ioprepped
@dataclass
class PingMessage(Message):
"""Standard ping."""
[docs]
@override
@classmethod
def get_response_types(cls) -> list[type[Response] | None]:
return [PingResponse]
[docs]
@ioprepped
@dataclass
class PingResponse(Response):
"""pong."""
[docs]
@ioprepped
@dataclass
class TestMessage(Message):
"""Can I get some of that workspace action?"""
testfoo: Annotated[int, IOAttrs('f')]
[docs]
@override
@classmethod
def get_response_types(cls) -> list[type[Response] | None]:
return [TestResponse]
[docs]
@ioprepped
@dataclass
class TestResponse(Response):
"""Here's that workspace you asked for, boss."""
testfoo: Annotated[int, IOAttrs('f')]
[docs]
@ioprepped
@dataclass
class SendInfoMessage(Message):
"""User is using the send-info function"""
description: Annotated[str, IOAttrs('c')]
[docs]
@override
@classmethod
def get_response_types(cls) -> list[type[Response] | None]:
return [SendInfoResponse]
[docs]
@ioprepped
@dataclass
class SendInfoResponse(Response):
"""Response to sending into the server."""
handled: Annotated[bool, IOAttrs('v')]
message: Annotated[str | None, IOAttrs('m', store_default=False)] = None
legacy_code: Annotated[str | None, IOAttrs('l', store_default=False)] = None
[docs]
@ioprepped
@dataclass
class WorkspaceFetchState:
"""Common state data for a workspace fetch."""
manifest: Annotated[DirectoryManifest, IOAttrs('m')]
iteration: Annotated[int, IOAttrs('i')] = 0
total_deletes: Annotated[int, IOAttrs('tdels')] = 0
total_downloads: Annotated[int, IOAttrs('tdlds')] = 0
total_up_to_date: Annotated[int | None, IOAttrs('tunmd')] = None
[docs]
@ioprepped
@dataclass
class WorkspaceFetchMessage(Message):
"""Can I get some of that workspace action?"""
workspaceid: Annotated[str, IOAttrs('w')]
state: Annotated[WorkspaceFetchState, IOAttrs('s')]
[docs]
@override
@classmethod
def get_response_types(cls) -> list[type[Response] | None]:
return [WorkspaceFetchResponse]
[docs]
@ioprepped
@dataclass
class WorkspaceFetchResponse(Response):
"""Here's that workspace you asked for, boss."""
state: Annotated[WorkspaceFetchState, IOAttrs('s')]
deletes: Annotated[list[str], IOAttrs('dlt', store_default=False)] = field(
default_factory=list
)
downloads_inline: Annotated[
dict[str, bytes], IOAttrs('dinl', store_default=False)
] = field(default_factory=dict)
done: Annotated[bool, IOAttrs('d')] = False
[docs]
@ioprepped
@dataclass
class MerchAvailabilityMessage(Message):
"""Can we show merch link?"""
[docs]
@override
@classmethod
def get_response_types(cls) -> list[type[Response] | None]:
return [MerchAvailabilityResponse]
[docs]
@ioprepped
@dataclass
class MerchAvailabilityResponse(Response):
"""About that merch..."""
url: Annotated[str | None, IOAttrs('u')]
[docs]
@ioprepped
@dataclass
class SignInMessage(Message):
"""Can I sign in please?"""
login_type: Annotated[LoginType, IOAttrs('l')]
sign_in_token: Annotated[str, IOAttrs('t')]
# For debugging. Can remove soft_default once build 20988+ is ubiquitous.
description: Annotated[str, IOAttrs('d', soft_default='-')]
apptime: Annotated[float, IOAttrs('at', soft_default=-1.0)]
[docs]
@override
@classmethod
def get_response_types(cls) -> list[type[Response] | None]:
return [SignInResponse]
[docs]
@ioprepped
@dataclass
class SignInResponse(Response):
"""Here's that sign-in result you asked for, boss."""
credentials: Annotated[str | None, IOAttrs('c')]
[docs]
@ioprepped
@dataclass
class ManageAccountMessage(Message):
"""Message asking for a manage-account url."""
weblocation: Annotated[WebLocation, IOAttrs('l')] = (
WebLocation.ACCOUNT_EDITOR
)
[docs]
@override
@classmethod
def get_response_types(cls) -> list[type[Response] | None]:
return [ManageAccountResponse]
[docs]
@ioprepped
@dataclass
class ManageAccountResponse(Response):
"""Here's that sign-in result you asked for, boss."""
url: Annotated[str | None, IOAttrs('u')]
[docs]
@ioprepped
@dataclass
class StoreQueryMessage(Message):
"""Message asking about purchasable stuff and store related state."""
[docs]
@override
@classmethod
def get_response_types(cls) -> list[type[Response] | None]:
return [StoreQueryResponse]
[docs]
@ioprepped
@dataclass
class StoreQueryResponse(Response):
"""Here's that store info you asked for, boss."""
[docs]
class Result(Enum):
"""Our overall result."""
SUCCESS = 's'
ERROR = 'e'
[docs]
@dataclass
class Purchase:
"""Info about a purchasable thing."""
purchaseid: Annotated[str, IOAttrs('id')]
# Overall result; all data is undefined if not SUCCESS.
result: Annotated[Result, IOAttrs('r')]
tokens: Annotated[int, IOAttrs('t')]
gold_pass: Annotated[bool, IOAttrs('g')]
available_purchases: Annotated[list[Purchase], IOAttrs('p')]
token_info_url: Annotated[str, IOAttrs('tiu')]
[docs]
@ioprepped
@dataclass
class BSPrivatePartyMessage(Message):
"""Message asking about info we need for private-party UI."""
need_datacode: Annotated[bool, IOAttrs('d')]
[docs]
@override
@classmethod
def get_response_types(cls) -> list[type[Response] | None]:
return [BSPrivatePartyResponse]
[docs]
@ioprepped
@dataclass
class BSPrivatePartyResponse(Response):
"""Here's that private party UI info you asked for, boss."""
success: Annotated[bool, IOAttrs('s')]
tokens: Annotated[int, IOAttrs('t')]
gold_pass: Annotated[bool, IOAttrs('g')]
datacode: Annotated[str | None, IOAttrs('d')]
[docs]
class BSClassicChestAppearance(Enum):
"""Appearances bombsquad classic chests can have."""
UNKNOWN = 'u'
DEFAULT = 'd'
[docs]
@ioprepped
@dataclass
class BSClassicAccountLiveData:
"""Account related data kept up to date live for classic app mode."""
[docs]
@dataclass
class Chest:
"""A lovely chest."""
appearance: Annotated[
BSClassicChestAppearance,
IOAttrs('a', enum_fallback=BSClassicChestAppearance.UNKNOWN),
]
unlock_time: Annotated[datetime.datetime, IOAttrs('t')]
ad_allow_time: Annotated[datetime.datetime | None, IOAttrs('at')]
[docs]
class LeagueType(Enum):
"""Type of league we are in."""
BRONZE = 'b'
SILVER = 's'
GOLD = 'g'
DIAMOND = 'd'
tickets: Annotated[int, IOAttrs('ti')]
tokens: Annotated[int, IOAttrs('to')]
gold_pass: Annotated[bool, IOAttrs('g')]
achievements: Annotated[int, IOAttrs('a')]
achievements_total: Annotated[int, IOAttrs('at')]
league_type: Annotated[LeagueType | None, IOAttrs('lt')]
league_num: Annotated[int | None, IOAttrs('ln')]
league_rank: Annotated[int | None, IOAttrs('lr')]
level: Annotated[int, IOAttrs('lv')]
xp: Annotated[int, IOAttrs('xp')]
xpmax: Annotated[int, IOAttrs('xpm')]
inbox_count: Annotated[int, IOAttrs('ibc')]
inbox_count_is_max: Annotated[bool, IOAttrs('ibcm')]
chests: Annotated[dict[str, Chest], IOAttrs('c')]
[docs]
class BSInboxEntryType(Enum):
"""Types of entries that can be in an inbox."""
UNKNOWN = 'u' # Entry types we don't support will be this.
SIMPLE = 's'
CLAIM = 'c'
CLAIM_DISCARD = 'cd'
[docs]
@ioprepped
@dataclass
class BSInboxEntry:
"""Single message in an inbox."""
type: Annotated[
BSInboxEntryType, IOAttrs('t', enum_fallback=BSInboxEntryType.UNKNOWN)
]
id: Annotated[str, IOAttrs('i')]
createtime: Annotated[datetime.datetime, IOAttrs('c')]
# If clients don't support format_version of a message they will
# display 'app needs to be updated to show this'.
format_version: Annotated[int, IOAttrs('f', soft_default=1)]
# These have soft defaults so can be removed in the future if desired.
message: Annotated[str, IOAttrs('m', soft_default='(invalid message)')]
subs: Annotated[list[str], IOAttrs('s', soft_default_factory=list)]
[docs]
@ioprepped
@dataclass
class BSInboxRequestMessage(Message):
"""Message requesting our inbox."""
[docs]
@override
@classmethod
def get_response_types(cls) -> list[type[Response] | None]:
return [BSInboxRequestResponse]
[docs]
@ioprepped
@dataclass
class BSInboxRequestResponse(Response):
"""Here's that inbox contents you asked for, boss."""
entries: Annotated[list[BSInboxEntry], IOAttrs('m')]
# Printable error if something goes wrong.
error: Annotated[str | None, IOAttrs('e')] = None
[docs]
@ioprepped
@dataclass
class BSChestInfoMessage(Message):
"""Request info about a chest."""
chest_id: Annotated[str, IOAttrs('i')]
[docs]
@override
@classmethod
def get_response_types(cls) -> list[type[Response] | None]:
return [BSChestInfoResponse]
[docs]
@ioprepped
@dataclass
class BSChestInfoResponse(Response):
"""Here's that inbox contents you asked for, boss."""
[docs]
@dataclass
class Chest:
"""A lovely chest."""
appearance: Annotated[
BSClassicChestAppearance,
IOAttrs('a', enum_fallback=BSClassicChestAppearance.UNKNOWN),
]
# How much to unlock *now*.
unlock_tokens: Annotated[int, IOAttrs('tk')]
# When unlocks on its own.
unlock_time: Annotated[datetime.datetime, IOAttrs('t')]
# Are ads allowed now?
ad_allow: Annotated[bool, IOAttrs('aa')]
chest: Annotated[Chest | None, IOAttrs('c')]
[docs]
@ioprepped
@dataclass
class BSChestActionMessage(Message):
"""Request action about a chest."""
[docs]
class Action(Enum):
"""Types of actions we can request."""
# Unlocking (for free or with tokens).
UNLOCK = 'u'
# Watched an ad to reduce wait.
AD = 'ad'
action: Annotated[Action, IOAttrs('a')]
# Tokens we are paying (only applies to unlock).
token_payment: Annotated[int, IOAttrs('t')]
chest_id: Annotated[str, IOAttrs('i')]
[docs]
@override
@classmethod
def get_response_types(cls) -> list[type[Response] | None]:
return [BSChestActionResponse]
[docs]
@ioprepped
@dataclass
class BSChestActionResponse(Response):
"""Here's the results of that action you asked for, boss."""
# If present, signifies the chest has been opened and we should show
# the user this stuff that was in it.
contents: Annotated[list[str] | None, IOAttrs('c')] = None
# Printable error if something goes wrong.
error: Annotated[str | None, IOAttrs('e')] = None
[docs]
class BSInboxEntryProcessType(Enum):
"""Types of processing we can ask for."""
POSITIVE = 'p'
NEGATIVE = 'n'
[docs]
@ioprepped
@dataclass
class BSInboxEntryProcessMessage(Message):
"""Do something to an inbox entry."""
id: Annotated[str, IOAttrs('i')]
process_type: Annotated[BSInboxEntryProcessType, IOAttrs('t')]
[docs]
@override
@classmethod
def get_response_types(cls) -> list[type[Response] | None]:
return [BSInboxEntryProcessResponse]
[docs]
@ioprepped
@dataclass
class BSInboxEntryProcessResponse(Response):
"""Did something to that inbox entry, boss."""
# Printable error if something goes wrong.
error: Annotated[str | None, IOAttrs('e')] = None