RYDE-WORK/ballistica
1
0
Fork 0
mirror of https://github.com/RYDE-WORK/ballistica.git synced 2026-02-02 13:23:27 +08:00
Code Issues Actions Packages Projects Releases Wiki Activity

cleaned up ba_meta scanning to return a dataclass instead of a dict

Browse Source
This commit is contained in:
Eric Froemling 2020-03-22 18:37:36 -07:00
parent 7366738622
commit 9bb0caa5bb
3 changed files with 76 additions and 69 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

6
assets/src/ba_data/python/ba/_account.py
View File

@ -173,10 +173,12 @@ def have_pro_options() -> bool:
# We expose pro options if the server tells us to # We expose pro options if the server tells us to
# (which is generally just when we own pro), # (which is generally just when we own pro),
# or also if we've been grandfathered in. # or also if we've been grandfathered in or are using ballistica-core
# builds.
return bool( return bool(
_ba.get_account_misc_read_val_2('proOptionsUnlocked', False) _ba.get_account_misc_read_val_2('proOptionsUnlocked', False)
or _ba.app.config.get('lc14292', 0) > 1) or _ba.app.config.get('lc14292', 0) > 1
or 'ballistica' + 'core' == 'ballisticacore')
def show_post_purchase_message() -> None: def show_post_purchase_message() -> None:

60
assets/src/ba_data/python/ba/_app.py
View File

@ -28,7 +28,7 @@ import _ba
if TYPE_CHECKING: if TYPE_CHECKING:
import ba import ba
from ba import _lang as bs_lang from ba import _lang, _meta
from bastd.actor import spazappearance from bastd.actor import spazappearance
from typing import (Optional, Dict, Tuple, Set, Any, List, Type, Tuple, from typing import (Optional, Dict, Tuple, Set, Any, List, Type, Tuple,
Callable) Callable)
@ -240,16 +240,7 @@ class App:
""" """
# pylint: disable=too-many-statements # pylint: disable=too-many-statements
test_https = False # _test_https()
if test_https:
# Testing https support (would be nice to get this working on
# our custom python builds; need to wrangle certificates somehow).
import urllib.request
try:
val = urllib.request.urlopen('https://example.com').read()
print("HTTPS TEST SUCCESS", len(val))
except Exception as exc:
print("HTTPS TEST FAIL:", exc)
# Config. # Config.
self.config_file_healthy = False self.config_file_healthy = False
@ -274,7 +265,7 @@ class App:
self._platform: str = env['platform'] self._platform: str = env['platform']
self._subplatform: str = env['subplatform'] self._subplatform: str = env['subplatform']
self._interface_type: str = env['interface_type'] self._interface_type: str = env['interface_type']
self._on_tv: bool = env['on_tv'] # self._on_tv: bool = env['on_tv']
self._vr_mode: bool = env['vr_mode'] self._vr_mode: bool = env['vr_mode']
self.protocol_version: int = env['protocol_version'] self.protocol_version: int = env['protocol_version']
self.toolbar_test: bool = env['toolbar_test'] self.toolbar_test: bool = env['toolbar_test']
@ -282,7 +273,7 @@ class App:
# Misc. # Misc.
self.default_language = self._get_default_language() self.default_language = self._get_default_language()
self.metascan: Optional[Dict[str, Any]] = None self.metascan: Optional[_meta.ScanResults] = None
self.tips: List[str] = [] self.tips: List[str] = []
self.stress_test_reset_timer: Optional[ba.Timer] = None self.stress_test_reset_timer: Optional[ba.Timer] = None
self.suppress_debug_reports = False self.suppress_debug_reports = False
@ -337,8 +328,8 @@ class App:
} }
# Language. # Language.
self.language_target: Optional[bs_lang.AttrDict] = None self.language_target: Optional[_lang.AttrDict] = None
self.language_merged: Optional[bs_lang.AttrDict] = None self.language_merged: Optional[_lang.AttrDict] = None
# Achievements. # Achievements.
self.achievements: List[ba.Achievement] = [] self.achievements: List[ba.Achievement] = []
@ -418,7 +409,6 @@ class App:
# pylint: disable=too-many-locals # pylint: disable=too-many-locals
# pylint: disable=cyclic-import # pylint: disable=cyclic-import
from ba import _apputils from ba import _apputils
# from ba._general import Call
from ba import _appconfig from ba import _appconfig
from ba import ui as bsui from ba import ui as bsui
from ba import _achievement from ba import _achievement
@ -443,6 +433,8 @@ class App:
self.music_player_type = _music.InternalMusicPlayer self.music_player_type = _music.InternalMusicPlayer
elif _ba.env()['platform'] == 'mac' and hasattr(_ba, 'itunes_init'): elif _ba.env()['platform'] == 'mac' and hasattr(_ba, 'itunes_init'):
self.music_player_type = _music.MacITunesMusicPlayer self.music_player_type = _music.MacITunesMusicPlayer
# FIXME: This should not be hard-coded.
for maptype in [ for maptype in [
stdmaps.HockeyStadium, stdmaps.FootballStadium, stdmaps.HockeyStadium, stdmaps.FootballStadium,
stdmaps.Bridgit, stdmaps.BigG, stdmaps.Roundabout, stdmaps.Bridgit, stdmaps.BigG, stdmaps.Roundabout,
@ -543,6 +535,7 @@ class App:
# Debugging - make note if we're using the local test server so we # Debugging - make note if we're using the local test server so we
# don't accidentally leave it on in a release. # don't accidentally leave it on in a release.
# FIXME - move this to native layer.
server_addr = _ba.get_master_server_address() server_addr = _ba.get_master_server_address()
if 'localhost' in server_addr: if 'localhost' in server_addr:
_ba.timer(2.0, _ba.timer(2.0,
@ -574,10 +567,8 @@ class App:
if self.subplatform != 'headless': if self.subplatform != 'headless':
_ba.timer(3.0, check_special_offer, timetype=TimeType.REAL) _ba.timer(3.0, check_special_offer, timetype=TimeType.REAL)
_meta.startscan() # Start scanning for things exposed via ba_meta.
_meta.start_scan()
# Start scanning for stuff available in our scripts.
# meta.get_game_types()
# Auto-sign-in to a local account in a moment if we're set to. # Auto-sign-in to a local account in a moment if we're set to.
def do_auto_sign_in() -> None: def do_auto_sign_in() -> None:
@ -592,8 +583,6 @@ class App:
from ba._dependency import test_depset from ba._dependency import test_depset
test_depset() test_depset()
# print('GAME TYPES ARE', meta.get_game_types())
# _bs.quit()
def read_config(self) -> None: def read_config(self) -> None:
"""(internal)""" """(internal)"""
@ -610,21 +599,21 @@ class App:
activity = _ba.get_foreground_host_activity() activity = _ba.get_foreground_host_activity()
if (activity is not None and activity.allow_pausing if (activity is not None and activity.allow_pausing
and not _ba.have_connected_clients()): and not _ba.have_connected_clients()):
from ba import _gameutils, _actor, _lang
# FIXME: Shouldn't be touching scene stuff here; # FIXME: Shouldn't be touching scene stuff here;
# should just pass the request on to the host-session. # should just pass the request on to the host-session.
import ba with _ba.Context(activity):
with ba.Context(activity): globs = _gameutils.sharedobj('globals')
globs = ba.sharedobj('globals')
if not globs.paused: if not globs.paused:
ba.playsound(ba.getsound('refWhistle')) _ba.playsound(_ba.getsound('refWhistle'))
globs.paused = True globs.paused = True
# FIXME: This should not be an attr on Actor. # FIXME: This should not be an attr on Actor.
activity.paused_text = ba.Actor( activity.paused_text = _actor.Actor(
ba.newnode( _ba.newnode(
'text', 'text',
attrs={ attrs={
'text': ba.Lstr(resource='pausedByHostText'), 'text': _lang.Lstr(resource='pausedByHostText'),
'client_only': True, 'client_only': True,
'flatness': 1.0, 'flatness': 1.0,
'h_align': 'center' 'h_align': 'center'
@ -821,3 +810,16 @@ class App:
else: else:
_ba.screenmessage(Lstr(resource='errorText'), color=(1, 0, 0)) _ba.screenmessage(Lstr(resource='errorText'), color=(1, 0, 0))
_ba.playsound(_ba.getsound('error')) _ba.playsound(_ba.getsound('error'))
def _test_https(self) -> None:
"""Testing https support.
(would be nice to get this working on our custom python builds; need
to wrangle certificates somehow).
"""
import urllib.request
try:
val = urllib.request.urlopen('https://example.com').read()
print("HTTPS TEST SUCCESS", len(val))
except Exception as exc:
print("HTTPS TEST FAIL:", exc)

79
assets/src/ba_data/python/ba/_meta.py
View File

@ -26,11 +26,12 @@ import os
import pathlib import pathlib
import threading import threading
from typing import TYPE_CHECKING from typing import TYPE_CHECKING
from dataclasses import dataclass, field
import _ba import _ba
if TYPE_CHECKING: if TYPE_CHECKING:
from typing import (Any, Dict, List, Tuple, Union, Optional, Type, Set) from typing import Dict, List, Tuple, Union, Optional, Type, Set
import ba import ba
# The API version of this build of the game. # The API version of this build of the game.
@ -40,7 +41,7 @@ if TYPE_CHECKING:
CURRENT_API_VERSION = 6 CURRENT_API_VERSION = 6
def startscan() -> None: def start_scan() -> None:
"""Begin scanning script directories for scripts containing metadata. """Begin scanning script directories for scripts containing metadata.
Should be called only once at launch.""" Should be called only once at launch."""
@ -52,23 +53,31 @@ def startscan() -> None:
thread.start() thread.start()
def handle_scan_results(results: Dict[str, Any]) -> None: @dataclass
class ScanResults:
"""Final results from a metadata scan."""
games: List[str] = field(default_factory=list)
errors: str = ''
warnings: str = ''
def handle_scan_results(results: ScanResults) -> None:
"""Called in the game thread with results of a completed scan.""" """Called in the game thread with results of a completed scan."""
from ba import _lang from ba import _lang
# Warnings generally only get printed locally for users' benefit # Warnings generally only get printed locally for users' benefit
# (things like out-of-date scripts being ignored, etc.) # (things like out-of-date scripts being ignored, etc.)
# Errors are more serious and will get included in the regular log # Errors are more serious and will get included in the regular log
warnings = results.get('warnings', '') # warnings = results.get('warnings', '')
errors = results.get('errors', '') # errors = results.get('errors', '')
if warnings != '' or errors != '': if results.warnings != '' or results.errors != '':
_ba.screenmessage(_lang.Lstr(resource='scanScriptsErrorText'), _ba.screenmessage(_lang.Lstr(resource='scanScriptsErrorText'),
color=(1, 0, 0)) color=(1, 0, 0))
_ba.playsound(_ba.getsound('error')) _ba.playsound(_ba.getsound('error'))
if warnings != '': if results.warnings != '':
_ba.log(warnings, to_server=False) _ba.log(results.warnings, to_server=False)
if errors != '': if results.errors != '':
_ba.log(errors) _ba.log(results.errors)
class ScanThread(threading.Thread): class ScanThread(threading.Thread):
@ -85,7 +94,8 @@ class ScanThread(threading.Thread):
scan.scan() scan.scan()
results = scan.results results = scan.results
except Exception as exc: except Exception as exc:
results = {'errors': 'Scan exception: ' + str(exc)} # results = {'errors': 'Scan exception: ' + str(exc)}
results = ScanResults(errors=f'Scan exception: {exc}')
# Push a call to the game thread to print warnings/errors # Push a call to the game thread to print warnings/errors
# or otherwise deal with scan results. # or otherwise deal with scan results.
@ -116,11 +126,7 @@ class DirectoryScan:
'errors': errors encountered during scan; should be fully logged 'errors': errors encountered during scan; should be fully logged
""" """
self.paths = [pathlib.Path(p) for p in paths] self.paths = [pathlib.Path(p) for p in paths]
self.results: Dict[str, Any] = { self.results = ScanResults()
'errors': '',
'warnings': '',
'games': []
}
def _get_path_module_entries( def _get_path_module_entries(
self, path: pathlib.Path, subpath: Union[str, pathlib.Path], self, path: pathlib.Path, subpath: Union[str, pathlib.Path],
@ -137,7 +143,7 @@ class DirectoryScan:
entries = [] entries = []
except Exception as exc: except Exception as exc:
# Unexpected; report this. # Unexpected; report this.
self.results['errors'] += str(exc) + '\n' self.results.errors += f'{exc}\n'
entries = [] entries = []
# Now identify python packages/modules out of what we found. # Now identify python packages/modules out of what we found.
@ -158,9 +164,8 @@ class DirectoryScan:
self.scan_module(moduledir, subpath) self.scan_module(moduledir, subpath)
except Exception: except Exception:
from ba import _error from ba import _error
self.results['warnings'] += ("Error scanning '" + self.results.warnings += ("Error scanning '" + str(subpath) +
str(subpath) + "': " + "': " + _error.exc_str() + '\n')
_error.exc_str() + '\n')
def scan_module(self, moduledir: pathlib.Path, def scan_module(self, moduledir: pathlib.Path,
subpath: pathlib.Path) -> None: subpath: pathlib.Path) -> None:
@ -187,11 +192,11 @@ class DirectoryScan:
# If we find a module requiring a different api version, warn # If we find a module requiring a different api version, warn
# and ignore. # and ignore.
if required_api is not None and required_api != CURRENT_API_VERSION: if required_api is not None and required_api != CURRENT_API_VERSION:
self.results['warnings'] += ('Warning: ' + str(subpath) + self.results.warnings += ('Warning: ' + str(subpath) +
' requires api ' + str(required_api) + ' requires api ' + str(required_api) +
' but we are running ' + ' but we are running ' +
str(CURRENT_API_VERSION) + str(CURRENT_API_VERSION) +
'; ignoring module.\n') '; ignoring module.\n')
return return
# Ok; can proceed with a full scan of this module. # Ok; can proceed with a full scan of this module.
@ -206,9 +211,8 @@ class DirectoryScan:
self.scan_module(submodule[0], submodule[1]) self.scan_module(submodule[0], submodule[1])
except Exception: except Exception:
from ba import _error from ba import _error
self.results['warnings'] += ("Error scanning '" + self.results.warnings += ("Error scanning '" + str(subpath) +
str(subpath) + "': " + "': " + _error.exc_str() + '\n')
_error.exc_str() + '\n')
def _process_module_meta_tags(self, subpath: pathlib.Path, def _process_module_meta_tags(self, subpath: pathlib.Path,
flines: List[str], flines: List[str],
@ -219,7 +223,7 @@ class DirectoryScan:
# the ba_meta is in the right place. # the ba_meta is in the right place.
if mline[0] != 'ba_meta': if mline[0] != 'ba_meta':
print(f'GOT "{mline[0]}"') print(f'GOT "{mline[0]}"')
self.results['warnings'] += ( self.results.warnings += (
'Warning: ' + str(subpath) + 'Warning: ' + str(subpath) +
': malformed ba_meta statement on line ' + ': malformed ba_meta statement on line ' +
str(lindex + 1) + '.\n') str(lindex + 1) + '.\n')
@ -230,7 +234,7 @@ class DirectoryScan:
elif len(mline) != 3 or mline[1] != 'export': elif len(mline) != 3 or mline[1] != 'export':
# Currently we only support 'ba_meta export FOO'; # Currently we only support 'ba_meta export FOO';
# complain for anything else we see. # complain for anything else we see.
self.results['warnings'] += ( self.results.warnings += (
'Warning: ' + str(subpath) + 'Warning: ' + str(subpath) +
': unrecognized ba_meta statement on line ' + ': unrecognized ba_meta statement on line ' +
str(lindex + 1) + '.\n') str(lindex + 1) + '.\n')
@ -245,9 +249,9 @@ class DirectoryScan:
if export_class_name is not None: if export_class_name is not None:
classname = modulename + '.' + export_class_name classname = modulename + '.' + export_class_name
if exporttype == 'game': if exporttype == 'game':
self.results['games'].append(classname) self.results.games.append(classname)
else: else:
self.results['warnings'] += ( self.results.warnings += (
'Warning: ' + str(subpath) + 'Warning: ' + str(subpath) +
': unrecognized export type "' + exporttype + ': unrecognized export type "' + exporttype +
'" on line ' + str(lindex + 1) + '.\n') '" on line ' + str(lindex + 1) + '.\n')
@ -272,7 +276,7 @@ class DirectoryScan:
classname = cbits[0] classname = cbits[0]
break # success! break # success!
if classname is None: if classname is None:
self.results['warnings'] += ( self.results.warnings += (
'Warning: ' + str(subpath) + ': class definition not found' 'Warning: ' + str(subpath) + ': class definition not found'
' below "ba_meta export" statement on line ' + ' below "ba_meta export" statement on line ' +
str(lindexorig + 1) + '.\n') str(lindexorig + 1) + '.\n')
@ -296,21 +300,21 @@ class DirectoryScan:
# Ok; not successful. lets issue warnings for a few error cases. # Ok; not successful. lets issue warnings for a few error cases.
if len(lines) > 1: if len(lines) > 1:
self.results['warnings'] += ( self.results.warnings += (
'Warning: ' + str(subpath) + 'Warning: ' + str(subpath) +
': multiple "# ba_meta api require <NUM>" lines found;' ': multiple "# ba_meta api require <NUM>" lines found;'
' ignoring module.\n') ' ignoring module.\n')
elif not lines and toplevel and meta_lines: elif not lines and toplevel and meta_lines:
# If we're a top-level module containing meta lines but # If we're a top-level module containing meta lines but
# no valid api require, complain. # no valid api require, complain.
self.results['warnings'] += ( self.results.warnings += (
'Warning: ' + str(subpath) + 'Warning: ' + str(subpath) +
': no valid "# ba_meta api require <NUM>" line found;' ': no valid "# ba_meta api require <NUM>" line found;'
' ignoring module.\n') ' ignoring module.\n')
return None return None
def get_scan_results() -> Dict[str, Any]: def get_scan_results() -> ScanResults:
"""Return meta scan results; blocking if the scan is not yet complete.""" """Return meta scan results; blocking if the scan is not yet complete."""
import time import time
app = _ba.app app = _ba.app
@ -326,7 +330,6 @@ def get_scan_results() -> Dict[str, Any]:
time.sleep(0.05) time.sleep(0.05)
if time.time() - starttime > 10.0: if time.time() - starttime > 10.0:
raise Exception('timeout waiting for meta scan to complete.') raise Exception('timeout waiting for meta scan to complete.')
print('RETURNING SCAN RESULTS')
return app.metascan return app.metascan
@ -334,7 +337,7 @@ def get_game_types() -> List[Type[ba.GameActivity]]:
"""Return available game types.""" """Return available game types."""
from ba import _general from ba import _general
from ba import _gameactivity from ba import _gameactivity
gameclassnames = get_scan_results().get('games', []) gameclassnames = get_scan_results().games
gameclasses = [] gameclasses = []
for gameclassname in gameclassnames: for gameclassname in gameclassnames:
try: try: