baplus

 26class CloudSubsystem(babase.AppSubsystem):
 27    """Manages communication with cloud components."""
 28
 29    def __init__(self) -> None:
 30        super().__init__()
 31        self.on_connectivity_changed_callbacks: CallbackSet[
 32            Callable[[bool], None]
 33        ] = CallbackSet()
 34
 35    @property
 36    def connected(self) -> bool:
 37        """Property equivalent of CloudSubsystem.is_connected()."""
 38        return self.is_connected()
 39
 40    def is_connected(self) -> bool:
 41        """Return whether a connection to the cloud is present.
 42
 43        This is a good indicator (though not for certain) that sending
 44        messages will succeed.
 45        """
 46        return False  # Needs to be overridden
 47
 48    def on_connectivity_changed(self, connected: bool) -> None:
 49        """Called when cloud connectivity state changes."""
 50        babase.balog.debug('Connectivity is now %s.', connected)
 51
 52        plus = babase.app.plus
 53        assert plus is not None
 54
 55        # Fire any registered callbacks for this.
 56        for call in self.on_connectivity_changed_callbacks.getcalls():
 57            try:
 58                call(connected)
 59            except Exception:
 60                logging.exception('Error in connectivity-changed callback.')
 61
 62    @overload
 63    def send_message_cb(
 64        self,
 65        msg: bacommon.cloud.LoginProxyRequestMessage,
 66        on_response: Callable[
 67            [bacommon.cloud.LoginProxyRequestResponse | Exception], None
 68        ],
 69    ) -> None: ...
 70
 71    @overload
 72    def send_message_cb(
 73        self,
 74        msg: bacommon.cloud.LoginProxyStateQueryMessage,
 75        on_response: Callable[
 76            [bacommon.cloud.LoginProxyStateQueryResponse | Exception], None
 77        ],
 78    ) -> None: ...
 79
 80    @overload
 81    def send_message_cb(
 82        self,
 83        msg: bacommon.cloud.LoginProxyCompleteMessage,
 84        on_response: Callable[[None | Exception], None],
 85    ) -> None: ...
 86
 87    @overload
 88    def send_message_cb(
 89        self,
 90        msg: bacommon.cloud.PingMessage,
 91        on_response: Callable[[bacommon.cloud.PingResponse | Exception], None],
 92    ) -> None: ...
 93
 94    @overload
 95    def send_message_cb(
 96        self,
 97        msg: bacommon.cloud.SignInMessage,
 98        on_response: Callable[
 99            [bacommon.cloud.SignInResponse | Exception], None
100        ],
101    ) -> None: ...
102
103    @overload
104    def send_message_cb(
105        self,
106        msg: bacommon.cloud.ManageAccountMessage,
107        on_response: Callable[
108            [bacommon.cloud.ManageAccountResponse | Exception], None
109        ],
110    ) -> None: ...
111
112    @overload
113    def send_message_cb(
114        self,
115        msg: bacommon.cloud.StoreQueryMessage,
116        on_response: Callable[
117            [bacommon.cloud.StoreQueryResponse | Exception], None
118        ],
119    ) -> None: ...
120
121    @overload
122    def send_message_cb(
123        self,
124        msg: bacommon.cloud.BSPrivatePartyMessage,
125        on_response: Callable[
126            [bacommon.cloud.BSPrivatePartyResponse | Exception], None
127        ],
128    ) -> None: ...
129
130    @overload
131    def send_message_cb(
132        self,
133        msg: bacommon.cloud.BSInboxRequestMessage,
134        on_response: Callable[
135            [bacommon.cloud.BSInboxRequestResponse | Exception], None
136        ],
137    ) -> None: ...
138
139    @overload
140    def send_message_cb(
141        self,
142        msg: bacommon.cloud.BSInboxEntryProcessMessage,
143        on_response: Callable[
144            [bacommon.cloud.BSInboxEntryProcessResponse | Exception], None
145        ],
146    ) -> None: ...
147
148    @overload
149    def send_message_cb(
150        self,
151        msg: bacommon.cloud.BSChestInfoMessage,
152        on_response: Callable[
153            [bacommon.cloud.BSChestInfoResponse | Exception], None
154        ],
155    ) -> None: ...
156
157    @overload
158    def send_message_cb(
159        self,
160        msg: bacommon.cloud.BSChestActionMessage,
161        on_response: Callable[
162            [bacommon.cloud.BSChestActionResponse | Exception], None
163        ],
164    ) -> None: ...
165
166    def send_message_cb(
167        self,
168        msg: Message,
169        on_response: Callable[[Any], None],
170    ) -> None:
171        """Asynchronously send a message to the cloud from the logic thread.
172
173        The provided on_response call will be run in the logic thread
174        and passed either the response or the error that occurred.
175        """
176        raise NotImplementedError(
177            'Cloud functionality is not present in this build.'
178        )
179
180    @overload
181    def send_message(
182        self, msg: bacommon.cloud.WorkspaceFetchMessage
183    ) -> bacommon.cloud.WorkspaceFetchResponse: ...
184
185    @overload
186    def send_message(
187        self, msg: bacommon.cloud.MerchAvailabilityMessage
188    ) -> bacommon.cloud.MerchAvailabilityResponse: ...
189
190    @overload
191    def send_message(
192        self, msg: bacommon.cloud.TestMessage
193    ) -> bacommon.cloud.TestResponse: ...
194
195    def send_message(self, msg: Message) -> Response | None:
196        """Synchronously send a message to the cloud.
197
198        Must be called from a background thread.
199        """
200        raise NotImplementedError(
201            'Cloud functionality is not present in this build.'
202        )
203
204    @overload
205    async def send_message_async(
206        self, msg: bacommon.cloud.SendInfoMessage
207    ) -> bacommon.cloud.SendInfoResponse: ...
208
209    @overload
210    async def send_message_async(
211        self, msg: bacommon.cloud.TestMessage
212    ) -> bacommon.cloud.TestResponse: ...
213
214    async def send_message_async(self, msg: Message) -> Response | None:
215        """Synchronously send a message to the cloud.
216
217        Must be called from the logic thread.
218        """
219        raise NotImplementedError(
220            'Cloud functionality is not present in this build.'
221        )
222
223    def subscribe_test(
224        self, updatecall: Callable[[int | None], None]
225    ) -> babase.CloudSubscription:
226        """Subscribe to some test data."""
227        raise NotImplementedError(
228            'Cloud functionality is not present in this build.'
229        )
230
231    def subscribe_classic_account_data(
232        self,
233        updatecall: Callable[[bacommon.cloud.BSClassicAccountLiveData], None],
234    ) -> babase.CloudSubscription:
235        """Subscribe to classic account data."""
236        raise NotImplementedError(
237            'Cloud functionality is not present in this build.'
238        )
239
240    def unsubscribe(self, subscription_id: int) -> None:
241        """Unsubscribe from some subscription.
242
243        Do not call this manually; it is called by CloudSubscription.
244        """
245        raise NotImplementedError(
246            'Cloud functionality is not present in this build.'
247        )

 21class PlusAppSubsystem(AppSubsystem):
 22    """Subsystem for plus functionality in the app.
 23
 24    The single shared instance of this app can be accessed at
 25    babase.app.plus. Note that it is possible for this to be None if the
 26    plus package is not present, and code should handle that case
 27    gracefully.
 28    """
 29
 30    # pylint: disable=too-many-public-methods
 31
 32    # Note: this is basically just a wrapper around _baplus for
 33    # type-checking purposes. Maybe there's some smart way we could skip
 34    # the overhead of this wrapper at runtime.
 35
 36    accounts: AccountV2Subsystem
 37    cloud: CloudSubsystem
 38
 39    @override
 40    def on_app_loading(self) -> None:
 41        _baplus.on_app_loading()
 42        self.accounts.on_app_loading()
 43
 44    @staticmethod
 45    def add_v1_account_transaction(
 46        transaction: dict, callback: Callable | None = None
 47    ) -> None:
 48        """(internal)"""
 49        return _baplus.add_v1_account_transaction(transaction, callback)
 50
 51    @staticmethod
 52    def game_service_has_leaderboard(game: str, config: str) -> bool:
 53        """(internal)
 54
 55        Given a game and config string, returns whether there is a leaderboard
 56        for it on the game service.
 57        """
 58        return _baplus.game_service_has_leaderboard(game, config)
 59
 60    @staticmethod
 61    def get_master_server_address(source: int = -1, version: int = 1) -> str:
 62        """(internal)
 63
 64        Return the address of the master server.
 65        """
 66        return _baplus.get_master_server_address(source, version)
 67
 68    @staticmethod
 69    def get_classic_news_show() -> str:
 70        """(internal)"""
 71        return _baplus.get_classic_news_show()
 72
 73    @staticmethod
 74    def get_price(item: str) -> str | None:
 75        """(internal)"""
 76        return _baplus.get_price(item)
 77
 78    @staticmethod
 79    def get_v1_account_product_purchased(item: str) -> bool:
 80        """(internal)"""
 81        return _baplus.get_v1_account_product_purchased(item)
 82
 83    @staticmethod
 84    def get_v1_account_product_purchases_state() -> int:
 85        """(internal)"""
 86        return _baplus.get_v1_account_product_purchases_state()
 87
 88    @staticmethod
 89    def get_v1_account_display_string(full: bool = True) -> str:
 90        """(internal)"""
 91        return _baplus.get_v1_account_display_string(full)
 92
 93    @staticmethod
 94    def get_v1_account_misc_read_val(name: str, default_value: Any) -> Any:
 95        """(internal)"""
 96        return _baplus.get_v1_account_misc_read_val(name, default_value)
 97
 98    @staticmethod
 99    def get_v1_account_misc_read_val_2(name: str, default_value: Any) -> Any:
100        """(internal)"""
101        return _baplus.get_v1_account_misc_read_val_2(name, default_value)
102
103    @staticmethod
104    def get_v1_account_misc_val(name: str, default_value: Any) -> Any:
105        """(internal)"""
106        return _baplus.get_v1_account_misc_val(name, default_value)
107
108    @staticmethod
109    def get_v1_account_name() -> str:
110        """(internal)"""
111        return _baplus.get_v1_account_name()
112
113    @staticmethod
114    def get_v1_account_public_login_id() -> str | None:
115        """(internal)"""
116        return _baplus.get_v1_account_public_login_id()
117
118    @staticmethod
119    def get_v1_account_state() -> str:
120        """(internal)"""
121        return _baplus.get_v1_account_state()
122
123    @staticmethod
124    def get_v1_account_state_num() -> int:
125        """(internal)"""
126        return _baplus.get_v1_account_state_num()
127
128    @staticmethod
129    def get_v1_account_ticket_count() -> int:
130        """(internal)
131
132        Return the number of tickets for the current account.
133        """
134        return _baplus.get_v1_account_ticket_count()
135
136    @staticmethod
137    def get_v1_account_type() -> str:
138        """(internal)"""
139        return _baplus.get_v1_account_type()
140
141    @staticmethod
142    def get_v2_fleet() -> str:
143        """(internal)"""
144        return _baplus.get_v2_fleet()
145
146    @staticmethod
147    def have_outstanding_v1_account_transactions() -> bool:
148        """(internal)"""
149        return _baplus.have_outstanding_v1_account_transactions()
150
151    @staticmethod
152    def in_game_purchase(item: str, price: int) -> None:
153        """(internal)"""
154        return _baplus.in_game_purchase(item, price)
155
156    @staticmethod
157    def is_blessed() -> bool:
158        """(internal)"""
159        return _baplus.is_blessed()
160
161    @staticmethod
162    def mark_config_dirty() -> None:
163        """(internal)
164
165        Category: General Utility Functions
166        """
167        return _baplus.mark_config_dirty()
168
169    @staticmethod
170    def power_ranking_query(callback: Callable, season: Any = None) -> None:
171        """(internal)"""
172        return _baplus.power_ranking_query(callback, season)
173
174    @staticmethod
175    def purchase(item: str) -> None:
176        """(internal)"""
177        return _baplus.purchase(item)
178
179    @staticmethod
180    def report_achievement(
181        achievement: str, pass_to_account: bool = True
182    ) -> None:
183        """(internal)"""
184        return _baplus.report_achievement(achievement, pass_to_account)
185
186    @staticmethod
187    def reset_achievements() -> None:
188        """(internal)"""
189        return _baplus.reset_achievements()
190
191    @staticmethod
192    def restore_purchases() -> None:
193        """(internal)"""
194        return _baplus.restore_purchases()
195
196    @staticmethod
197    def run_v1_account_transactions() -> None:
198        """(internal)"""
199        return _baplus.run_v1_account_transactions()
200
201    @staticmethod
202    def sign_in_v1(account_type: str) -> None:
203        """(internal)
204
205        Category: General Utility Functions
206        """
207        return _baplus.sign_in_v1(account_type)
208
209    @staticmethod
210    def sign_out_v1(v2_embedded: bool = False) -> None:
211        """(internal)
212
213        Category: General Utility Functions
214        """
215        return _baplus.sign_out_v1(v2_embedded)
216
217    @staticmethod
218    def submit_score(
219        game: str,
220        config: str,
221        name: Any,
222        score: int | None,
223        callback: Callable,
224        *,
225        order: str = 'increasing',
226        tournament_id: str | None = None,
227        score_type: str = 'points',
228        campaign: str | None = None,
229        level: str | None = None,
230    ) -> None:
231        """(internal)
232
233        Submit a score to the server; callback will be called with the results.
234        As a courtesy, please don't send fake scores to the server. I'd prefer
235        to devote my time to improving the game instead of trying to make the
236        score server more mischief-proof.
237        """
238        return _baplus.submit_score(
239            game,
240            config,
241            name,
242            score,
243            callback,
244            order,
245            tournament_id,
246            score_type,
247            campaign,
248            level,
249        )
250
251    @staticmethod
252    def tournament_query(
253        callback: Callable[[dict | None], None], args: dict
254    ) -> None:
255        """(internal)"""
256        return _baplus.tournament_query(callback, args)
257
258    @staticmethod
259    def supports_purchases() -> bool:
260        """Does this platform support in-app-purchases?"""
261        return _baplus.supports_purchases()
262
263    @staticmethod
264    def have_incentivized_ad() -> bool:
265        """Is an incentivized ad available?"""
266        return _baplus.have_incentivized_ad()
267
268    @staticmethod
269    def has_video_ads() -> bool:
270        """Are video ads available?"""
271        return _baplus.has_video_ads()
272
273    @staticmethod
274    def can_show_ad() -> bool:
275        """Can we show an ad?"""
276        return _baplus.can_show_ad()
277
278    @staticmethod
279    def show_ad(
280        purpose: str, on_completion_call: Callable[[], None] | None = None
281    ) -> None:
282        """Show an ad."""
283        _baplus.show_ad(purpose, on_completion_call)
284
285    @staticmethod
286    def show_ad_2(
287        purpose: str, on_completion_call: Callable[[bool], None] | None = None
288    ) -> None:
289        """Show an ad."""
290        _baplus.show_ad_2(purpose, on_completion_call)
291
292    @staticmethod
293    def show_game_service_ui(
294        show: str = 'general',
295        game: str | None = None,
296        game_version: str | None = None,
297    ) -> None:
298        """Show game-service provided UI."""
299        _baplus.show_game_service_ui(show, game, game_version)