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    def send_message_cb(
105        self,
106        msg: Message,
107        on_response: Callable[[Any], None],
108    ) -> None:
109        """Asynchronously send a message to the cloud from the logic thread.
110
111        The provided on_response call will be run in the logic thread
112        and passed either the response or the error that occurred.
113        """
114
115        del msg  # Unused.
116
117        babase.pushcall(
118            babase.Call(
119                on_response,
120                RuntimeError('Cloud functionality is not available.'),
121            )
122        )
123
124    @overload
125    def send_message(
126        self, msg: bacommon.cloud.WorkspaceFetchMessage
127    ) -> bacommon.cloud.WorkspaceFetchResponse: ...
128
129    @overload
130    def send_message(
131        self, msg: bacommon.cloud.MerchAvailabilityMessage
132    ) -> bacommon.cloud.MerchAvailabilityResponse: ...
133
134    @overload
135    def send_message(
136        self, msg: bacommon.cloud.TestMessage
137    ) -> bacommon.cloud.TestResponse: ...
138
139    def send_message(self, msg: Message) -> Response | None:
140        """Synchronously send a message to the cloud.
141
142        Must be called from a background thread.
143        """
144        raise RuntimeError('Cloud functionality is not available.')
145
146    @overload
147    async def send_message_async(
148        self, msg: bacommon.cloud.PromoCodeMessage
149    ) -> bacommon.cloud.PromoCodeResponse: ...
150
151    @overload
152    async def send_message_async(
153        self, msg: bacommon.cloud.TestMessage
154    ) -> bacommon.cloud.TestResponse: ...
155
156    async def send_message_async(self, msg: Message) -> Response | None:
157        """Synchronously send a message to the cloud.
158
159        Must be called from the logic thread.
160        """
161        raise RuntimeError('Cloud functionality is not available.')

 21class PlusSubsystem(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 have_incentivized_ad() -> bool:
260        """Is an incentivized ad available?"""
261        return _baplus.have_incentivized_ad()
262
263    @staticmethod
264    def has_video_ads() -> bool:
265        """Are video ads available?"""
266        return _baplus.has_video_ads()
267
268    @staticmethod
269    def can_show_ad() -> bool:
270        """Can we show an ad?"""
271        return _baplus.can_show_ad()
272
273    @staticmethod
274    def show_ad(
275        purpose: str, on_completion_call: Callable[[], None] | None = None
276    ) -> None:
277        """Show an ad."""
278        _baplus.show_ad(purpose, on_completion_call)
279
280    @staticmethod
281    def show_ad_2(
282        purpose: str, on_completion_call: Callable[[bool], None] | None = None
283    ) -> None:
284        """Show an ad."""
285        _baplus.show_ad_2(purpose, on_completion_call)
286
287    @staticmethod
288    def show_game_service_ui(
289        show: str = 'general',
290        game: str | None = None,
291        game_version: str | None = None,
292    ) -> None:
293        """Show game-service provided UI."""
294        _baplus.show_game_service_ui(show, game, game_version)