# Released under the MIT License. See LICENSE for details.
#
"""BombSquad specific bits."""
from __future__ import annotations
import datetime
from enum import Enum
from dataclasses import dataclass, field
from typing import Annotated, override, assert_never
from efro.util import pairs_to_flat
from efro.dataclassio import ioprepped, IOAttrs, IOMultiType
from efro.message import Message, Response
# Token counts for our various packs.
TOKENS1_COUNT = 50
TOKENS2_COUNT = 500
TOKENS3_COUNT = 1200
TOKENS4_COUNT = 2600
[docs]
@ioprepped
@dataclass
class PrivatePartyMessage(Message):
"""Message asking about info we need for private-party UI."""
need_datacode: bool
[docs]
@override
@classmethod
def get_response_types(cls) -> list[type[Response] | None]:
return [PrivatePartyResponse]
[docs]
@ioprepped
@dataclass
class PrivatePartyResponse(Response):
"""Here's that private party UI info you asked for, boss."""
success: bool
tokens: int
gold_pass: bool
datacode: str | None
[docs]
class ClassicChestAppearance(Enum):
"""Appearances bombsquad classic chests can have."""
UNKNOWN = 'u'
DEFAULT = 'd'
L1 = 'l1'
L2 = 'l2'
L3 = 'l3'
L4 = 'l4'
L5 = 'l5'
L6 = 'l6'
@property
def pretty_name(self) -> str:
"""Pretty name for the chest in English."""
# pylint: disable=too-many-return-statements
cls = type(self)
if self is cls.UNKNOWN:
return 'Unknown Chest'
if self is cls.DEFAULT:
return 'Chest'
if self is cls.L1:
return 'L1 Chest'
if self is cls.L2:
return 'L2 Chest'
if self is cls.L3:
return 'L3 Chest'
if self is cls.L4:
return 'L4 Chest'
if self is cls.L5:
return 'L5 Chest'
if self is cls.L6:
return 'L6 Chest'
assert_never(self)
[docs]
@ioprepped
@dataclass
class ClassicAccountLiveData:
"""Live account data fed to the client in the bs classic app mode."""
[docs]
@dataclass
class Chest:
"""A lovely chest."""
appearance: ClassicChestAppearance
unlock_time: datetime.datetime
ad_allow_time: datetime.datetime | None
[docs]
class LeagueType(Enum):
"""Type of league we are in."""
BRONZE = 'b'
SILVER = 's'
GOLD = 'g'
DIAMOND = 'd'
tickets: int
tokens: int
gold_pass: bool
remove_ads: bool
achievements: int
achievements_total: int
league_type: LeagueType | None
league_num: int | None
league_rank: int | None
level: int
xp: int
xpmax: int
inbox_count: int
inbox_count_is_max: bool
chests: dict[str, Chest]
[docs]
class DisplayItemTypeID(Enum):
"""Type ID for each of our subclasses."""
UNKNOWN = 'u'
TICKETS = 't'
TOKENS = 'k'
TEST = 's'
CHEST = 'c'
[docs]
class DisplayItem(IOMultiType[DisplayItemTypeID]):
"""Some amount of something that can be shown or described.
Used to depict chest contents or other rewards or prices.
"""
[docs]
@override
@classmethod
def get_type_id(cls) -> DisplayItemTypeID:
# Require child classes to supply this themselves. If we did a
# full type registry/lookup here it would require us to import
# everything and would prevent lazy loading.
raise NotImplementedError()
[docs]
@override
@classmethod
def get_type(cls, type_id: DisplayItemTypeID) -> type[DisplayItem]:
"""Return the subclass for each of our type-ids."""
# pylint: disable=cyclic-import
t = DisplayItemTypeID
if type_id is t.UNKNOWN:
return UnknownDisplayItem
if type_id is t.TICKETS:
return TicketsDisplayItem
if type_id is t.TOKENS:
return TokensDisplayItem
if type_id is t.TEST:
return TestDisplayItem
if type_id is t.CHEST:
return ChestDisplayItem
# Important to make sure we provide all types.
assert_never(type_id)
[docs]
def get_description(self) -> tuple[str, list[tuple[str, str]]]:
"""Return a string description and subs for the item.
These decriptions are baked into the DisplayItemWrapper and
should be accessed from there when available. This allows
clients to give descriptions even for newer display items they
don't recognize.
"""
raise NotImplementedError()
# Implement fallbacks so client can digest item lists even if they
# contain unrecognized stuff. DisplayItemWrapper contains basic
# baked down info that they can still use in such cases.
[docs]
@override
@classmethod
def get_unknown_type_fallback(cls) -> DisplayItem:
return UnknownDisplayItem()
[docs]
@ioprepped
@dataclass
class UnknownDisplayItem(DisplayItem):
"""Something we don't know how to display."""
[docs]
@override
@classmethod
def get_type_id(cls) -> DisplayItemTypeID:
return DisplayItemTypeID.UNKNOWN
[docs]
@override
def get_description(self) -> tuple[str, list[tuple[str, str]]]:
import logging
# Make noise but don't break.
logging.exception(
'UnknownDisplayItem.get_description() should never be called.'
' Always access descriptions on the DisplayItemWrapper.'
)
return 'Unknown', []
[docs]
@ioprepped
@dataclass
class TicketsDisplayItem(DisplayItem):
"""Some amount of tickets."""
count: int
[docs]
@override
@classmethod
def get_type_id(cls) -> DisplayItemTypeID:
return DisplayItemTypeID.TICKETS
[docs]
@override
def get_description(self) -> tuple[str, list[tuple[str, str]]]:
return '${C} Tickets', [('${C}', str(self.count))]
[docs]
@ioprepped
@dataclass
class TokensDisplayItem(DisplayItem):
"""Some amount of tokens."""
count: int
[docs]
@override
@classmethod
def get_type_id(cls) -> DisplayItemTypeID:
return DisplayItemTypeID.TOKENS
[docs]
@override
def get_description(self) -> tuple[str, list[tuple[str, str]]]:
return '${C} Tokens', [('${C}', str(self.count))]
[docs]
@ioprepped
@dataclass
class TestDisplayItem(DisplayItem):
"""Fills usable space for a display-item - good for calibration."""
[docs]
@override
@classmethod
def get_type_id(cls) -> DisplayItemTypeID:
return DisplayItemTypeID.TEST
[docs]
@override
def get_description(self) -> tuple[str, list[tuple[str, str]]]:
return 'Test Display Item Here', []
[docs]
@ioprepped
@dataclass
class ChestDisplayItem(DisplayItem):
"""Display a chest."""
appearance: ClassicChestAppearance
[docs]
@override
@classmethod
def get_type_id(cls) -> DisplayItemTypeID:
return DisplayItemTypeID.CHEST
[docs]
@override
def get_description(self) -> tuple[str, list[tuple[str, str]]]:
return self.appearance.pretty_name, []
[docs]
@ioprepped
@dataclass
class DisplayItemWrapper:
"""Wraps a DisplayItem and common info."""
item: DisplayItem
description: str
description_subs: list[str] | None
[docs]
@classmethod
def for_display_item(cls, item: DisplayItem) -> DisplayItemWrapper:
"""Convenience method to wrap a DisplayItem."""
desc, subs = item.get_description()
return DisplayItemWrapper(item, desc, pairs_to_flat(subs))
[docs]
@ioprepped
@dataclass
class ChestInfoMessage(Message):
"""Request info about a chest."""
chest_id: str
[docs]
@override
@classmethod
def get_response_types(cls) -> list[type[Response] | None]:
return [ChestInfoResponse]
[docs]
@ioprepped
@dataclass
class ChestInfoResponse(Response):
"""Here's that chest info you asked for, boss."""
[docs]
@dataclass
class Chest:
"""A lovely chest."""
[docs]
@dataclass
class PrizeSet:
"""A possible set of prizes for this chest."""
weight: float
contents: list[DisplayItemWrapper]
appearance: ClassicChestAppearance
# How much it costs to unlock *now*.
unlock_tokens: int
# When it unlocks on its own.
unlock_time: datetime.datetime
# Possible prizes we contain.
prizesets: list[PrizeSet]
# Are ads allowed now?
ad_allow: bool
chest: Chest | None
user_tokens: int | None
[docs]
class ClientUITypeID(Enum):
"""Type ID for each of our subclasses."""
UNKNOWN = 'u'
BASIC = 'b'
[docs]
class ClientUI(IOMultiType[ClientUITypeID]):
"""Defines some user interface on the client."""
[docs]
@override
@classmethod
def get_type_id(cls) -> ClientUITypeID:
# Require child classes to supply this themselves. If we did a
# full type registry/lookup here it would require us to import
# everything and would prevent lazy loading.
raise NotImplementedError()
[docs]
@override
@classmethod
def get_type(cls, type_id: ClientUITypeID) -> type[ClientUI]:
"""Return the subclass for each of our type-ids."""
# pylint: disable=cyclic-import
out: type[ClientUI]
t = ClientUITypeID
if type_id is t.UNKNOWN:
out = UnknownClientUI
elif type_id is t.BASIC:
out = BasicClientUI
else:
# Important to make sure we provide all types.
assert_never(type_id)
return out
[docs]
@override
@classmethod
def get_unknown_type_fallback(cls) -> ClientUI:
# If we encounter some future message type we don't know
# anything about, drop in a placeholder.
return UnknownClientUI()
[docs]
@ioprepped
@dataclass
class UnknownClientUI(ClientUI):
"""Fallback type for unrecognized entries."""
[docs]
@override
@classmethod
def get_type_id(cls) -> ClientUITypeID:
return ClientUITypeID.UNKNOWN
[docs]
class BasicClientUIComponentTypeID(Enum):
"""Type ID for each of our subclasses."""
UNKNOWN = 'u'
TEXT = 't'
LINK = 'l'
BS_CLASSIC_TOURNEY_RESULT = 'ct'
DISPLAY_ITEMS = 'di'
EXPIRE_TIME = 'd'
[docs]
class BasicClientUIComponent(IOMultiType[BasicClientUIComponentTypeID]):
"""Top level class for our multitype."""
[docs]
@override
@classmethod
def get_type_id(cls) -> BasicClientUIComponentTypeID:
# Require child classes to supply this themselves. If we did a
# full type registry/lookup here it would require us to import
# everything and would prevent lazy loading.
raise NotImplementedError()
[docs]
@override
@classmethod
def get_type(
cls, type_id: BasicClientUIComponentTypeID
) -> type[BasicClientUIComponent]:
"""Return the subclass for each of our type-ids."""
# pylint: disable=cyclic-import
t = BasicClientUIComponentTypeID
if type_id is t.UNKNOWN:
return BasicClientUIComponentUnknown
if type_id is t.TEXT:
return BasicClientUIComponentText
if type_id is t.LINK:
return BasicClientUIComponentLink
if type_id is t.BS_CLASSIC_TOURNEY_RESULT:
return BasicClientUIBsClassicTourneyResult
if type_id is t.DISPLAY_ITEMS:
return BasicClientUIDisplayItems
if type_id is t.EXPIRE_TIME:
return BasicClientUIExpireTime
# Important to make sure we provide all types.
assert_never(type_id)
[docs]
@override
@classmethod
def get_unknown_type_fallback(cls) -> BasicClientUIComponent:
# If we encounter some future message type we don't know
# anything about, drop in a placeholder.
return BasicClientUIComponentUnknown()
[docs]
@ioprepped
@dataclass
class BasicClientUIComponentUnknown(BasicClientUIComponent):
"""An unknown basic client component type.
In practice these should never show up since the master-server
generates these on the fly for the client and so should not send
clients one they can't digest.
"""
[docs]
@override
@classmethod
def get_type_id(cls) -> BasicClientUIComponentTypeID:
return BasicClientUIComponentTypeID.UNKNOWN
[docs]
@ioprepped
@dataclass
class BasicClientUIComponentText(BasicClientUIComponent):
"""Show some text in the inbox message."""
text: str
subs: list[str] = field(
default_factory=list
)
scale: float = 1.0
color: tuple[float, float, float, float] = (1.0, 1.0, 1.0, 1.0)
spacing_top: float = 0.0
spacing_bottom: float = 0.0
[docs]
@override
@classmethod
def get_type_id(cls) -> BasicClientUIComponentTypeID:
return BasicClientUIComponentTypeID.TEXT
[docs]
@ioprepped
@dataclass
class BasicClientUIComponentLink(BasicClientUIComponent):
"""Show a link in the inbox message."""
url: str
label: str
subs: list[str] = field(
default_factory=list
)
spacing_top: float = 0.0
spacing_bottom: float = 0.0
[docs]
@override
@classmethod
def get_type_id(cls) -> BasicClientUIComponentTypeID:
return BasicClientUIComponentTypeID.LINK
[docs]
@ioprepped
@dataclass
class BasicClientUIBsClassicTourneyResult(BasicClientUIComponent):
"""Show info about a classic tourney."""
tournament_id: str
game: str
players: int
rank: int
trophy: str | None
prizes: list[DisplayItemWrapper]
[docs]
@override
@classmethod
def get_type_id(cls) -> BasicClientUIComponentTypeID:
return BasicClientUIComponentTypeID.BS_CLASSIC_TOURNEY_RESULT
[docs]
@ioprepped
@dataclass
class BasicClientUIDisplayItems(BasicClientUIComponent):
"""Show some display-items."""
items: list[DisplayItemWrapper]
width: float = 100.0
spacing_top: float = 0.0
spacing_bottom: float = 0.0
[docs]
@override
@classmethod
def get_type_id(cls) -> BasicClientUIComponentTypeID:
return BasicClientUIComponentTypeID.DISPLAY_ITEMS
[docs]
@ioprepped
@dataclass
class BasicClientUIExpireTime(BasicClientUIComponent):
"""Show expire-time."""
time: datetime.datetime
spacing_top: float = 0.0
spacing_bottom: float = 0.0
[docs]
@override
@classmethod
def get_type_id(cls) -> BasicClientUIComponentTypeID:
return BasicClientUIComponentTypeID.EXPIRE_TIME
[docs]
@ioprepped
@dataclass
class BasicClientUI(ClientUI):
"""A basic UI for the client."""
[docs]
class InteractionStyle(Enum):
"""Overall interaction styles we support."""
UNKNOWN = 'u'
BUTTON_POSITIVE = 'p'
BUTTON_POSITIVE_NEGATIVE = 'pn'
components: list[BasicClientUIComponent]
interaction_style: InteractionStyle = InteractionStyle.BUTTON_POSITIVE
button_label_positive: ButtonLabel = ButtonLabel.OK
button_label_negative: ButtonLabel = ButtonLabel.CANCEL
[docs]
@override
@classmethod
def get_type_id(cls) -> ClientUITypeID:
return ClientUITypeID.BASIC
[docs]
def contains_unknown_elements(self) -> bool:
"""Whether something within us is an unknown type or enum."""
return (
self.interaction_style is self.InteractionStyle.UNKNOWN
or self.button_label_positive is self.ButtonLabel.UNKNOWN
or self.button_label_negative is self.ButtonLabel.UNKNOWN
or any(
c.get_type_id() is BasicClientUIComponentTypeID.UNKNOWN
for c in self.components
)
)
[docs]
@ioprepped
@dataclass
class ClientUIWrapper:
"""Wrapper for a ClientUI and its common data."""
id: str
createtime: datetime.datetime
ui: ClientUI
[docs]
@ioprepped
@dataclass
class InboxRequestMessage(Message):
"""Message requesting our inbox."""
[docs]
@override
@classmethod
def get_response_types(cls) -> list[type[Response] | None]:
return [InboxRequestResponse]
[docs]
@ioprepped
@dataclass
class InboxRequestResponse(Response):
"""Here's that inbox contents you asked for, boss."""
wrappers: list[ClientUIWrapper]
# Printable error if something goes wrong.
error: str | None = None
[docs]
class ClientUIAction(Enum):
"""Types of actions we can run."""
BUTTON_PRESS_POSITIVE = 'p'
BUTTON_PRESS_NEGATIVE = 'n'
[docs]
class ClientEffectTypeID(Enum):
"""Type ID for each of our subclasses."""
UNKNOWN = 'u'
SCREEN_MESSAGE = 'm'
SOUND = 's'
DELAY = 'd'
CHEST_WAIT_TIME_ANIMATION = 't'
TICKETS_ANIMATION = 'ta'
TOKENS_ANIMATION = 'toa'
[docs]
class ClientEffect(IOMultiType[ClientEffectTypeID]):
"""Something that can happen on the client.
This can include screen messages, sounds, visual effects, etc.
"""
[docs]
@override
@classmethod
def get_type_id(cls) -> ClientEffectTypeID:
# Require child classes to supply this themselves. If we did a
# full type registry/lookup here it would require us to import
# everything and would prevent lazy loading.
raise NotImplementedError()
[docs]
@override
@classmethod
def get_type(cls, type_id: ClientEffectTypeID) -> type[ClientEffect]:
"""Return the subclass for each of our type-ids."""
# pylint: disable=cyclic-import
# pylint: disable=too-many-return-statements
t = ClientEffectTypeID
if type_id is t.UNKNOWN:
return ClientEffectUnknown
if type_id is t.SCREEN_MESSAGE:
return ClientEffectScreenMessage
if type_id is t.SOUND:
return ClientEffectSound
if type_id is t.DELAY:
return ClientEffectDelay
if type_id is t.CHEST_WAIT_TIME_ANIMATION:
return ClientEffectChestWaitTimeAnimation
if type_id is t.TICKETS_ANIMATION:
return ClientEffectTicketsAnimation
if type_id is t.TOKENS_ANIMATION:
return ClientEffectTokensAnimation
# Important to make sure we provide all types.
assert_never(type_id)
[docs]
@override
@classmethod
def get_unknown_type_fallback(cls) -> ClientEffect:
# If we encounter some future message type we don't know
# anything about, drop in a placeholder.
return ClientEffectUnknown()
[docs]
@ioprepped
@dataclass
class ClientEffectUnknown(ClientEffect):
"""Fallback substitute for types we don't recognize."""
[docs]
@override
@classmethod
def get_type_id(cls) -> ClientEffectTypeID:
return ClientEffectTypeID.UNKNOWN
[docs]
@ioprepped
@dataclass
class ClientEffectScreenMessage(ClientEffect):
"""Display a screen-message."""
message: str
subs: list[str]
color: tuple[float, float, float] = (1.0, 1.0, 1.0)
[docs]
@override
@classmethod
def get_type_id(cls) -> ClientEffectTypeID:
return ClientEffectTypeID.SCREEN_MESSAGE
[docs]
@ioprepped
@dataclass
class ClientEffectSound(ClientEffect):
"""Play a sound."""
[docs]
class Sound(Enum):
"""Sounds that can be made alongside the message."""
UNKNOWN = 'u'
CASH_REGISTER = 'c'
ERROR = 'e'
POWER_DOWN = 'p'
GUN_COCKING = 'g'
sound: Sound
volume: float = 1.0
[docs]
@override
@classmethod
def get_type_id(cls) -> ClientEffectTypeID:
return ClientEffectTypeID.SOUND
[docs]
@ioprepped
@dataclass
class ClientEffectChestWaitTimeAnimation(ClientEffect):
"""Animate chest wait time changing."""
chestid: str
duration: float
startvalue: datetime.datetime
endvalue: datetime.datetime
[docs]
@override
@classmethod
def get_type_id(cls) -> ClientEffectTypeID:
return ClientEffectTypeID.CHEST_WAIT_TIME_ANIMATION
[docs]
@ioprepped
@dataclass
class ClientEffectTicketsAnimation(ClientEffect):
"""Animate tickets count."""
duration: float
startvalue: int
endvalue: int
[docs]
@override
@classmethod
def get_type_id(cls) -> ClientEffectTypeID:
return ClientEffectTypeID.TICKETS_ANIMATION
[docs]
@ioprepped
@dataclass
class ClientEffectTokensAnimation(ClientEffect):
"""Animate tokens count."""
duration: float
startvalue: int
endvalue: int
[docs]
@override
@classmethod
def get_type_id(cls) -> ClientEffectTypeID:
return ClientEffectTypeID.TOKENS_ANIMATION
[docs]
@ioprepped
@dataclass
class ClientEffectDelay(ClientEffect):
"""Delay effect processing."""
seconds: float
[docs]
@override
@classmethod
def get_type_id(cls) -> ClientEffectTypeID:
return ClientEffectTypeID.DELAY
[docs]
@ioprepped
@dataclass
class ClientUIActionMessage(Message):
"""Do something to a client ui."""
id: str
action: ClientUIAction
[docs]
@override
@classmethod
def get_response_types(cls) -> list[type[Response] | None]:
return [ClientUIActionResponse]
[docs]
@ioprepped
@dataclass
class ClientUIActionResponse(Response):
"""Did something to that inbox entry, boss."""
[docs]
class ErrorType(Enum):
"""Types of errors that may have occurred."""
# Probably a future error type we don't recognize.
UNKNOWN = 'u'
# Something went wrong on the server, but specifics are not
# relevant.
INTERNAL = 'i'
# The entry expired on the server. In various cases such as 'ok'
# buttons this can generally be ignored.
EXPIRED = 'e'
error_type: ErrorType | None
# User facing error message in the case of errors.
error_message: str | None
effects: list[ClientEffect]
[docs]
@ioprepped
@dataclass
class ScoreSubmitMessage(Message):
"""Let the server know we got some score in something."""
score_token: str
[docs]
@override
@classmethod
def get_response_types(cls) -> list[type[Response] | None]:
return [ScoreSubmitResponse]
[docs]
@ioprepped
@dataclass
class ScoreSubmitResponse(Response):
"""Did something to that inbox entry, boss."""
# Things we should show on our end.
effects: list[ClientEffect]
[docs]
@ioprepped
@dataclass
class ChestActionMessage(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: Action
# Tokens we are paying (only applies to unlock).
token_payment: int
chest_id: str
[docs]
@override
@classmethod
def get_response_types(cls) -> list[type[Response] | None]:
return [ChestActionResponse]
[docs]
@ioprepped
@dataclass
class ChestActionResponse(Response):
"""Here's the results of that action you asked for, boss."""
# Tokens that were actually charged.
tokens_charged: int = 0
# If present, signifies the chest has been opened and we should show
# the user this stuff that was in it.
contents: list[DisplayItemWrapper] | None = None
# If contents are present, which of the chest's prize-sets they
# represent.
prizeindex: int = 0
# Printable error if something goes wrong.
error: str | None = None
# Printable warning. Shown in orange with an error sound. Does not
# mean the action failed; only that there's something to tell the
# users such as 'It looks like you are faking ad views; stop it or
# you won't have ad options anymore.'
warning: str | None = None
# Printable success message. Shown in green with a cash-register
# sound. Can be used for things like successful wait reductions via
# ad views. Used in builds earlier than 22311; can remove once
# 22311+ is ubiquitous.
success_msg: str | None = None
# Effects to show on the client. Replaces warning and success_msg in
# build 22311 or newer.
effects: list[ClientEffect] = field(default_factory=list)
# Docs-generation hack; import some stuff that we likely only forward-declared
# in our actual source code so that docs tools can find it.
from typing import (Coroutine, Any, Literal, Callable,
Generator, Awaitable, Sequence, Self)
import asyncio
from concurrent.futures import Future