|
| 1 | +from copy import deepcopy |
| 2 | +import logging |
| 3 | +import random |
| 4 | +from types import MappingProxyType |
| 5 | +from typing import Any, NamedTuple |
| 6 | + |
| 7 | +from app.model.Level import Level |
| 8 | +from app.model.LevelLoader.LevelLoader import LevelLoader |
| 9 | +from app.model.TutorialStatus import TutorialStatus |
| 10 | +from app.utilsGame import LevelType |
| 11 | +from app.config import load_config |
| 12 | + |
| 13 | +CONFIG_KEY_LEVEL_LIST = 'pools' |
| 14 | + |
| 15 | +class LeanSlide(NamedTuple): |
| 16 | + slideType: LevelType |
| 17 | + fileName: str |
| 18 | + |
| 19 | + |
| 20 | +class JsonLevelList(LevelLoader): |
| 21 | + singleton: dict[str, Any]|None = None |
| 22 | + |
| 23 | + |
| 24 | + def __init__(self, |
| 25 | + phaseName: str, |
| 26 | + phaseConfig: dict[str, Any], |
| 27 | + tutorialStatus: dict[str, TutorialStatus], |
| 28 | + levelList: dict[str, Any] |
| 29 | + ) -> None: |
| 30 | + super().__init__(phaseName, phaseConfig, tutorialStatus) |
| 31 | + |
| 32 | + self.levelList = levelList |
| 33 | + self.pool: dict[str, list[dict[str, Any]|list[dict[str, Any]]]] = {} |
| 34 | + |
| 35 | + |
| 36 | + def loadLevels(self) -> list[Level]: |
| 37 | + """ |
| 38 | + |
| 39 | + NOTE: This method is single use. After you have loaded the levels for a |
| 40 | + Player/Phase combo, you have to create a new `JsonLevelList`. |
| 41 | + """ |
| 42 | + self._levels = [] |
| 43 | + list_names: str|list[str] = self._phaseConfig[CONFIG_KEY_LEVEL_LIST] |
| 44 | + if isinstance(list_names, str): |
| 45 | + list_names = [list_names] |
| 46 | + assert isinstance(list_names, list), "Expected an array of strings containing the level list names" |
| 47 | + |
| 48 | + for list_name in list_names: |
| 49 | + self.parse_list(list_name) |
| 50 | + |
| 51 | + assert self._levels is not None and len(self._levels) > 0, "No levels have been loaded, this is probably an error" |
| 52 | + return self._levels |
| 53 | + |
| 54 | + |
| 55 | + def parse_list(self, list_name: str): |
| 56 | + # read only, since this is loaded from a JSON and we wan't no side effects. |
| 57 | + # But be careful, this is not a frozendict, children can still be modified |
| 58 | + current_list = MappingProxyType(self.levelList[list_name]) |
| 59 | + |
| 60 | + # Load settings from current_list |
| 61 | + amount: int|str|list[str] = current_list.get('amount', 'all') |
| 62 | + shuffle: bool = current_list.get('shuffle', False) |
| 63 | + eliminate: bool = current_list.get('eliminate', True) |
| 64 | + shuffle_amount: bool = current_list.get('shuffle_amount', True) |
| 65 | + |
| 66 | + # We are working on a copy of the current_list to allow for elimination of levels. |
| 67 | + # We call this copy pool and we only store the actual level list |
| 68 | + if list_name not in self.pool: |
| 69 | + self.pool[list_name] = deepcopy(current_list['levels']) |
| 70 | + |
| 71 | + # Shuffle the pool if enabled |
| 72 | + if shuffle: |
| 73 | + random.shuffle(self.pool[list_name]) |
| 74 | + |
| 75 | + # Set amount to length of pool if we wan't to load all levels (default) |
| 76 | + if amount == 'all': |
| 77 | + amount = len(self.pool[list_name]) |
| 78 | + |
| 79 | + if isinstance(amount, int): |
| 80 | + self.load_entries(self.pool[list_name], list_name, amount, eliminate) |
| 81 | + |
| 82 | + # Special case for when we need multiple versions/difficulties of the same task |
| 83 | + # and the task group shall only be shown once |
| 84 | + elif isinstance(amount, list): |
| 85 | + assert eliminate, "This configuration only makes sense with eliminate enabled" |
| 86 | + self.load_entries_multiversion(self.pool[list_name], list_name, amount, shuffle_amount) |
| 87 | + |
| 88 | + else: |
| 89 | + raise ValueError('An invalid value for amount has made it through the pre checks') |
| 90 | + |
| 91 | + |
| 92 | + def load_entries(self, |
| 93 | + current_pool: list[dict[str, Any] | list[dict[str, Any]]], |
| 94 | + list_name: str, |
| 95 | + amount: int, |
| 96 | + eliminate: bool |
| 97 | + ): |
| 98 | + # Check that the validator has caught all invalid edge cases |
| 99 | + assert amount > 0, f'Amount of pool "{list_name}" must be > 0, got {amount}' |
| 100 | + assert amount <= len(current_pool), f'Requested {amount} levels from pool "{list_name}" but the pool only contains {len(current_pool)} levels' |
| 101 | + |
| 102 | + for i in range(0, amount): |
| 103 | + entry = current_pool[0 if eliminate else i] |
| 104 | + assert not isinstance(entry, list), "Level Groups are not allowed when the amount is an integer or 'all'!" |
| 105 | + self._appendLevel(slideType=entry['type'], fileName=entry['name']) |
| 106 | + |
| 107 | + # Remove entry from pool if it shall only be shown once |
| 108 | + if eliminate: |
| 109 | + current_pool.remove(entry) |
| 110 | + |
| 111 | + |
| 112 | + def load_entries_multiversion(self, |
| 113 | + current_pool: list[dict[str, Any] | list[dict[str, Any]]], |
| 114 | + list_name: str, |
| 115 | + amount: list[str], |
| 116 | + shuffle_amount: bool |
| 117 | + ): |
| 118 | + group_order = amount.copy() |
| 119 | + assert all(len(level_group) == len(group_order) for level_group in current_pool), f"Something in {list_name} slipped through the validator" |
| 120 | + |
| 121 | + # Shuffle the level group order if enabled |
| 122 | + if shuffle_amount: |
| 123 | + random.shuffle(group_order) |
| 124 | + |
| 125 | + # Draw exactly one level of each group in the specified order (might be randomized) |
| 126 | + for group_name in group_order: |
| 127 | + level_group = current_pool.pop(0) |
| 128 | + assert isinstance(level_group, list), "Please specify a list that contains one level of each group" |
| 129 | + entry = [level for level in level_group if level['group'] == group_name] |
| 130 | + assert len(entry) == 1, "There should be exactly one entry matching this group in the level_group" |
| 131 | + self._appendLevel(slideType=entry[0]['type'], fileName=entry[0]['name']) |
| 132 | + |
| 133 | + |
| 134 | + @staticmethod |
| 135 | + def fromFile( |
| 136 | + fileName: str = 'conf/levelList.json', |
| 137 | + instanceFolder: str = 'instance' |
| 138 | + ) -> dict[str, Any]: |
| 139 | + """Load all level lists from `conf/levelList.json` into a `dict`""" |
| 140 | + |
| 141 | + try: |
| 142 | + conf = load_config(fileName=fileName, instanceFolder=instanceFolder) |
| 143 | + |
| 144 | + # TODO Run checks |
| 145 | + |
| 146 | + logging.info(f'Successfully loaded {len(conf)} level lists.') |
| 147 | + return conf |
| 148 | + |
| 149 | + except Exception as e: |
| 150 | + logging.info(f'No level lists loaded from "{fileName}".') |
| 151 | + logging.debug(e) |
| 152 | + return {} |
0 commit comments