baplus

 27class CloudSubsystem(babase.AppSubsystem):
 28    """Manages communication with cloud components."""
 29
 30    @property
 31    def connected(self) -> bool:
 32        """Property equivalent of CloudSubsystem.is_connected()."""
 33        return self.is_connected()
 34
 35    def is_connected(self) -> bool:
 36        """Return whether a connection to the cloud is present.
 37
 38        This is a good indicator (though not for certain) that sending
 39        messages will succeed.
 40        """
 41        return False  # Needs to be overridden
 42
 43    def on_connectivity_changed(self, connected: bool) -> None:
 44        """Called when cloud connectivity state changes."""
 45        logger.debug('Connectivity is now %s.', connected)
 46
 47        plus = babase.app.plus
 48        assert plus is not None
 49
 50        # Inform things that use this.
 51        # (TODO: should generalize this into some sort of registration system)
 52        plus.accounts.on_cloud_connectivity_changed(connected)
 53
 54    @overload
 55    def send_message_cb(
 56        self,
 57        msg: bacommon.cloud.LoginProxyRequestMessage,
 58        on_response: Callable[
 59            [bacommon.cloud.LoginProxyRequestResponse | Exception], None
 60        ],
 61    ) -> None: ...
 62
 63    @overload
 64    def send_message_cb(
 65        self,
 66        msg: bacommon.cloud.LoginProxyStateQueryMessage,
 67        on_response: Callable[
 68            [bacommon.cloud.LoginProxyStateQueryResponse | Exception], None
 69        ],
 70    ) -> None: ...
 71
 72    @overload
 73    def send_message_cb(
 74        self,
 75        msg: bacommon.cloud.LoginProxyCompleteMessage,
 76        on_response: Callable[[None | Exception], None],
 77    ) -> None: ...
 78
 79    @overload
 80    def send_message_cb(
 81        self,
 82        msg: bacommon.cloud.PingMessage,
 83        on_response: Callable[[bacommon.cloud.PingResponse | Exception], None],
 84    ) -> None: ...
 85
 86    @overload
 87    def send_message_cb(
 88        self,
 89        msg: bacommon.cloud.SignInMessage,
 90        on_response: Callable[
 91            [bacommon.cloud.SignInResponse | Exception], None
 92        ],
 93    ) -> None: ...
 94
 95    @overload
 96    def send_message_cb(
 97        self,
 98        msg: bacommon.cloud.ManageAccountMessage,
 99        on_response: Callable[
100            [bacommon.cloud.ManageAccountResponse | Exception], None
101        ],
102    ) -> None: ...
103
104    @overload
105    def send_message_cb(
106        self,
107        msg: bacommon.cloud.StoreQueryMessage,
108        on_response: Callable[
109            [bacommon.cloud.StoreQueryResponse | Exception], None
110        ],
111    ) -> None: ...
112
113    @overload
114    def send_message_cb(
115        self,
116        msg: bacommon.cloud.BSPrivatePartyMessage,
117        on_response: Callable[
118            [bacommon.cloud.BSPrivatePartyResponse | Exception], None
119        ],
120    ) -> None: ...
121
122    def send_message_cb(
123        self,
124        msg: Message,
125        on_response: Callable[[Any], None],
126    ) -> None:
127        """Asynchronously send a message to the cloud from the logic thread.
128
129        The provided on_response call will be run in the logic thread
130        and passed either the response or the error that occurred.
131        """
132        raise NotImplementedError(
133            'Cloud functionality is not present in this build.'
134        )
135
136    @overload
137    def send_message(
138        self, msg: bacommon.cloud.WorkspaceFetchMessage
139    ) -> bacommon.cloud.WorkspaceFetchResponse: ...
140
141    @overload
142    def send_message(
143        self, msg: bacommon.cloud.MerchAvailabilityMessage
144    ) -> bacommon.cloud.MerchAvailabilityResponse: ...
145
146    @overload
147    def send_message(
148        self, msg: bacommon.cloud.TestMessage
149    ) -> bacommon.cloud.TestResponse: ...
150
151    def send_message(self, msg: Message) -> Response | None:
152        """Synchronously send a message to the cloud.
153
154        Must be called from a background thread.
155        """
156        raise NotImplementedError(
157            'Cloud functionality is not present in this build.'
158        )
159
160    @overload
161    async def send_message_async(
162        self, msg: bacommon.cloud.SendInfoMessage
163    ) -> bacommon.cloud.SendInfoResponse: ...
164
165    @overload
166    async def send_message_async(
167        self, msg: bacommon.cloud.TestMessage
168    ) -> bacommon.cloud.TestResponse: ...
169
170    async def send_message_async(self, msg: Message) -> Response | None:
171        """Synchronously send a message to the cloud.
172
173        Must be called from the logic thread.
174        """
175        raise NotImplementedError(
176            'Cloud functionality is not present in this build.'
177        )
178
179    def subscribe_test(
180        self, updatecall: Callable[[int | None], None]
181    ) -> babase.CloudSubscription:
182        """Subscribe to some data."""
183        from bacommon.cloud import TestCloudSubscriptionRequest
184
185        assert babase.in_logic_thread()
186
187        return self._subscribe(
188            TestCloudSubscriptionRequest(),
189            partial(self._on_sub_value_test, updatecall),
190        )
191
192    @staticmethod
193    def _on_sub_value_test(
194        cb: Callable[[int | None], None],
195        invalue: bacommon.cloud.CloudSubscriptionValue,
196    ) -> None:
197        from bacommon.cloud import TestCloudSubscriptionValue
198
199        assert babase.in_logic_thread()
200
201        # Make sure we got the type we expected for this sub and pass it
202        # to the user callback.
203        assert isinstance(invalue, TestCloudSubscriptionValue)
204        cb(invalue.value)
205
206    def _subscribe(
207        self,
208        request: bacommon.cloud.CloudSubscriptionRequest,
209        updatecall: Callable[[Any], None],
210    ) -> babase.CloudSubscription:
211        """Subscribe to some cloud data."""
212        raise NotImplementedError(
213            'Cloud functionality is not present in this build.'
214        )
215
216    def unsubscribe(self, subscription_id: int) -> None:
217        """Unsubscribe from some subscription.
218
219        Do not call this manually; it is called by CloudSubscription.
220        """
221        raise NotImplementedError(
222            'Cloud functionality is not present in this build.'
223        )

 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)