76 lines
2.5 KiB
Python
76 lines
2.5 KiB
Python
|
import json
|
||
|
|
import sys
|
||
|
|
import warnings
|
||
|
|
from pathlib import Path
|
||
|
|
from typing import Iterable, Union, TYPE_CHECKING
|
||
|
|
|
||
|
|
if TYPE_CHECKING:
|
||
|
|
from multiprocessing.managers import ListProxy # noqa
|
||
|
|
|
||
|
|
__all__ = [
|
||
|
|
"generate_local",
|
||
|
|
]
|
||
|
|
|
||
|
|
|
||
|
|
def _generate_local_inner(games: Iterable[str],
|
||
|
|
dest: Union[Path, str],
|
||
|
|
results: "ListProxy[Union[Path, BaseException]]") -> None:
|
||
|
|
original_argv = sys.argv
|
||
|
|
warnings.simplefilter("ignore")
|
||
|
|
try:
|
||
|
|
from tempfile import TemporaryDirectory
|
||
|
|
|
||
|
|
if not isinstance(dest, Path):
|
||
|
|
dest = Path(dest)
|
||
|
|
|
||
|
|
with TemporaryDirectory() as players_dir:
|
||
|
|
with TemporaryDirectory() as output_dir:
|
||
|
|
import Generate
|
||
|
|
|
||
|
|
for n, game in enumerate(games, 1):
|
||
|
|
player_path = Path(players_dir) / f"{n}.yaml"
|
||
|
|
with open(player_path, "w", encoding="utf-8") as f:
|
||
|
|
f.write(json.dumps({
|
||
|
|
"name": f"Player{n}",
|
||
|
|
"game": game,
|
||
|
|
game: {"hard_mode": "true"},
|
||
|
|
"description": f"generate_local slot {n} ('Player{n}'): {game}",
|
||
|
|
}))
|
||
|
|
|
||
|
|
# this is basically copied from test/programs/test_generate.py
|
||
|
|
# uses a reproducible seed that is different for each set of games
|
||
|
|
sys.argv = [sys.argv[0], "--seed", str(hash(tuple(games))),
|
||
|
|
"--player_files_path", players_dir,
|
||
|
|
"--outputpath", output_dir]
|
||
|
|
Generate.main()
|
||
|
|
output_files = list(Path(output_dir).glob('*.zip'))
|
||
|
|
assert len(output_files) == 1
|
||
|
|
final_file = dest / output_files[0].name
|
||
|
|
output_files[0].rename(final_file)
|
||
|
|
results.append(final_file)
|
||
|
|
except BaseException as e:
|
||
|
|
results.append(e)
|
||
|
|
raise e
|
||
|
|
finally:
|
||
|
|
sys.argv = original_argv
|
||
|
|
|
||
|
|
|
||
|
|
def generate_local(games: Iterable[str], dest: Union[Path, str]) -> Path:
|
||
|
|
from multiprocessing import Manager, Process, set_start_method
|
||
|
|
|
||
|
|
try:
|
||
|
|
set_start_method("spawn")
|
||
|
|
except RuntimeError:
|
||
|
|
pass
|
||
|
|
|
||
|
|
manager = Manager()
|
||
|
|
results: "ListProxy[Union[Path, Exception]]" = manager.list()
|
||
|
|
|
||
|
|
p = Process(target=_generate_local_inner, args=(games, dest, results))
|
||
|
|
p.start()
|
||
|
|
p.join()
|
||
|
|
result = results[0]
|
||
|
|
if isinstance(result, BaseException):
|
||
|
|
raise Exception("Could not generate multiworld") from result
|
||
|
|
return result
|