diff --git a/Cargo.lock b/Cargo.lock index ff364a9..49de4ef 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -4,8 +4,8 @@ version = 4 [[package]] name = "allium-deck" -version = "0.0.4" -source = "git+https://github.com/empty-sekai/allium-deck.git?rev=2cf7e77736c1d545f0858ec96711feae9e6fcbed#2cf7e77736c1d545f0858ec96711feae9e6fcbed" +version = "0.0.5" +source = "git+https://github.com/empty-sekai/allium-deck.git?rev=b799502a44a71488e82463ccfa607b2143600b23#b799502a44a71488e82463ccfa607b2143600b23" dependencies = [ "serde", "serde_json", @@ -14,7 +14,7 @@ dependencies = [ [[package]] name = "allium-deck-python" -version = "0.0.1" +version = "0.0.2" dependencies = [ "allium-deck", "pyo3", @@ -219,9 +219,9 @@ dependencies = [ [[package]] name = "syn" -version = "2.0.118" +version = "2.0.119" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "1b9ae57f904213ebb649ce6895b8a66c66f0203b9319718f69a5612a065b1422" +checksum = "872831b642d1a07999a962a351ed35b955ea2cfc8f3862091e2a240a84f17297" dependencies = [ "proc-macro2", "quote", diff --git a/Cargo.toml b/Cargo.toml index abf4d37..4d8cae6 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -1,6 +1,6 @@ [package] name = "allium-deck-python" -version = "0.0.1" +version = "0.0.2" edition = "2021" license = "MIT" publish = false @@ -10,7 +10,7 @@ name = "_native" crate-type = ["cdylib"] [dependencies] -allium-deck = { git = "https://github.com/empty-sekai/allium-deck.git", rev = "2cf7e77736c1d545f0858ec96711feae9e6fcbed" } +allium-deck = { version = "0.0.5", git = "https://github.com/empty-sekai/allium-deck.git", rev = "b799502a44a71488e82463ccfa607b2143600b23" } pyo3 = { version = "0.23", features = ["abi3-py310"] } serde_json = "1" diff --git a/README.md b/README.md index 5dc8729..b3cfa1b 100644 --- a/README.md +++ b/README.md @@ -41,6 +41,23 @@ from sekai_deck_recommend_cpp import ( Masterdata, music metadata, and user data remain runtime inputs and are not bundled into the wheel. The recommendation engine uses Allium's DFS search. +## API coverage + +The `sekai_deck_recommend_cpp` interface includes the complete LunaBot deck +workflow: + +- mutable option, user-data, card, deck, support-deck, and result objects +- single and batch recommendation +- World Bloom support-deck calculation +- area-item upgrade recommendation +- per-music score and event-point calculation +- note-level exact live calculation +- configurable batch worker count + +Each recommendation result includes `cost_ms`, the wall-clock time spent in the +native search itself. Batch results report this value independently for every +request. + ## License MIT diff --git a/pyproject.toml b/pyproject.toml index 14b279e..a5a3f8f 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -4,7 +4,7 @@ build-backend = "maturin" [project] name = "allium-sekai-deck" -version = "0.0.1" +version = "0.0.2" description = "Precompiled Rust deck recommendation engine for Python and LunaBot integrations" requires-python = ">=3.10,<4" readme = "README.md" diff --git a/python/sekai_deck_recommend_cpp/__init__.py b/python/sekai_deck_recommend_cpp/__init__.py index f7249cf..efb5c6c 100644 --- a/python/sekai_deck_recommend_cpp/__init__.py +++ b/python/sekai_deck_recommend_cpp/__init__.py @@ -1,6 +1,9 @@ from __future__ import annotations import json +import os +from collections.abc import Sequence, Set +from concurrent.futures import ThreadPoolExecutor from pathlib import Path from ._models import ( @@ -16,6 +19,51 @@ RecommendSupportDeckCard, ) +_VALID_REGIONS = {"jp", "en", "tw", "kr", "cn"} +_VALID_LIVE_TYPES = { + "multi", + "solo", + "auto", + "challenge", + "challenge_auto", + "mysekai", +} +_engine_thread_count = 0 + + +def _native_sequence(value, name: str): + if isinstance(value, (str, bytes)) or not isinstance(value, (Sequence, Set)): + raise TypeError(f"{name} must be a sequence") + return list(value) + + +def _raise_batch_error(index: int, error: Exception): + if isinstance(error, RuntimeError): + raise RuntimeError(f"batch item {index} failed: {error}") from error + raise ValueError(f"batch item {index}: {error}") from error + + +def set_engine_thread_count(threads: int) -> None: + global _engine_thread_count + if not isinstance(threads, int): + raise TypeError("threads must be an int") + if threads < 0: + raise ValueError("threads must be non-negative") + _engine_thread_count = min(threads, os.cpu_count() or 1) + + +def _effective_engine_thread_count() -> int: + if _engine_thread_count > 0: + return _engine_thread_count + raw = os.environ.get("DECK_ENGINE_THREADS", "") + try: + configured = int(raw) + except ValueError: + configured = 1 + if configured <= 0: + return 1 + return min(configured, os.cpu_count() or 1) + class SekaiDeckRecommend: def __init__(self) -> None: @@ -31,52 +79,159 @@ def _require_native(self): raise RuntimeError("_allium_deck_native is not installed") return self._native + @staticmethod + def _validate_region(region: str) -> str: + if region not in _VALID_REGIONS: + raise ValueError(f"Invalid region: {region}") + return region + + @staticmethod + def _resolve_user_data(options: DeckRecommendOptions) -> DeckRecommendUserData: + user_data = options.user_data + if user_data is None and options.user_data_file_path: + user_data = DeckRecommendUserData() + user_data.load_from_file(options.user_data_file_path) + if user_data is None and options.user_data_str is not None: + user_data = DeckRecommendUserData() + user_data.load_from_bytes(options.user_data_str) + if not isinstance(user_data, DeckRecommendUserData) or user_data._native is None: + raise ValueError( + "Either user_data / user_data_file_path / user_data_str is required." + ) + return user_data + + @classmethod + def _validate_options(cls, options: DeckRecommendOptions) -> str: + if not isinstance(options, DeckRecommendOptions): + raise TypeError("options must be DeckRecommendOptions") + if options.region is None: + raise ValueError("region is required.") + return cls._validate_region(options.region) + def update_masterdata(self, base_dir: str, region: str) -> None: - self._require_native().update_masterdata(str(Path(base_dir)), region) + self._require_native().update_masterdata( + str(Path(base_dir)), self._validate_region(region) + ) def update_masterdata_from_strings(self, data, region: str) -> None: + if not isinstance(data, dict): + raise TypeError("data must be a dict") normalized = { (name if str(name).endswith(".json") else f"{name}.json"): ( value.decode("utf-8") if isinstance(value, bytes) else str(value) ) for name, value in data.items() } - self._require_native().update_masterdata_from_strings(normalized, region) + self._require_native().update_masterdata_from_strings( + normalized, self._validate_region(region) + ) def update_musicmetas(self, path: str, region: str) -> None: - self._require_native().update_musicmetas(Path(path).read_text("utf-8"), region) + self._require_native().update_musicmetas( + Path(path).read_text("utf-8"), self._validate_region(region) + ) def update_musicmetas_from_string(self, data, region: str) -> None: text = data.decode("utf-8") if isinstance(data, bytes) else str(data) - self._require_native().update_musicmetas(text, region) + self._require_native().update_musicmetas(text, self._validate_region(region)) def recommend(self, options: DeckRecommendOptions) -> DeckRecommendResult: - if not isinstance(options, DeckRecommendOptions): - raise TypeError("options must be DeckRecommendOptions") - user_data = options.user_data - if user_data is None and options.user_data_file_path: - user_data = DeckRecommendUserData() - user_data.load_from_file(options.user_data_file_path) - if user_data is None and options.user_data_str is not None: - user_data = DeckRecommendUserData() - user_data.load_from_bytes(options.user_data_str) - if not isinstance(user_data, DeckRecommendUserData) or user_data._native is None: - raise ValueError("options.user_data is required") + region = self._validate_options(options) + user_data = self._resolve_user_data(options) payload = self._require_native().recommend( - options.region, - json.dumps(options.to_native_dict(), separators=(",", ":")), + region, + json.dumps(options._to_native_dict(), separators=(",", ":")), user_data._native, ) return DeckRecommendResult.from_dict(json.loads(payload)) - def recommend_area_items(self, *args, **kwargs): - raise NotImplementedError("recommend_area_items is not available in 0.0.1") + def recommend_batch( + self, options_list: list[DeckRecommendOptions] + ) -> list[DeckRecommendResult]: + options_list = _native_sequence(options_list, "options_list") + for index, options in enumerate(options_list): + if not isinstance(options, DeckRecommendOptions): + raise TypeError(f"batch item {index}: expected DeckRecommendOptions") + workers = min(_effective_engine_thread_count(), max(1, len(options_list))) + if workers <= 1 or len(options_list) <= 1: + result = [] + for index, options in enumerate(options_list): + try: + result.append(self.recommend(options)) + except Exception as error: + _raise_batch_error(index, error) + return result + with ThreadPoolExecutor(max_workers=workers) as executor: + futures = [executor.submit(self.recommend, options) for options in options_list] + result = [] + for index, future in enumerate(futures): + try: + result.append(future.result()) + except Exception as error: + _raise_batch_error(index, error) + return result + + def get_world_bloom_support_cards( + self, options: DeckRecommendOptions + ) -> list[RecommendSupportDeckCard]: + region = self._validate_options(options) + user_data = self._resolve_user_data(options) + payload = self._require_native().get_world_bloom_support_cards( + region, + json.dumps(options._to_native_dict(), separators=(",", ":")), + user_data._native, + ) + return [RecommendSupportDeckCard.from_dict(item) for item in json.loads(payload)] + + def recommend_area_items( + self, options: DeckRecommendOptions, card_ids: list[int] + ) -> list[dict]: + region = self._validate_options(options) + user_data = self._resolve_user_data(options) + card_ids = _native_sequence(card_ids, "card_ids") + if not 1 <= len(card_ids) <= 5: + raise ValueError("cardIds must contain 1 to 5 cards") + payload = self._require_native().recommend_area_items( + region, card_ids, user_data._native + ) + return json.loads(payload) - def recommend_music(self, *args, **kwargs): - raise NotImplementedError("recommend_music is not available in 0.0.1") + def recommend_music( + self, options: DeckRecommendOptions, deck: RecommendDeck + ) -> list[dict]: + region = self._validate_options(options) + if not isinstance(deck, RecommendDeck): + raise TypeError("deck must be RecommendDeck") + payload = self._require_native().recommend_music( + region, + json.dumps(options._to_native_dict(), separators=(",", ":")), + json.dumps(deck.to_dict(), separators=(",", ":")), + ) + return json.loads(payload) - def calculate_exact_live(self, *args, **kwargs): - raise NotImplementedError("calculate_exact_live is not available in 0.0.1") + def calculate_exact_live( + self, + region: str, + power: int, + skills: list[float], + live_type: str, + music_score_json: str, + multi_sum_power: int = 0, + fever_music_score_json: str | None = None, + ) -> dict: + skills = _native_sequence(skills, "skills") + if live_type not in _VALID_LIVE_TYPES or live_type == "mysekai": + raise ValueError(f"Invalid live type: {live_type}") + payload = self._require_native().calculate_exact_live( + self._validate_region(region), + power, + skills, + live_type, + music_score_json, + multi_sum_power, + fever_music_score_json, + ) + return json.loads(payload) __all__ = [ @@ -91,4 +246,5 @@ def calculate_exact_live(self, *args, **kwargs): "RecommendDeck", "RecommendSupportDeckCard", "SekaiDeckRecommend", + "set_engine_thread_count", ] diff --git a/python/sekai_deck_recommend_cpp/__init__.pyi b/python/sekai_deck_recommend_cpp/__init__.pyi new file mode 100644 index 0000000..4c76166 --- /dev/null +++ b/python/sekai_deck_recommend_cpp/__init__.pyi @@ -0,0 +1,207 @@ +from typing import Optional, Dict, Any, List, Union + +class DeckRecommendUserData: + def __init__(self) -> None: ... + def load_from_file(self, path: str) -> None: ... + def load_from_bytes(self, data: Union[str, bytes]) -> None: ... + +class DeckRecommendCardConfig: + disable: Optional[bool] + level_max: Optional[bool] + episode_read: Optional[bool] + master_max: Optional[bool] + skill_max: Optional[bool] + canvas: Optional[bool] + level: Optional[int] + skill_level: Optional[int] + master_rank: Optional[int] + episode_read_count: Optional[int] + def to_dict(self) -> Dict[str, Any]: ... + @classmethod + def from_dict(cls, data: Dict[str, Any]) -> 'DeckRecommendCardConfig': ... + +class DeckRecommendSingleCardConfig: + card_id: int + disable: Optional[bool] + level_max: Optional[bool] + episode_read: Optional[bool] + master_max: Optional[bool] + skill_max: Optional[bool] + canvas: Optional[bool] + level: Optional[int] + skill_level: Optional[int] + master_rank: Optional[int] + episode_read_count: Optional[int] + def to_dict(self) -> Dict[str, Any]: ... + @classmethod + def from_dict(cls, data: Dict[str, Any]) -> 'DeckRecommendSingleCardConfig': ... + +class DeckRecommendSaOptions: + run_num: Optional[int] + seed: Optional[int] + max_iter: Optional[int] + max_no_improve_iter: Optional[int] + time_limit_ms: Optional[int] + start_temprature: Optional[float] + cooling_rate: Optional[float] + debug: Optional[bool] + def to_dict(self) -> Dict[str, Any]: ... + @classmethod + def from_dict(cls, data: Dict[str, Any]) -> 'DeckRecommendSaOptions': ... + +class DeckRecommendGaOptions: + seed: Optional[int] + debug: Optional[bool] + max_iter: Optional[int] + max_no_improve_iter: Optional[int] + pop_size: Optional[int] + parent_size: Optional[int] + elite_size: Optional[int] + crossover_rate: Optional[float] + base_mutation_rate: Optional[float] + no_improve_iter_to_mutation_rate: Optional[float] + def to_dict(self) -> Dict[str, Any]: ... + @classmethod + def from_dict(cls, data: Dict[str, Any]) -> 'DeckRecommendGaOptions': ... + +class DeckRecommendOptions: + target: Optional[str] + algorithm: Optional[str] + region: str + user_data: Optional[DeckRecommendUserData] + user_data_file_path: Optional[str] + user_data_str: Optional[Union[str, bytes]] + live_type: str + music_id: int + music_diff: str + event_id: Optional[int] + event_attr: Optional[str] + event_unit: Optional[str] + event_type: Optional[str] + world_bloom_event_turn: int + world_bloom_character_id: Optional[int] + challenge_live_character_id: Optional[int] + limit: Optional[int] + member: Optional[int] + timeout_ms: Optional[int] + rarity_1_config: Optional[DeckRecommendCardConfig] + rarity_2_config: Optional[DeckRecommendCardConfig] + rarity_3_config: Optional[DeckRecommendCardConfig] + rarity_birthday_config: Optional[DeckRecommendCardConfig] + rarity_4_config: Optional[DeckRecommendCardConfig] + single_card_configs: Optional[List[DeckRecommendSingleCardConfig]] + support_master_max: Optional[bool] + support_skill_max: Optional[bool] + filter_other_unit: Optional[bool] + fixed_cards: Optional[List[int]] + fixed_characters: Optional[List[int]] + forcedLeaderCharacterId: Optional[int] + target_bonus_list: Optional[List[int]] + custom_bonus_character_ids: Optional[List[int]] + custom_bonus_attr: Optional[str] + custom_bonus_character_support_units: Optional[Dict[int, str]] + skill_reference_choose_strategy: Optional[str] + keep_after_training_state: Optional[bool] + multi_live_teammate_score_up: Optional[int] + multi_live_teammate_power: Optional[int] + best_skill_as_leader: Optional[bool] + multi_live_score_up_lower_bound: Optional[float] + skill_order_choose_strategy: Optional[str] + specific_skill_order: Optional[List[int]] + sa_options: Optional[DeckRecommendSaOptions] + ga_options: Optional[DeckRecommendGaOptions] + def to_dict(self) -> Dict[str, Any]: ... + @classmethod + def from_dict(cls, data: Dict[str, Any]) -> 'DeckRecommendOptions': ... + +class RecommendCard: + card_id: int + total_power: int + base_power: int + event_bonus_rate: float + master_rank: int + level: int + skill_level: int + skill_score_up: int + skill_life_recovery: int + episode1_read: bool + episode2_read: bool + after_training: bool + default_image: str + has_canvas_bonus: bool + def to_dict(self) -> Dict[str, Any]: ... + @classmethod + def from_dict(cls, data: Dict[str, Any]) -> 'RecommendCard': ... + +class RecommendSupportDeckCard: + card_id: int + bonus: float + skill_level: int + master_rank: int + level: int + after_training: bool + default_image: str + def to_dict(self) -> Dict[str, Any]: ... + @classmethod + def from_dict(cls, data: Dict[str, Any]) -> 'RecommendSupportDeckCard': ... + +class RecommendDeck: + score: int + live_score: int + mysekai_event_point: int + total_power: int + base_power: int + area_item_bonus_power: int + character_bonus_power: int + honor_bonus_power: int + fixture_bonus_power: int + gate_bonus_power: int + event_bonus_rate: float + support_deck_bonus_rate: float + multi_live_score_up: float + support_deck_cards: List['RecommendSupportDeckCard'] + cards: List[RecommendCard] + def to_dict(self) -> Dict[str, Any]: ... + @classmethod + def from_dict(cls, data: Dict[str, Any]) -> 'RecommendDeck': ... + +class DeckRecommendResult: + decks: List[RecommendDeck] + cost_ms: float + def to_dict(self) -> Dict[str, Any]: ... + @classmethod + def from_dict(cls, data: Dict[str, Any]) -> 'DeckRecommendResult': ... + +class SekaiDeckRecommend: + def __init__(self) -> None: ... + def update_masterdata(self, base_dir: str, region: str) -> None: ... + def update_masterdata_from_strings( + self, data: Dict[str, Union[str, bytes]], region: str + ) -> None: ... + def update_musicmetas(self, file_path: str, region: str) -> None: ... + def update_musicmetas_from_string(self, data: Union[str, bytes], region: str) -> None: ... + def recommend(self, options: DeckRecommendOptions) -> DeckRecommendResult: ... + def recommend_batch( + self, options_list: List[DeckRecommendOptions] + ) -> List[DeckRecommendResult]: ... + def get_world_bloom_support_cards( + self, options: DeckRecommendOptions + ) -> List[RecommendSupportDeckCard]: ... + def recommend_area_items( + self, options: DeckRecommendOptions, card_ids: List[int] + ) -> List[Dict[str, Any]]: ... + def recommend_music( + self, options: DeckRecommendOptions, deck: RecommendDeck + ) -> List[Dict[str, Any]]: ... + def calculate_exact_live( + self, + region: str, + power: int, + skills: List[float], + live_type: str, + music_score_json: str, + multi_sum_power: int = 0, + fever_music_score_json: Optional[str] = None, + ) -> Dict[str, Any]: ... + +def set_engine_thread_count(threads: int) -> None: ... diff --git a/python/sekai_deck_recommend_cpp/_models.py b/python/sekai_deck_recommend_cpp/_models.py index b030907..aaff33c 100644 --- a/python/sekai_deck_recommend_cpp/_models.py +++ b/python/sekai_deck_recommend_cpp/_models.py @@ -6,66 +6,141 @@ from typing import Any, ClassVar -def _safe_copy(value: Any) -> Any: - try: - return deepcopy(value) - except (TypeError, ValueError): - return value - - -def _drop_none(value: Any) -> Any: - if isinstance(value, dict): - return {key: _drop_none(item) for key, item in value.items() if item is not None} - if isinstance(value, list): - return [_drop_none(item) for item in value] - return value - - class _CompatModel: + __slots__ = () + _defaults: ClassVar[dict[str, Any]] = {} _nested: ClassVar[dict[str, tuple[type, bool]]] = {} - - def __init__(self, source: Any = None, **values: Any) -> None: + _omit_none: ClassVar[bool] = False + _required: ClassVar[tuple[str, ...]] = () + _field_kinds: ClassVar[dict[str, Any]] = {} + _reject_none_from_dict: ClassVar[bool] = False + _from_dict_ignored: ClassVar[tuple[str, ...]] = () + + def __setattr__(self, name: str, value: Any) -> None: + kind = self._field_kinds.get(name) + if kind is not None: + value = _convert_field(name, value, kind) + object.__setattr__(self, name, value) + + def __init__(self, source: Any = None) -> None: for name, default in self._defaults.items(): setattr(self, name, deepcopy(default)) if source is not None: - if isinstance(source, _CompatModel): - self._load_dict(source.to_dict()) - elif isinstance(source, dict): - self._load_dict(source) - else: - raise TypeError(f"expected {type(self).__name__} or dict") - self._load_dict(values) + if type(source) is not type(self): + raise TypeError(f"expected {type(self).__name__}") + self._load_dict(source._as_dict(include_user_data=True)) def _load_dict(self, data: dict[str, Any]) -> None: - for name, value in data.items(): + for name in self._defaults: + if name not in data: + continue + value = data[name] nested = self._nested.get(name) if nested is not None and value is not None: model, many = nested if many: - value = [item if isinstance(item, model) else model.from_dict(item) for item in value] + value = [ + item if isinstance(item, model) else model.from_dict(item) + for item in value + ] elif not isinstance(value, model): value = model.from_dict(value) - setattr(self, name, _safe_copy(value)) + setattr(self, name, value if name == "user_data" else deepcopy(value)) - def to_dict(self) -> dict[str, Any]: + def _as_dict(self, *, include_user_data: bool = False) -> dict[str, Any]: result: dict[str, Any] = {} - for name, value in vars(self).items(): - if name.startswith("_"): + for name in self._defaults: + if name == "user_data" and not include_user_data: + continue + value = getattr(self, name) + if self._omit_none and value is None: continue if isinstance(value, _CompatModel): - result[name] = value.to_dict() + value = value.to_dict() elif isinstance(value, list): - result[name] = [item.to_dict() if isinstance(item, _CompatModel) else _safe_copy(item) for item in value] - else: - result[name] = _safe_copy(value) + value = [ + item.to_dict() if isinstance(item, _CompatModel) else deepcopy(item) + for item in value + ] + elif name != "user_data": + value = deepcopy(value) + result[name] = value return result + def to_dict(self) -> dict[str, Any]: + return self._as_dict() + @classmethod def from_dict(cls, data: dict[str, Any]): if not isinstance(data, dict): raise TypeError("data must be a dict") - return cls(data) + for name in cls._required: + if name not in data: + raise KeyError(name) + instance = cls() + filtered = {key: value for key, value in data.items() if key not in cls._from_dict_ignored} + if cls._reject_none_from_dict: + for name, value in filtered.items(): + if name in cls._field_kinds and value is None: + raise TypeError(f"{name} cannot be None in from_dict") + instance._load_dict(filtered) + return instance + + +def _conversion_error(name: str, expected: str, value: Any) -> TypeError: + return TypeError( + f"incompatible value for {name}: expected {expected}, got {type(value).__name__}" + ) + + +def _convert_field(name: str, value: Any, kind: Any) -> Any: + optional = isinstance(kind, tuple) and kind[0] == "optional" + if optional: + if value is None: + return None + kind = kind[1] + + if kind == "bool": + if not isinstance(value, (bool, int)): + raise _conversion_error(name, "bool", value) + return bool(value) + if kind == "int": + if not isinstance(value, int): + raise _conversion_error(name, "int", value) + return int(value) + if kind == "float": + if not isinstance(value, (int, float)): + raise _conversion_error(name, "float", value) + return float(value) + if kind == "str": + if isinstance(value, bytes): + return value.decode("utf-8") + if not isinstance(value, str): + raise _conversion_error(name, "str", value) + return value + if isinstance(kind, tuple) and kind[0] == "model": + model = globals()[kind[1]] + if not isinstance(value, model): + raise _conversion_error(name, kind[1], value) + return value + if isinstance(kind, tuple) and kind[0] == "list": + if not isinstance(value, (list, tuple)): + raise _conversion_error(name, "list", value) + item_kind = kind[1] + return [_convert_field(name, item, item_kind) for item in value] + if kind == "int_str_dict": + if not isinstance(value, dict): + raise _conversion_error(name, "dict[int, str]", value) + return { + _convert_field(name, key, "int"): _convert_field(name, item, "str") + for key, item in value.items() + } + raise RuntimeError(f"unknown field conversion: {kind!r}") + + +def _optional(kind: Any) -> tuple[str, Any]: + return ("optional", kind) _CARD_CONFIG_DEFAULTS = { @@ -83,42 +158,139 @@ def from_dict(cls, data: dict[str, Any]): class DeckRecommendCardConfig(_CompatModel): + __slots__ = tuple(_CARD_CONFIG_DEFAULTS) _defaults = _CARD_CONFIG_DEFAULTS + _omit_none = True + _reject_none_from_dict = True + _field_kinds = { + **{name: _optional("bool") for name in ( + "disable", "level_max", "episode_read", "master_max", "skill_max", "canvas" + )}, + **{name: _optional("int") for name in ( + "level", "skill_level", "master_rank", "episode_read_count" + )}, + } class DeckRecommendSingleCardConfig(_CompatModel): + __slots__ = ("card_id", *tuple(_CARD_CONFIG_DEFAULTS)) _defaults = {"card_id": 0, **_CARD_CONFIG_DEFAULTS} + _omit_none = True + _required = ("card_id",) + _reject_none_from_dict = True + _field_kinds = {"card_id": "int", **DeckRecommendCardConfig._field_kinds} class DeckRecommendSaOptions(_CompatModel): - _defaults = { - "run_num": None, - "seed": None, - "max_iter": None, - "max_no_improve_iter": None, - "time_limit_ms": None, - "start_temprature": None, - "cooling_rate": None, - "debug": None, + __slots__ = ( + "run_num", + "seed", + "max_iter", + "max_no_improve_iter", + "time_limit_ms", + "start_temprature", + "cooling_rate", + "debug", + ) + _defaults = dict.fromkeys(__slots__) + _omit_none = True + _reject_none_from_dict = True + _field_kinds = { + **{name: _optional("int") for name in ( + "run_num", "seed", "max_iter", "max_no_improve_iter", "time_limit_ms" + )}, + **{name: _optional("float") for name in ("start_temprature", "cooling_rate")}, + "debug": _optional("bool"), } class DeckRecommendGaOptions(_CompatModel): - _defaults = { - "seed": None, - "debug": None, - "max_iter": None, - "max_no_improve_iter": None, - "pop_size": None, - "parent_size": None, - "elite_size": None, - "crossover_rate": None, - "base_mutation_rate": None, - "no_improve_iter_to_mutation_rate": None, + __slots__ = ( + "seed", + "debug", + "max_iter", + "max_no_improve_iter", + "pop_size", + "parent_size", + "elite_size", + "crossover_rate", + "base_mutation_rate", + "no_improve_iter_to_mutation_rate", + ) + _defaults = dict.fromkeys(__slots__) + _omit_none = True + _reject_none_from_dict = True + _field_kinds = { + **{name: _optional("int") for name in ( + "seed", "max_iter", "max_no_improve_iter", "pop_size", "parent_size", "elite_size" + )}, + **{name: _optional("float") for name in ( + "crossover_rate", "base_mutation_rate", "no_improve_iter_to_mutation_rate" + )}, + "debug": _optional("bool"), } class DeckRecommendOptions(_CompatModel): + __slots__ = ( + "target", + "algorithm", + "region", + "user_data", + "user_data_file_path", + "user_data_str", + "live_type", + "music_id", + "music_diff", + "event_id", + "event_attr", + "event_unit", + "event_type", + "world_bloom_event_turn", + "world_bloom_character_id", + "challenge_live_character_id", + "limit", + "member", + "timeout_ms", + "rarity_1_config", + "rarity_2_config", + "rarity_3_config", + "rarity_birthday_config", + "rarity_4_config", + "single_card_configs", + "support_master_max", + "support_skill_max", + "filter_other_unit", + "fixed_cards", + "fixed_characters", + "forcedLeaderCharacterId", + "target_bonus_list", + "custom_bonus_character_ids", + "custom_bonus_attr", + "custom_bonus_character_support_units", + "skill_reference_choose_strategy", + "keep_after_training_state", + "multi_live_teammate_score_up", + "multi_live_teammate_power", + "best_skill_as_leader", + "multi_live_score_up_lower_bound", + "skill_order_choose_strategy", + "specific_skill_order", + "sa_options", + "ga_options", + ) + _defaults = dict.fromkeys(__slots__) + _nested = { + "rarity_1_config": (DeckRecommendCardConfig, False), + "rarity_2_config": (DeckRecommendCardConfig, False), + "rarity_3_config": (DeckRecommendCardConfig, False), + "rarity_birthday_config": (DeckRecommendCardConfig, False), + "rarity_4_config": (DeckRecommendCardConfig, False), + "single_card_configs": (DeckRecommendSingleCardConfig, True), + "sa_options": (DeckRecommendSaOptions, False), + "ga_options": (DeckRecommendGaOptions, False), + } + _omit_none = True _native_key_aliases = { "fixed_cards": "fixedCards", "fixed_characters": "fixedCharacters", @@ -131,80 +303,78 @@ class DeckRecommendOptions(_CompatModel): "multi_live_teammate_power": "multiLiveTeammatePower", } _algorithm_aliases = {"dfs", "ga", "dfs_ga", "rl", "sa"} - _defaults = { - "target": "score", - "algorithm": "dfs", - "region": "jp", - "user_data": None, - "user_data_file_path": None, - "user_data_str": None, - "live_type": "solo", - "music_id": None, - "music_diff": "master", - "event_id": None, - "event_attr": None, - "event_unit": None, - "event_type": None, - "world_bloom_event_turn": None, - "world_bloom_character_id": None, - "challenge_live_character_id": None, - "limit": 10, - "member": None, - "timeout_ms": None, - "rarity_1_config": None, - "rarity_2_config": None, - "rarity_3_config": None, - "rarity_birthday_config": None, - "rarity_4_config": None, - "single_card_configs": None, - "support_master_max": None, - "support_skill_max": None, - "filter_other_unit": None, - "fixed_cards": None, - "fixed_characters": None, - "forcedLeaderCharacterId": None, - "target_bonus_list": None, - "custom_bonus_character_ids": None, - "custom_bonus_attr": None, - "custom_bonus_character_support_units": None, - "skill_reference_choose_strategy": None, - "keep_after_training_state": None, - "multi_live_teammate_score_up": None, - "multi_live_teammate_power": None, - "best_skill_as_leader": None, - "multi_live_score_up_lower_bound": None, - "skill_order_choose_strategy": None, - "specific_skill_order": None, - "sa_options": None, - "ga_options": None, - } - _nested = { - "rarity_1_config": (DeckRecommendCardConfig, False), - "rarity_2_config": (DeckRecommendCardConfig, False), - "rarity_3_config": (DeckRecommendCardConfig, False), - "rarity_birthday_config": (DeckRecommendCardConfig, False), - "rarity_4_config": (DeckRecommendCardConfig, False), - "single_card_configs": (DeckRecommendSingleCardConfig, True), - "sa_options": (DeckRecommendSaOptions, False), - "ga_options": (DeckRecommendGaOptions, False), + _reject_none_from_dict = True + _from_dict_ignored = ("user_data",) + _field_kinds = { + **{name: _optional("str") for name in ( + "target", "algorithm", "region", "user_data_file_path", "user_data_str", + "live_type", "music_diff", "event_attr", "event_unit", "event_type", + "custom_bonus_attr", "skill_reference_choose_strategy", "skill_order_choose_strategy", + )}, + **{name: _optional("int") for name in ( + "music_id", "event_id", "world_bloom_event_turn", "world_bloom_character_id", + "challenge_live_character_id", "limit", "member", "timeout_ms", + "forcedLeaderCharacterId", "multi_live_teammate_score_up", "multi_live_teammate_power", + )}, + **{name: _optional("bool") for name in ( + "support_master_max", "support_skill_max", "filter_other_unit", + "keep_after_training_state", "best_skill_as_leader", + )}, + "multi_live_score_up_lower_bound": _optional("float"), + "user_data": _optional(("model", "DeckRecommendUserData")), + **{name: _optional(("model", "DeckRecommendCardConfig")) for name in ( + "rarity_1_config", "rarity_2_config", "rarity_3_config", + "rarity_birthday_config", "rarity_4_config", + )}, + "single_card_configs": _optional(("list", ("model", "DeckRecommendSingleCardConfig"))), + **{name: _optional(("list", "int")) for name in ( + "fixed_cards", "fixed_characters", "target_bonus_list", + "custom_bonus_character_ids", "specific_skill_order", + )}, + "custom_bonus_character_support_units": _optional("int_str_dict"), + "sa_options": _optional(("model", "DeckRecommendSaOptions")), + "ga_options": _optional(("model", "DeckRecommendGaOptions")), } - def to_native_dict(self) -> dict[str, Any]: + def to_dict(self) -> dict[str, Any]: + if self.user_data is not None: + raise RuntimeError("Cannot be converted to dict when user_data is set.") + return self._as_dict() + + def _to_native_dict(self) -> dict[str, Any]: if self.member not in (None, 5): raise ValueError("allium-deck only supports 5-member decks") - result = _drop_none(self.to_dict()) - algorithm = str(self.algorithm).strip().lower() + algorithm = "ga" if self.algorithm is None else str(self.algorithm).strip().lower() if algorithm not in self._algorithm_aliases: raise ValueError(f"unsupported algorithm: {self.algorithm}") + result = self._as_dict(include_user_data=False) for source, target in self._native_key_aliases.items(): if source in result: result[target] = result.pop(source) result["algorithm"] = "dfs" - result.pop("user_data", None) + result.setdefault("skillOrderChooseStrategy", "average") + if "event_id" not in result and self.live_type not in ("challenge", "challenge_auto"): + result.setdefault("event_type", "marathon") return result class RecommendCard(_CompatModel): + __slots__ = ( + "card_id", + "total_power", + "base_power", + "event_bonus_rate", + "master_rank", + "level", + "skill_level", + "skill_score_up", + "skill_life_recovery", + "episode1_read", + "episode2_read", + "after_training", + "default_image", + "has_canvas_bonus", + ) _defaults = { "card_id": 0, "total_power": 0, @@ -213,29 +383,77 @@ class RecommendCard(_CompatModel): "master_rank": 0, "level": 0, "skill_level": 0, - "skill_score_up": 0.0, - "skill_life_recovery": 0.0, - "episode1_read": None, - "episode2_read": None, + "skill_score_up": 0, + "skill_life_recovery": 0, + "episode1_read": False, + "episode2_read": False, "after_training": False, - "default_image": "original", + "default_image": "", "has_canvas_bonus": False, } + _required = tuple(_defaults) + _field_kinds = { + **{name: "int" for name in ( + "card_id", "total_power", "base_power", "master_rank", "level", "skill_level", + "skill_score_up", "skill_life_recovery", + )}, + "event_bonus_rate": "float", + **{name: "bool" for name in ( + "episode1_read", "episode2_read", "after_training", "has_canvas_bonus" + )}, + "default_image": "str", + } class RecommendSupportDeckCard(_CompatModel): + __slots__ = ( + "card_id", + "bonus", + "skill_level", + "master_rank", + "level", + "after_training", + "default_image", + ) _defaults = { "card_id": 0, "bonus": 0.0, - "skill_level": 0, + "skill_level": 1, "master_rank": 0, - "level": 0, + "level": 1, "after_training": False, - "default_image": "original", + "default_image": "", + } + _required = ("card_id", "bonus") + _field_kinds = { + "card_id": "int", + "bonus": "float", + "skill_level": "int", + "master_rank": "int", + "level": "int", + "after_training": "bool", + "default_image": "str", } class RecommendDeck(_CompatModel): + __slots__ = ( + "score", + "live_score", + "mysekai_event_point", + "total_power", + "base_power", + "area_item_bonus_power", + "character_bonus_power", + "honor_bonus_power", + "fixture_bonus_power", + "gate_bonus_power", + "event_bonus_rate", + "support_deck_bonus_rate", + "multi_live_score_up", + "support_deck_cards", + "cards", + ) _defaults = { "score": 0, "live_score": 0, @@ -257,31 +475,94 @@ class RecommendDeck(_CompatModel): "cards": (RecommendCard, True), "support_deck_cards": (RecommendSupportDeckCard, True), } + _required = ( + "score", + "live_score", + "mysekai_event_point", + "total_power", + "base_power", + "area_item_bonus_power", + "character_bonus_power", + "honor_bonus_power", + "fixture_bonus_power", + "gate_bonus_power", + "event_bonus_rate", + "support_deck_bonus_rate", + "multi_live_score_up", + "cards", + ) + _field_kinds = { + **{name: "int" for name in ( + "score", "live_score", "mysekai_event_point", "total_power", "base_power", + "area_item_bonus_power", "character_bonus_power", "honor_bonus_power", + "fixture_bonus_power", "gate_bonus_power", + )}, + **{name: "float" for name in ( + "event_bonus_rate", "support_deck_bonus_rate", "multi_live_score_up" + )}, + "support_deck_cards": ("list", ("model", "RecommendSupportDeckCard")), + "cards": ("list", ("model", "RecommendCard")), + } class DeckRecommendResult(_CompatModel): + __slots__ = ("decks", "cost_ms") _defaults = {"decks": [], "cost_ms": 0.0} _nested = {"decks": (RecommendDeck, True)} + _required = ("decks",) + _field_kinds = {"decks": ("list", ("model", "RecommendDeck")), "cost_ms": "float"} class DeckRecommendUserData: - def __init__(self) -> None: - self._raw: bytes | None = None - self._native = None + __slots__ = ("_raw", "_native") + + def __init__(self, source: DeckRecommendUserData | None = None) -> None: + if source is not None and not isinstance(source, DeckRecommendUserData): + raise TypeError("expected DeckRecommendUserData") + self._raw = None if source is None else source._raw + self._native = None if source is None else source._native def load_from_file(self, path: str) -> None: - self.load_from_bytes(Path(path).read_bytes()) + if not isinstance(path, str): + raise TypeError("path must be a str") + try: + raw = Path(path).read_bytes() + except OSError as error: + raise RuntimeError(f"Failed to load user data from file: {path}") from error + self.load_from_bytes(raw) def load_from_bytes(self, data: str | bytes) -> None: - raw = data.encode("utf-8") if isinstance(data, str) else bytes(data) - json.loads(raw) + if isinstance(data, str): + raw = data.encode("utf-8") + elif isinstance(data, (bytes, bytearray)): + raw = bytes(data) + else: + raise TypeError("data must be str or bytes") + try: + json.loads(raw) + except (UnicodeDecodeError, json.JSONDecodeError) as error: + raise RuntimeError("Failed to load user data from bytes") from error self._raw = raw try: from _allium_deck_native import NativeUserData except ImportError: self._native = None else: - self._native = NativeUserData(raw) - - -__all__ = [name for name in globals() if name.startswith("DeckRecommend") or name.startswith("Recommend")] + try: + self._native = NativeUserData(raw) + except (TypeError, ValueError) as error: + raise RuntimeError("Failed to load user data from bytes") from error + + +__all__ = [ + "DeckRecommendCardConfig", + "DeckRecommendGaOptions", + "DeckRecommendOptions", + "DeckRecommendResult", + "DeckRecommendSaOptions", + "DeckRecommendSingleCardConfig", + "DeckRecommendUserData", + "RecommendCard", + "RecommendDeck", + "RecommendSupportDeckCard", +] diff --git a/python/sekai_deck_recommend_cpp/py.typed b/python/sekai_deck_recommend_cpp/py.typed new file mode 100644 index 0000000..8b13789 --- /dev/null +++ b/python/sekai_deck_recommend_cpp/py.typed @@ -0,0 +1 @@ + diff --git a/src/lib.rs b/src/lib.rs index c9aa7c4..ff7f876 100644 --- a/src/lib.rs +++ b/src/lib.rs @@ -6,22 +6,30 @@ use std::time::Instant; use allium_deck::engine::{ parse_build_params_json, parse_user_profile_json, MasterdataSources, OwnedGameData, }; +use allium_deck::auxiliary::{ + recommend_music, AuxiliaryData, MusicDeck, MusicDeckCard, MusicRecommendOptions, +}; use allium_deck::handler::{ - build_card_pool_with_details, cultivated_user_cards, FullPrecisionCard, UserCard, UserProfile, + build_card_pool_with_details, cultivated_user_cards, world_bloom_support_cards, + FullPrecisionCard, UserCard, UserProfile, }; use allium_deck::pool::{CardIdx, CardPool}; use allium_deck::search::{ search, search_bonus_targets, summarize_deck, SearchContext, SearchParams, }; -use allium_deck::{DefaultImage, PowerDetail, ScoreTarget, DECK_SIZE}; +use allium_deck::{DefaultImage, EventType, LiveType, PowerDetail, ScoreTarget, DECK_SIZE}; use pyo3::exceptions::{PyRuntimeError, PyValueError}; use pyo3::prelude::*; use serde_json::{json, Value}; +const MYSEKAI_EVENT_POINT: i32 = 2_500; + struct RegionData { tables: BTreeMap, music_metas: String, + loaded_music_metas: Arc>, game: Arc, + auxiliary: Arc, } #[pyclass] @@ -90,13 +98,19 @@ impl NativeEngine { PyRuntimeError::new_err(format!("masterdata for region {region} is not loaded")) })?; let tables = current.tables.clone(); + let loaded_music_metas = Arc::new( + parse_loaded_music_metas(music_metas).map_err(PyValueError::new_err)?, + ); let game = Arc::new(build_game(&tables, music_metas).map_err(PyValueError::new_err)?); + let auxiliary = Arc::clone(¤t.auxiliary); regions.insert( region.to_string(), RegionData { tables, music_metas: music_metas.to_string(), + loaded_music_metas, game, + auxiliary, }, ); Ok(()) @@ -124,28 +138,369 @@ impl NativeEngine { py.allow_threads(move || recommend_json(&user, &game, ¶ms)) .map_err(PyRuntimeError::new_err) } + + fn get_world_bloom_support_cards( + &self, + py: Python<'_>, + region: &str, + options_json: &str, + user_data: &NativeUserData, + ) -> PyResult { + let game = self.region_game(region)?; + let mut params = parse_build_params_json(options_json) + .map_err(|error| PyValueError::new_err(error.to_string()))?; + let raw: Value = serde_json::from_str(options_json) + .map_err(|error| PyValueError::new_err(error.to_string()))?; + if params.world_bloom_character_id.is_none() { + params.world_bloom_character_id = raw + .get("forcedLeaderCharacterId") + .and_then(Value::as_i64) + .and_then(|value| i32::try_from(value).ok()); + } + let support_master_max = raw + .get("support_master_max") + .or_else(|| raw.get("supportMasterMax")) + .and_then(Value::as_bool) + .unwrap_or(false); + let support_skill_max = raw + .get("support_skill_max") + .or_else(|| raw.get("supportSkillMax")) + .and_then(Value::as_bool) + .unwrap_or(false); + let filter_other_unit = raw + .get("filter_other_unit") + .or_else(|| raw.get("filterOtherUnit")) + .and_then(Value::as_bool) + .unwrap_or(false); + let user = Arc::clone(&user_data.profile); + py.allow_threads(move || { + let game_ref = game.as_ref().as_ref(); + let cards = world_bloom_support_cards( + &user, + &game_ref, + ¶ms, + support_master_max, + support_skill_max, + filter_other_unit, + ) + .map_err(|error| error.to_string())?; + let output = cards + .into_iter() + .map(|card| { + json!({ + "card_id": card.card_id, + "bonus": card.bonus, + "skill_level": card.skill_level, + "master_rank": card.master_rank, + "level": card.level, + "after_training": card.after_training, + "default_image": default_image(card.default_image), + }) + }) + .collect::>(); + serde_json::to_string(&output).map_err(|error| error.to_string()) + }) + .map_err(PyRuntimeError::new_err) + } + + fn recommend_area_items( + &self, + py: Python<'_>, + region: &str, + card_ids: Vec, + user_data: &NativeUserData, + ) -> PyResult { + let (game, auxiliary) = self.region_calculators(region)?; + let user = Arc::clone(&user_data.profile); + py.allow_threads(move || { + let game_ref = game.as_ref().as_ref(); + let result = auxiliary.recommend_area_items(&user, &game_ref, &card_ids)?; + serde_json::to_string(&result).map_err(|error| error.to_string()) + }) + .map_err(PyRuntimeError::new_err) + } + + fn recommend_music( + &self, + py: Python<'_>, + region: &str, + options_json: &str, + deck_json: &str, + ) -> PyResult { + let (game, loaded_music_metas) = self.region_music(region)?; + let params = parse_build_params_json(options_json) + .map_err(|error| PyValueError::new_err(error.to_string()))?; + let deck_value: Value = serde_json::from_str(deck_json) + .map_err(|error| PyValueError::new_err(error.to_string()))?; + let deck = music_deck_from_json(&deck_value).map_err(PyValueError::new_err)?; + let event_type = resolve_event_type(&game, ¶ms).map_err(PyValueError::new_err)?; + let live_type = if matches!(params.live_type, LiveType::Multi) + && matches!(event_type, EventType::CheerfulCarnival) + { + LiveType::Cheerful + } else if matches!(params.live_type, LiveType::Mysekai) { + LiveType::Multi + } else { + params.live_type + }; + let options = MusicRecommendOptions { + live_type, + event_type, + skill_order: params.live_skill_order, + specific_skill_order: params + .specific_skill_order + .map(|order| order.into_iter().collect()), + multi_teammate_score_up: params.multi_teammate_score_up, + multi_teammate_power: params.multi_teammate_power, + }; + py.allow_threads(move || { + let result = recommend_music(&loaded_music_metas, &deck, &options)?; + serde_json::to_string(&result).map_err(|error| error.to_string()) + }) + .map_err(PyRuntimeError::new_err) + } + + #[allow(clippy::too_many_arguments)] + #[pyo3(signature = ( + region, + power, + skills, + live_type, + music_score_json, + multi_sum_power = 0, + fever_music_score_json = None + ))] + fn calculate_exact_live( + &self, + py: Python<'_>, + region: &str, + power: i32, + skills: Vec, + live_type: &str, + music_score_json: &str, + multi_sum_power: i32, + fever_music_score_json: Option<&str>, + ) -> PyResult { + let (_, auxiliary) = self.region_calculators(region)?; + let live_type = parse_live_type(live_type).map_err(PyValueError::new_err)?; + py.allow_threads(move || { + let result = auxiliary.calculate_exact_live( + power, + &skills, + live_type, + music_score_json, + multi_sum_power, + fever_music_score_json, + )?; + serde_json::to_string(&result).map_err(|error| error.to_string()) + }) + .map_err(PyRuntimeError::new_err) + } } impl NativeEngine { + fn region_game(&self, region: &str) -> PyResult> { + self.regions + .read() + .map_err(lock_error)? + .get(region) + .map(|data| Arc::clone(&data.game)) + .ok_or_else(|| { + PyRuntimeError::new_err(format!("masterdata for region {region} is not loaded")) + }) + } + + fn region_calculators( + &self, + region: &str, + ) -> PyResult<(Arc, Arc)> { + self.regions + .read() + .map_err(lock_error)? + .get(region) + .map(|data| (Arc::clone(&data.game), Arc::clone(&data.auxiliary))) + .ok_or_else(|| { + PyRuntimeError::new_err(format!("masterdata for region {region} is not loaded")) + }) + } + + fn region_music( + &self, + region: &str, + ) -> PyResult<(Arc, Arc>)> { + self.regions + .read() + .map_err(lock_error)? + .get(region) + .map(|data| (Arc::clone(&data.game), Arc::clone(&data.loaded_music_metas))) + .ok_or_else(|| { + PyRuntimeError::new_err(format!("masterdata for region {region} is not loaded")) + }) + } + fn replace_tables(&self, region: &str, tables: BTreeMap) -> PyResult<()> { let mut regions = self.regions.write().map_err(lock_error)?; let music_metas = regions .get(region) .map(|data| data.music_metas.clone()) .unwrap_or_else(|| "[]".to_string()); + let loaded_music_metas = Arc::new( + parse_loaded_music_metas(&music_metas).map_err(PyValueError::new_err)?, + ); let game = Arc::new(build_game(&tables, &music_metas).map_err(PyValueError::new_err)?); + let auxiliary = Arc::new( + AuxiliaryData::from_strings(&tables).map_err(PyValueError::new_err)?, + ); regions.insert( region.to_string(), RegionData { tables, music_metas, + loaded_music_metas, game, + auxiliary, }, ); Ok(()) } } +fn parse_loaded_music_metas(text: &str) -> Result, String> { + let rows: Vec = + serde_json::from_str(text).map_err(|error| format!("invalid music metas: {error}"))?; + rows.iter() + .map(|row| { + let event_rate = value_i32(row, "event_rate")?; + Ok(allium_deck::handler::MusicMeta { + music_id: value_i32(row, "music_id")?, + difficulty: row + .get("difficulty") + .and_then(Value::as_str) + .unwrap_or_default() + .to_string(), + event_rate_solo: event_rate, + event_rate_multi: event_rate, + event_rate_auto: event_rate, + base_score: value_f64(row, "base_score")?, + base_score_auto: value_f64(row, "base_score_auto")?, + fever_score: value_f64(row, "fever_score")?, + solo_skill_scores: value_f64_array(row, "skill_score_solo")?, + multi_skill_scores: value_f64_array(row, "skill_score_multi")?, + auto_skill_scores: value_f64_array(row, "skill_score_auto")?, + music_time: value_f64(row, "music_time")?, + tap_count: value_i32(row, "tap_count")?, + }) + }) + .collect() +} + +fn value_i32(value: &Value, key: &str) -> Result { + value + .get(key) + .and_then(Value::as_i64) + .and_then(|value| i32::try_from(value).ok()) + .ok_or_else(|| format!("music meta field {key} is required")) +} + +fn value_f64(value: &Value, key: &str) -> Result { + value + .get(key) + .and_then(Value::as_f64) + .ok_or_else(|| format!("music meta field {key} is required")) +} + +fn value_f64_array(value: &Value, key: &str) -> Result<[f64; 6], String> { + let values = value + .get(key) + .and_then(Value::as_array) + .ok_or_else(|| format!("music meta field {key} is required"))?; + if values.len() != 6 { + return Err(format!("music meta field {key} must contain 6 values")); + } + let mut result = [0.0; 6]; + for (index, item) in values.iter().enumerate() { + result[index] = item + .as_f64() + .ok_or_else(|| format!("music meta field {key}[{index}] must be numeric"))?; + } + Ok(result) +} + +fn music_deck_from_json(value: &Value) -> Result { + let cards = value + .get("cards") + .and_then(Value::as_array) + .ok_or_else(|| "deck.cards is required".to_string())? + .iter() + .map(|card| MusicDeckCard { + skill_score_up: card + .get("skill_score_up") + .and_then(Value::as_f64) + .unwrap_or(0.0), + skill_life_recovery: card + .get("skill_life_recovery") + .and_then(Value::as_f64) + .unwrap_or(0.0), + }) + .collect::>(); + Ok(MusicDeck { + total_power: json_i32(value, "total_power")?, + event_bonus_rate: value + .get("event_bonus_rate") + .and_then(Value::as_f64) + .unwrap_or(0.0), + support_deck_bonus_rate: value + .get("support_deck_bonus_rate") + .and_then(Value::as_f64) + .unwrap_or(0.0), + cards, + }) +} + +fn json_i32(value: &Value, key: &str) -> Result { + value + .get(key) + .and_then(Value::as_i64) + .and_then(|value| i32::try_from(value).ok()) + .ok_or_else(|| format!("deck.{key} is required")) +} + +fn resolve_event_type( + game: &OwnedGameData, + params: &allium_deck::handler::BuildParams, +) -> Result { + let value = params.event_id.and_then(|event_id| { + game.events + .iter() + .find(|event| event.id == event_id) + .map(|event| event.event_type.as_str()) + }); + parse_event_type(value.or(params.event_type.as_deref()).unwrap_or("marathon")) +} + +fn parse_event_type(value: &str) -> Result { + match value.trim().to_ascii_lowercase().as_str() { + "marathon" => Ok(EventType::Marathon), + "cheerful" | "cheerful_carnival" | "cheerfulcarnival" => { + Ok(EventType::CheerfulCarnival) + } + "world_bloom" | "worldbloom" | "wl" => Ok(EventType::WorldBloom), + _ => Err(format!("invalid event type: {value}")), + } +} + +fn parse_live_type(value: &str) -> Result { + match value.trim().to_ascii_lowercase().as_str() { + "solo" => Ok(LiveType::Solo), + "auto" => Ok(LiveType::Auto), + "multi" => Ok(LiveType::Multi), + "cheerful" | "cheerful_live" => Ok(LiveType::Cheerful), + "challenge" => Ok(LiveType::Challenge), + "challenge_auto" => Ok(LiveType::ChallengeAuto), + _ => Err(format!("invalid live type: {value}")), + } +} + fn build_game( tables: &BTreeMap, music_metas: &str, @@ -173,6 +528,12 @@ fn recommend_json( search_bonus_targets(&pool, &ctx, &search_params, ¶ms.target_bonus_list).0 }; let cost_ms = search_started.elapsed().as_secs_f64() * 1000.0; + if results.is_empty() && params.target_bonus_list.is_empty() { + return Err(format!( + "Cannot recommend any deck in {} cards", + user.user_cards.len() + )); + } let cultivated = cultivated_user_cards(user, &game, params) .into_iter() .map(|card| (card.card_id, card)) @@ -182,7 +543,6 @@ fn recommend_json( .filter_map(|result| { materialize_deck( result.cards, - result.score, &pool, &ctx, &details, @@ -202,7 +562,6 @@ fn recommend_json( #[allow(clippy::too_many_arguments)] fn materialize_deck( deck: [CardIdx; DECK_SIZE], - target_score: u64, pool: &CardPool, ctx: &SearchContext, details: &[FullPrecisionCard], @@ -232,7 +591,7 @@ fn materialize_deck( "skill_level": user_card.map_or(0, |card| card.skill_level), "master_rank": user_card.map_or(0, |card| card.master_rank), "level": user_card.map_or(0, |card| card.level), - "after_training": user_card.is_some_and(|card| card.special_training_status != "none"), + "after_training": user_card.is_some_and(user_card_after_training), "default_image": user_card.map_or("original", |card| card.default_image.as_str()), })); } @@ -259,12 +618,19 @@ fn materialize_deck( "master_rank": detail.master_rank, "level": user_card.map_or(0, |card| card.level), "skill_level": detail.skill_level, - "skill_score_up": summary.card_skill_score_up[position], - "skill_life_recovery": detail.skill.life_recovery, + "skill_score_up": summary.card_skill_score_up[position] as i32, + "skill_life_recovery": detail.skill.life_recovery as i32, "episode1_read": episodes >= 1, "episode2_read": episodes >= 2, - "after_training": ctx.skill_is_after_training_at(dense), - "default_image": default_image(detail.default_image), + "after_training": detail.after_training, + "default_image": if detail.skill_state_controls_image { + default_image(detail.default_image) + } else { + user_card.map_or_else( + || default_image(detail.default_image), + |card| card.default_image.as_str(), + ) + }, "has_canvas_bonus": has_canvas, }) }) @@ -289,15 +655,21 @@ fn materialize_deck( .sum::(); let total_bonus = summary.event_bonus_total.unwrap_or(0.0); let event_bonus = (total_bonus - support_bonus).max(0.0); - let mysekai_event_point = if matches!(target, ScoreTarget::Mysekai) { - target_score.min(i32::MAX as u64) as i32 + let mysekai_event_point = if matches!(ctx.live_type, LiveType::Mysekai) + || matches!(target, ScoreTarget::Mysekai) + { + MYSEKAI_EVENT_POINT } else { 0 }; - let score = summary.event_point.unwrap_or(summary.live_score); + let score = if matches!(target, ScoreTarget::Mysekai) { + 0 + } else { + summary.event_point.unwrap_or(summary.live_score) + }; Some(json!({ - "score": if matches!(target, ScoreTarget::Mysekai) { mysekai_event_point } else { score }, + "score": score, "live_score": summary.live_score, "mysekai_event_point": mysekai_event_point, "total_power": summary.total_power, @@ -309,12 +681,26 @@ fn materialize_deck( "gate_bonus_power": gate_bonus_power, "event_bonus_rate": event_bonus, "support_deck_bonus_rate": support_bonus, - "multi_live_score_up": summary.multi_live_score_up, + "multi_live_score_up": if matches!(ctx.live_type, LiveType::Mysekai) { + 0.0 + } else { + summary.multi_live_score_up + }, "support_deck_cards": support_cards, "cards": cards, })) } +fn user_card_after_training(card: &UserCard) -> bool { + matches!( + card.special_training_status + .trim() + .to_ascii_lowercase() + .as_str(), + "done" | "special_training" | "trained" | "after_training" + ) +} + fn resolve_power_details( pool: &CardPool, details: &[FullPrecisionCard], diff --git a/tests/test_full_api_integration.py b/tests/test_full_api_integration.py new file mode 100644 index 0000000..c269d20 --- /dev/null +++ b/tests/test_full_api_integration.py @@ -0,0 +1,233 @@ +import json +import os +from pathlib import Path + +import pytest + +from sekai_deck_recommend_cpp import ( + DeckRecommendOptions, + DeckRecommendUserData, + RecommendDeck, + SekaiDeckRecommend, + set_engine_thread_count, +) + + +def _fixture_paths(): + keys = ("ALLIUM_MASTERDATA", "ALLIUM_MUSIC_METAS", "ALLIUM_USER_DATA") + if not all(os.environ.get(key) for key in keys): + pytest.skip("native integration fixture paths are not configured") + return tuple(Path(os.environ[key]) for key in keys) + + +def _ready_engine_and_options(): + masterdata, music_metas, user_path = _fixture_paths() + engine = SekaiDeckRecommend() + engine.update_masterdata(str(masterdata), "cn") + engine.update_musicmetas(str(music_metas), "cn") + user = DeckRecommendUserData() + user.load_from_file(str(user_path)) + options = DeckRecommendOptions.from_dict( + { + "region": "cn", + "algorithm": "dfs", + "live_type": "multi", + "event_id": 133, + "music_id": 1, + "music_diff": "master", + "target": "score", + "limit": 1, + } + ) + options.user_data = user + return engine, options, masterdata, music_metas + + +def test_batch_support_area_and_music_methods_execute_with_loaded_data(): + engine, options, _, music_metas = _ready_engine_and_options() + result = engine.recommend(options) + assert result.cost_ms >= 0.0 + assert result.to_dict()["cost_ms"] == result.cost_ms + deck = result.decks[0] + + set_engine_thread_count(2) + batch = engine.recommend_batch([DeckRecommendOptions(options), DeckRecommendOptions(options)]) + assert len(batch) == 2 + assert all(item.cost_ms >= 0.0 for item in batch) + assert [card.card_id for card in batch[0].decks[0].cards] == [ + card.card_id for card in batch[1].decks[0].cards + ] + + support_options = DeckRecommendOptions(options) + support_options.event_id = None + support_options.world_bloom_event_turn = 1 + support_options.world_bloom_character_id = 1 + support = engine.get_world_bloom_support_cards(support_options) + assert support + assert all( + (left.bonus, -left.card_id) >= (right.bonus, -right.card_id) + for left, right in zip(support, support[1:]) + ) + + card_ids = [card.card_id for card in deck.cards] + area_items = engine.recommend_area_items(options, card_ids) + assert all( + { + "area_id", + "area_type", + "area_view_type", + "area_item_id", + "next_level", + "shop_item_id", + "cost", + "power", + "power_per_coin", + } + <= item.keys() + for item in area_items + ) + + music = engine.recommend_music(options, deck) + expected_rows = len(json.loads(music_metas.read_text("utf-8"))) + assert len(music) == expected_rows + assert all(item["event_point"] is not None for item in music) + assert all( + (left["event_point"], left["live_score"]) + >= (right["event_point"], right["live_score"]) + for left, right in zip(music, music[1:]) + ) + + +def test_exact_live_returns_complete_note_details(): + engine, _, masterdata, _ = _ready_engine_and_options() + notes = json.loads((masterdata / "ingameNotes.json").read_text("utf-8")) + note_type = notes[0]["id"] + score = json.dumps( + { + "notes": [ + {"time": 1.0, "type": note_type, "longId": 0}, + {"time": 7.0, "type": note_type, "longId": 0}, + ], + "skills": [{"time": 1.0}], + "fevers": [], + } + ) + detail = engine.calculate_exact_live( + "cn", 100_000, [100.0], "multi", score, multi_sum_power=500_000 + ) + assert set(detail) == {"total", "active_bonus", "notes"} + assert detail["active_bonus"] == pytest.approx(37_500.0) + assert len(detail["notes"]) == 2 + assert set(detail["notes"][0]) == { + "note_coefficient", + "combo_coefficient", + "judge_coefficient", + "effect_bonuses", + "score", + } + + +def test_auxiliary_methods_validate_public_inputs(): + engine = SekaiDeckRecommend() + options = DeckRecommendOptions() + with pytest.raises(ValueError, match="region"): + engine.recommend_music(options, RecommendDeck()) + with pytest.raises(ValueError, match="region"): + engine.calculate_exact_live("invalid", 1, [], "solo", "{}") + + +def test_native_sequence_arguments_accept_sequence_and_set_inputs(): + engine, options, masterdata, _ = _ready_engine_and_options() + result = engine.recommend(options) + card_ids = [card.card_id for card in result.decks[0].cards] + assert len(engine.recommend_batch((options,))) == 1 + assert engine.recommend_area_items(options, tuple(card_ids)) + + notes = json.loads((masterdata / "ingameNotes.json").read_text("utf-8")) + score = json.dumps( + { + "notes": [{"time": 1.0, "type": notes[0]["id"], "longId": 0}], + "skills": [{"time": 1.0}], + "fevers": [], + } + ) + assert engine.calculate_exact_live("cn", 100_000, (100.0,), "multi", score) + with pytest.raises(TypeError): + engine.calculate_exact_live("cn", 100_000, iter([100.0]), "multi", score) + + +def test_non_trainable_fixed_card_reports_original_training_state(): + engine, options, _, _ = _ready_engine_and_options() + fixed = DeckRecommendOptions(options) + fixed.fixed_cards = [1] + card = next(card for card in engine.recommend(fixed).decks[0].cards if card.card_id == 1) + assert card.after_training is False + assert card.default_image == "original" + + +def test_skill_state_can_select_original_art_for_a_trained_card(): + engine, options, _, _ = _ready_engine_and_options() + world_bloom = DeckRecommendOptions(options) + world_bloom.event_id = 179 + world_bloom.world_bloom_character_id = 23 + world_bloom.fixed_cards = [617, 761, 798, 953, 1145] + card = next( + card + for card in engine.recommend(world_bloom).decks[0].cards + if card.card_id == 1145 + ) + assert card.after_training is True + assert card.default_image == "original" + + +@pytest.mark.parametrize("target", ["score", "mysekai"]) +def test_mysekai_result_fields_use_the_dedicated_event_point(target): + engine, options, _, _ = _ready_engine_and_options() + mysekai = DeckRecommendOptions(options) + mysekai.live_type = "mysekai" + mysekai.target = target + mysekai.fixed_cards = [162, 664, 666, 938, 939] + + deck = engine.recommend(mysekai).decks[0] + assert deck.mysekai_event_point == 2500 + assert deck.multi_live_score_up == 0.0 + assert deck.score == 0 + + +def test_string_and_bytes_data_loading_match_file_loading(): + file_engine, file_options, masterdata, music_metas = _ready_engine_and_options() + file_options.fixed_cards = [162, 664, 666, 938, 939] + expected = file_engine.recommend(file_options).to_dict() + expected.pop("cost_ms") + + tables = { + path.stem: path.read_bytes() + for path in masterdata.iterdir() + if path.suffix == ".json" + } + memory_engine = SekaiDeckRecommend() + memory_engine.update_masterdata_from_strings(tables, "cn") + memory_engine.update_musicmetas_from_string(music_metas.read_bytes(), "cn") + actual = memory_engine.recommend(file_options).to_dict() + actual.pop("cost_ms") + assert actual == expected + + +def test_semantic_option_errors_are_value_errors(): + engine, options, _, _ = _ready_engine_and_options() + + wrong_target = DeckRecommendOptions(options) + wrong_target.target_bonus_list = [300] + with pytest.raises(ValueError): + engine.recommend(wrong_target) + + solo_bound = DeckRecommendOptions(options) + solo_bound.live_type = "solo" + solo_bound.multi_live_score_up_lower_bound = 100.0 + with pytest.raises(ValueError): + engine.recommend(solo_bound) + + missing_specific = DeckRecommendOptions(options) + missing_specific.skill_order_choose_strategy = "specific" + with pytest.raises(ValueError): + engine.recommend(missing_specific) diff --git a/tests/test_lunabot_contract.py b/tests/test_lunabot_contract.py index 296ea04..cd3ba91 100644 --- a/tests/test_lunabot_contract.py +++ b/tests/test_lunabot_contract.py @@ -1,4 +1,5 @@ import pytest +import sekai_deck_recommend_cpp as api from sekai_deck_recommend_cpp import ( DeckRecommendCardConfig, @@ -20,15 +21,33 @@ def test_lunabot_import_surface_is_available(): assert DeckRecommendUserData is not None -def test_mutable_config_roundtrip_and_unknown_fields(): +def test_public_module_exports_the_complete_deck_api(): + assert set(api.__all__) == { + "DeckRecommendCardConfig", + "DeckRecommendGaOptions", + "DeckRecommendOptions", + "DeckRecommendResult", + "DeckRecommendSaOptions", + "DeckRecommendSingleCardConfig", + "DeckRecommendUserData", + "RecommendCard", + "RecommendDeck", + "RecommendSupportDeckCard", + "SekaiDeckRecommend", + "set_engine_thread_count", + } + + +def test_mutable_config_roundtrip_omits_unset_fields_and_rejects_unknown_fields(): config = DeckRecommendCardConfig() assert config.disable is None + assert config.to_dict() == {} config.level_max = True - config.future_option = 17 + with pytest.raises(AttributeError): + config.future_option = 17 restored = DeckRecommendCardConfig.from_dict(config.to_dict()) assert restored.level_max is True - assert restored.future_option == 17 def test_options_copy_constructor_is_independent_and_preserves_nested_types(): @@ -57,11 +76,11 @@ def test_options_copy_constructor_is_independent_and_preserves_nested_types(): def test_only_five_member_compat_values_are_accepted(member): options = DeckRecommendOptions() options.member = member - assert options.to_native_dict().get("member") in (None, 5) + assert options._to_native_dict().get("member") in (None, 5) def test_native_payload_omits_unset_optional_fields(): - payload = DeckRecommendOptions().to_native_dict() + payload = DeckRecommendOptions()._to_native_dict() assert "timeout_ms" not in payload assert "target_bonus_list" not in payload assert "custom_bonus_character_ids" not in payload @@ -83,7 +102,7 @@ def test_native_payload_maps_lunabot_constraints_to_rust_keys(): } ) - payload = options.to_native_dict() + payload = options._to_native_dict() assert payload["fixedCards"] == [1] assert payload["fixedCharacters"] == [2] assert payload["filterOtherUnit"] is True @@ -102,47 +121,51 @@ def test_other_member_counts_are_rejected_before_native(member): options = DeckRecommendOptions() options.member = member with pytest.raises(ValueError, match="5"): - options.to_native_dict() + options._to_native_dict() @pytest.mark.parametrize("algorithm", ["dfs", "ga", "dfs_ga", "rl", "sa"]) def test_legacy_algorithm_names_normalize_to_dfs(algorithm): options = DeckRecommendOptions() options.algorithm = algorithm - assert options.to_native_dict()["algorithm"] == "dfs" + assert options._to_native_dict()["algorithm"] == "dfs" def test_unknown_algorithm_is_rejected(): options = DeckRecommendOptions() options.algorithm = "typo" with pytest.raises(ValueError, match="algorithm"): - options.to_native_dict() + options._to_native_dict() def test_result_objects_roundtrip_and_remain_mutable(): + card = RecommendCard().to_dict() + card.update( + { + "card_id": 123, + "event_bonus_rate": 70.0, + "skill_level": 4, + "skill_score_up": 100, + "default_image": "special_training", + } + ) + deck = RecommendDeck().to_dict() + deck.update( + { + "score": 1000, + "live_score": 900, + "event_bonus_rate": 250.0, + "support_deck_bonus_rate": 12.5, + "cards": [card], + "support_deck_cards": [ + {"card_id": 456, "bonus": 5.5, "skill_level": 4} + ], + } + ) result = DeckRecommendResult.from_dict( { "cost_ms": 1.25, - "decks": [ - { - "score": 1000, - "live_score": 900, - "event_bonus_rate": 250.0, - "support_deck_bonus_rate": 12.5, - "cards": [ - { - "card_id": 123, - "event_bonus_rate": 70.0, - "skill_level": 4, - "skill_score_up": 100.0, - "default_image": "special_training", - } - ], - "support_deck_cards": [ - {"card_id": 456, "bonus": 5.5, "skill_level": 4} - ], - } - ], + "decks": [deck], } ) @@ -151,3 +174,101 @@ def test_result_objects_roundtrip_and_remain_mutable(): assert isinstance(result.decks[0].support_deck_cards[0], RecommendSupportDeckCard) result.decks = list(result.decks) assert DeckRecommendResult.from_dict(result.to_dict()).to_dict() == result.to_dict() + + +def test_options_defaults_copy_and_serialization_boundary(): + options = DeckRecommendOptions() + assert options.to_dict() == {} + assert options.target is None + assert options.algorithm is None + assert options.region is None + + copied = DeckRecommendOptions(options) + assert copied.to_dict() == {} + + options.user_data = DeckRecommendUserData() + with pytest.raises(RuntimeError, match="user_data"): + options.to_dict() + + +def test_required_result_fields_are_enforced(): + with pytest.raises(KeyError, match="card_id"): + RecommendCard.from_dict({}) + with pytest.raises(KeyError, match="score"): + RecommendDeck.from_dict({}) + with pytest.raises(KeyError, match="decks"): + DeckRecommendResult.from_dict({}) + + +@pytest.mark.parametrize( + ("field", "value"), + [ + ("event_id", 1.5), + ("event_id", "1"), + ("region", 1), + ("support_master_max", "true"), + ("fixed_cards", "1,2"), + ("custom_bonus_character_support_units", {"1": "idol"}), + ], +) +def test_option_properties_reject_values_the_native_contract_cannot_convert(field, value): + options = DeckRecommendOptions() + with pytest.raises(TypeError): + setattr(options, field, value) + + +def test_option_properties_apply_native_compatible_conversions(): + options = DeckRecommendOptions() + options.region = b"cn" + options.support_master_max = 1 + options.fixed_cards = (1, 2) + assert options.region == "cn" + assert options.support_master_max is True + assert options.fixed_cards == [1, 2] + + +def test_result_cost_is_non_negative_and_roundtrips(): + result = DeckRecommendResult.from_dict({"decks": [], "cost_ms": 0.125}) + assert result.cost_ms >= 0 + assert result.to_dict()["cost_ms"] == 0.125 + + +@pytest.mark.parametrize( + ("model", "field"), + [ + (DeckRecommendCardConfig, "level_max"), + (DeckRecommendSingleCardConfig, "card_id"), + (DeckRecommendGaOptions, "pop_size"), + (DeckRecommendSaOptions, "run_num"), + (DeckRecommendOptions, "region"), + ], +) +def test_from_dict_rejects_explicit_none_for_known_fields(model, field): + with pytest.raises(TypeError): + model.from_dict({field: None}) + + +def test_options_from_dict_ignores_user_data_like_the_native_contract(): + user = DeckRecommendUserData() + options = DeckRecommendOptions.from_dict({"user_data": user, "region": "cn"}) + assert options.region == "cn" + assert options.user_data is None + + +def test_internal_payload_conversion_is_not_part_of_the_public_surface(): + assert "to_native_dict" not in dir(DeckRecommendOptions()) + + +def test_user_data_loaders_preserve_public_exception_boundaries(tmp_path): + user = DeckRecommendUserData() + + with pytest.raises(RuntimeError): + user.load_from_bytes("{") + with pytest.raises(TypeError): + user.load_from_bytes(1) + with pytest.raises(TypeError): + user.load_from_bytes(memoryview(b"{}")) + with pytest.raises(RuntimeError): + user.load_from_file(str(tmp_path / "missing.json")) + with pytest.raises(TypeError): + user.load_from_file(tmp_path / "missing.json") diff --git a/tests/test_native_integration.py b/tests/test_native_integration.py index f1fbfe5..988159f 100644 --- a/tests/test_native_integration.py +++ b/tests/test_native_integration.py @@ -80,7 +80,6 @@ def test_native_bonus_target_returns_exact_requested_bucket(): baseline = DeckRecommendOptions.from_dict( { "region": "cn", - "user_data": user, "live_type": "multi", "event_id": 133, "music_id": 1, @@ -89,6 +88,7 @@ def test_native_bonus_target_returns_exact_requested_bucket(): "limit": 1, } ) + baseline.user_data = user maximum = engine.recommend(baseline).decks[0].event_bonus_rate baseline.target_bonus_list = [int(maximum)] exact = engine.recommend(baseline)