baplus

 26class CloudSubsystem(babase.AppSubsystem):
 27    """Manages communication with cloud components."""
 28
 29    @property
 30    def connected(self) -> bool:
 31        """Property equivalent of CloudSubsystem.is_connected()."""
 32        return self.is_connected()
 33
 34    def is_connected(self) -> bool:
 35        """Return whether a connection to the cloud is present.
 36
 37        This is a good indicator (though not for certain) that sending
 38        messages will succeed.
 39        """
 40        return False  # Needs to be overridden
 41
 42    def on_connectivity_changed(self, connected: bool) -> None:
 43        """Called when cloud connectivity state changes."""
 44        if DEBUG_LOG:
 45            logging.debug('CloudSubsystem: 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
133        del msg  # Unused.
134
135        babase.pushcall(
136            babase.Call(
137                on_response,
138                RuntimeError('Cloud functionality is not available.'),
139            )
140        )
141
142    @overload
143    def send_message(
144        self, msg: bacommon.cloud.WorkspaceFetchMessage
145    ) -> bacommon.cloud.WorkspaceFetchResponse: ...
146
147    @overload
148    def send_message(
149        self, msg: bacommon.cloud.MerchAvailabilityMessage
150    ) -> bacommon.cloud.MerchAvailabilityResponse: ...
151
152    @overload
153    def send_message(
154        self, msg: bacommon.cloud.TestMessage
155    ) -> bacommon.cloud.TestResponse: ...
156
157    def send_message(self, msg: Message) -> Response | None:
158        """Synchronously send a message to the cloud.
159
160        Must be called from a background thread.
161        """
162        raise RuntimeError('Cloud functionality is not available.')
163
164    @overload
165    async def send_message_async(
166        self, msg: bacommon.cloud.SendInfoMessage
167    ) -> bacommon.cloud.SendInfoResponse: ...
168
169    @overload
170    async def send_message_async(
171        self, msg: bacommon.cloud.TestMessage
172    ) -> bacommon.cloud.TestResponse: ...
173
174    async def send_message_async(self, msg: Message) -> Response | None:
175        """Synchronously send a message to the cloud.
176
177        Must be called from the logic thread.
178        """
179        raise RuntimeError('Cloud functionality is not available.')

 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    # noinspection PyUnresolvedReferences
 45    @staticmethod
 46    def add_v1_account_transaction(
 47        transaction: dict, callback: Callable | None = None
 48    ) -> None:
 49        """(internal)"""
 50        return _baplus.add_v1_account_transaction(transaction, callback)
 51
 52    @staticmethod
 53    def game_service_has_leaderboard(game: str, config: str) -> bool:
 54        """(internal)
 55
 56        Given a game and config string, returns whether there is a leaderboard
 57        for it on the game service.
 58        """
 59        return _baplus.game_service_has_leaderboard(game, config)
 60
 61    @staticmethod
 62    def get_master_server_address(source: int = -1, version: int = 1) -> str:
 63        """(internal)
 64
 65        Return the address of the master server.
 66        """
 67        return _baplus.get_master_server_address(source, version)
 68
 69    @staticmethod
 70    def get_news_show() -> str:
 71        """(internal)"""
 72        return _baplus.get_news_show()
 73
 74    @staticmethod
 75    def get_price(item: str) -> str | None:
 76        """(internal)"""
 77        return _baplus.get_price(item)
 78
 79    @staticmethod
 80    def get_purchased(item: str) -> bool:
 81        """(internal)"""
 82        return _baplus.get_purchased(item)
 83
 84    @staticmethod
 85    def get_purchases_state() -> int:
 86        """(internal)"""
 87        return _baplus.get_purchases_state()
 88
 89    @staticmethod
 90    def get_v1_account_display_string(full: bool = True) -> str:
 91        """(internal)"""
 92        return _baplus.get_v1_account_display_string(full)
 93
 94    @staticmethod
 95    def get_v1_account_misc_read_val(name: str, default_value: Any) -> Any:
 96        """(internal)"""
 97        return _baplus.get_v1_account_misc_read_val(name, default_value)
 98
 99    @staticmethod
100    def get_v1_account_misc_read_val_2(name: str, default_value: Any) -> Any:
101        """(internal)"""
102        return _baplus.get_v1_account_misc_read_val_2(name, default_value)
103
104    @staticmethod
105    def get_v1_account_misc_val(name: str, default_value: Any) -> Any:
106        """(internal)"""
107        return _baplus.get_v1_account_misc_val(name, default_value)
108
109    @staticmethod
110    def get_v1_account_name() -> str:
111        """(internal)"""
112        return _baplus.get_v1_account_name()
113
114    @staticmethod
115    def get_v1_account_public_login_id() -> str | None:
116        """(internal)"""
117        return _baplus.get_v1_account_public_login_id()
118
119    @staticmethod
120    def get_v1_account_state() -> str:
121        """(internal)"""
122        return _baplus.get_v1_account_state()
123
124    @staticmethod
125    def get_v1_account_state_num() -> int:
126        """(internal)"""
127        return _baplus.get_v1_account_state_num()
128
129    @staticmethod
130    def get_v1_account_ticket_count() -> int:
131        """(internal)
132
133        Returns the number of tickets for the current account.
134        """
135        return _baplus.get_v1_account_ticket_count()
136
137    @staticmethod
138    def get_v1_account_type() -> str:
139        """(internal)"""
140        return _baplus.get_v1_account_type()
141
142    @staticmethod
143    def get_v2_fleet() -> str:
144        """(internal)"""
145        return _baplus.get_v2_fleet()
146
147    @staticmethod
148    def have_outstanding_v1_account_transactions() -> bool:
149        """(internal)"""
150        return _baplus.have_outstanding_v1_account_transactions()
151
152    @staticmethod
153    def in_game_purchase(item: str, price: int) -> None:
154        """(internal)"""
155        return _baplus.in_game_purchase(item, price)
156
157    @staticmethod
158    def is_blessed() -> bool:
159        """(internal)"""
160        return _baplus.is_blessed()
161
162    @staticmethod
163    def mark_config_dirty() -> None:
164        """(internal)
165
166        Category: General Utility Functions
167        """
168        return _baplus.mark_config_dirty()
169
170    @staticmethod
171    def power_ranking_query(callback: Callable, season: Any = None) -> None:
172        """(internal)"""
173        return _baplus.power_ranking_query(callback, season)
174
175    @staticmethod
176    def purchase(item: str) -> None:
177        """(internal)"""
178        return _baplus.purchase(item)
179
180    @staticmethod
181    def report_achievement(
182        achievement: str, pass_to_account: bool = True
183    ) -> None:
184        """(internal)"""
185        return _baplus.report_achievement(achievement, pass_to_account)
186
187    @staticmethod
188    def reset_achievements() -> None:
189        """(internal)"""
190        return _baplus.reset_achievements()
191
192    @staticmethod
193    def restore_purchases() -> None:
194        """(internal)"""
195        return _baplus.restore_purchases()
196
197    @staticmethod
198    def run_v1_account_transactions() -> None:
199        """(internal)"""
200        return _baplus.run_v1_account_transactions()
201
202    @staticmethod
203    def sign_in_v1(account_type: str) -> None:
204        """(internal)
205
206        Category: General Utility Functions
207        """
208        return _baplus.sign_in_v1(account_type)
209
210    @staticmethod
211    def sign_out_v1(v2_embedded: bool = False) -> None:
212        """(internal)
213
214        Category: General Utility Functions
215        """
216        return _baplus.sign_out_v1(v2_embedded)
217
218    @staticmethod
219    def submit_score(
220        game: str,
221        config: str,
222        name: Any,
223        score: int | None,
224        callback: Callable,
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)