Source code for babase._meta
# Released under the MIT License. See LICENSE for details.
#
"""Functionality related to dynamic discoverability of classes."""
from __future__ import annotations # Docs-generation hack.
import os
import logging
from threading import Thread, Event
from functools import partial
from typing import TYPE_CHECKING
import _babase
from efro.util import strip_exception_tracebacks
from bacommon.metascan import (
APP_SCAN_ALIAS,
DEPRECATED_EXPORT_SHORTCUTS,
DirectoryScan,
ScanResults,
)
from bacommon.metascancache import ScanCache
from babase._logging import lifecyclelog
if TYPE_CHECKING:
from typing import Callable
[docs]
class MetadataSubsystem:
"""Subsystem for working with script metadata in the app.
Access the single shared instance of this class via the
:attr:`~babase.App.meta` attr on the :class:`~babase.App` class.
"""
def __init__(self) -> None:
self._scan: DirectoryScan | None = None
# Can be populated before starting the scan.
self.extra_scan_dirs: list[str] = []
# Results populated once scan is complete.
self.scanresults: ScanResults | None = None
# Set (after scanresults is in place) when the scan completes.
self._scan_results_ready = Event()
self._scan_complete_cb: Callable[[], None] | None = None
# Where per-module scan outcomes persist between launches (see
# ScanCache), what keys files with frozen mtimes, and where a
# build-time seed may live.
self._scan_cache_path: str | None = None
self._scan_cache_frozen_salt: str | None = None
self._scan_seed_path: str | None = None
def start_scan(self, scan_complete_cb: Callable[[], None]) -> None:
"""Begin the overall scan.
This will start scanning built in directories (which for vanilla
installs should be the vast majority of the work). This should
only be called once.
:meta private:
"""
assert self._scan_complete_cb is None
assert self._scan is None
env = _babase.app.env
self._scan_complete_cb = scan_complete_cb
appdir = env.python_directory_app
userdir = env.python_directory_user
aliases: dict[str, str] = {}
if appdir is not None:
aliases[appdir] = APP_SCAN_ALIAS
if userdir is not None:
aliases[userdir] = 'user'
self._scan = DirectoryScan(
paths=[path for path in [appdir, userdir] if path is not None],
expected_api_version=env.api_version,
deprecated_export_shortcuts=DEPRECATED_EXPORT_SHORTCUTS,
expects_extras=True,
path_aliases=aliases,
)
# Reusing last launch's per-module outcomes (for whatever
# hasn't changed) saves reading and parsing most of the source
# each launch. Installs that freeze file mtimes (flatpak/nix)
# get their files keyed by build instead, as with our pyc cache.
# Store apks also ship a seed for the first launch.
self._scan_cache_path = os.path.join(
env.cache_directory, 'metascan.json'
)
self._scan_cache_frozen_salt = f'build {env.engine_build_number}'
if appdir is not None:
self._scan_seed_path = os.path.join(
os.path.dirname(appdir), 'metascan_seed.json'
)
lifecyclelog.debug('meta-scan bg thread kicked off')
Thread(target=self._run_scan_in_bg).start()
def start_extra_scan(self) -> None:
"""Proceed to the extra_scan_dirs portion of the scan.
This is for parts of the scan that must be delayed until
workspace sync completion or other such events. This must be
called exactly once.
:meta private:
"""
assert self._scan is not None
self._scan.set_extras(self.extra_scan_dirs)
[docs]
def load_exported_classes[T](
self,
exportname: str,
cls: type[T],
completion_cb: Callable[[list[type[T]]], None],
completion_cb_in_bg_thread: bool = False,
) -> None:
"""High level function to load meta-exported classes.
Will wait for scanning to complete if necessary, and will load all
registered classes of a particular type in a background thread before
calling the passed callback in the logic thread. Errors may be logged
to messaged to the user in some way but the callback will be called
regardless.
To run the completion callback directly in the bg thread where the
loading work happens, pass ``completion_cb_in_bg_thread=True``.
"""
Thread(
target=partial(
self._load_exported_classes,
exportname,
cls,
completion_cb,
completion_cb_in_bg_thread,
)
).start()
def _load_exported_classes[T](
self,
exportname: str,
cls: type[T],
completion_cb: Callable[[list[type[T]]], None],
completion_cb_in_bg_thread: bool,
) -> None:
from babase._general import getclass
classes: list[type[T]] = []
try:
classnames = self._wait_for_scan_results().exports.get(
exportname, []
)
for classname in classnames:
try:
classes.append(getclass(classname, cls))
except Exception:
logging.exception('error importing %s', classname)
except Exception:
logging.exception('Error loading exported classes.')
completion_call = partial(completion_cb, classes)
if completion_cb_in_bg_thread:
completion_call()
else:
_babase.pushcall(completion_call, from_other_thread=True)
def _wait_for_scan_results(self) -> ScanResults:
"""Return scan results, blocking if the scan is not yet complete."""
if self.scanresults is None:
if _babase.in_logic_thread():
logging.warning(
'babase.meta._wait_for_scan_results()'
' called in logic thread before scan completed;'
' this can cause hitches.'
)
# Now wait a bit for the scan to complete. Eventually error
# though if it doesn't.
if not self._scan_results_ready.wait(timeout=10.0):
raise TimeoutError('timeout waiting for meta scan to complete.')
assert self.scanresults is not None
return self.scanresults
def _run_scan_in_bg(self) -> None:
"""Runs a scan (for use in background thread)."""
cache: ScanCache | None = None
seeded = False
try:
assert self._scan is not None
if self._scan_cache_path is not None:
cache = ScanCache.load(
self._scan_cache_path, self._scan_cache_frozen_salt
)
# With no cache of our own yet (first launch, cleared
# cache) or one from an older build (whose entries for
# anything the update changed will miss), fall back on
# this build's seed if it has one.
if self._scan_seed_path is not None and (
cache.empty or cache.install_changed
):
seeded = cache.load_seed(self._scan_seed_path)
self._scan.cache = cache
self._scan.run()
results = self._scan.results
self._scan = None
except Exception:
logging.exception('metascan: Error running scan in bg.')
results = ScanResults(announce_errors_occurred=True)
cache = None
# Place results and tell the logic thread they're ready. Note
# that scanresults must be fully in place before we set the
# event; waiters read it as soon as they wake.
self.scanresults = results
self._scan_results_ready.set()
lifecyclelog.debug(
'meta-scan bg thread done (cache: %s)',
(
'none'
if cache is None
else (
f'{cache.hits} hits ({cache.seed_hits} from seed),'
f' {cache.misses} misses'
)
),
)
_babase.pushcall(self._handle_scan_results, from_other_thread=True)
if cache is not None:
_check_scan_cache_health(cache, seeded)
# Save the cache for next launch (after handing off results,
# so this never delays anyone waiting on them).
if cache is not None and self._scan_cache_path is not None:
try:
cache.save(self._scan_cache_path)
except Exception as exc:
# Just costs next launch a full scan; not worth more
# than a log line.
logging.warning(
'metascan: Unable to save scan cache to %s: %s',
self._scan_cache_path,
exc,
)
strip_exception_tracebacks(exc)
def _handle_scan_results(self) -> None:
"""Called in the logic thread with results of a completed scan."""
from babase import _builtinassets
assert _babase.in_logic_thread()
results = self.scanresults
assert results is not None
# Any Python at all in the user's mods dir marks this run as
# user-modified for error reporting, whether or not anything
# ends up loading it (same presence-based stance the blessing
# hash takes). Plugins need no commands run to take effect, so
# the commands-run flag alone misses them. Report the verdict
# either way: until we do, that signal reads as pending.
userdir = _babase.app.env.python_directory_user
_babase.set_user_python_scan_result(
userdir is not None and results.module_counts.get(userdir, 0) > 0
)
do_play_error_sound = False
# If we found modules needing to be updated to the newer api version,
# mention that specifically.
if results.incorrect_api_modules:
if len(results.incorrect_api_modules) > 1:
msg = _builtinassets.strings.scripts.modules_need_update(
path=results.incorrect_api_modules[0],
count=len(results.incorrect_api_modules) - 1,
api=str(_babase.app.env.api_version),
)
else:
msg = _builtinassets.strings.scripts.module_needs_update(
path=results.incorrect_api_modules[0],
api=str(_babase.app.env.api_version),
)
_babase.screenmessage(msg, color=(1, 0, 0))
do_play_error_sound = True
# Let the user know if there's warning/errors in their log
# they may want to look at.
if results.announce_errors_occurred:
_babase.screenmessage(
_builtinassets.strings.scripts.scan_error, color=(1, 0, 0)
)
do_play_error_sound = True
if do_play_error_sound:
_builtinassets.audio.error.get().play()
# Let the game know we're done.
assert self._scan_complete_cb is not None
self._scan_complete_cb()
def _check_scan_cache_health(cache: ScanCache, seeded: bool) -> None:
"""Warn if the scan cache plainly isn't doing its job.
These would otherwise only show up as slower launches.
"""
if seeded:
# A seed is built from exactly the app files this build ships,
# so it should cover every one of them. (The count includes
# any app module that logs its own warning, but none should.)
app_misses = cache.misses_by_alias.get(APP_SCAN_ALIAS, 0)
if app_misses:
logging.warning(
'metascan: Seed missed %d app module(s); does it match'
' this build?',
app_misses,
)
elif (
not cache.empty
and not cache.install_changed
and cache.hits == 0
and cache.misses >= 20
):
# A cache from this same install matching nothing at all means
# module identities aren't stable here (something touching
# every file each launch, or keys computed inconsistently).
logging.warning(
'metascan: Scan cache matched none of %d module(s);'
' caching is not working.',
cache.misses,
)
# Docs-generation hack; import some stuff that we likely only forward-declared
# in our actual source code so that docs tools can find it.
from typing import (Coroutine, Any, Literal, Callable,
Generator, Awaitable, Sequence, Self, TypeIs)
import asyncio
from concurrent.futures import Future
from pathlib import Path
from enum import Enum