From 38b44cdb3d736338387bdc61fbde697c60dd4b60 Mon Sep 17 00:00:00 2001 From: MarTrepodi Date: Thu, 22 Oct 2026 07:26:55 -0400 Subject: [PATCH] feat(helpers): add get_era_units for an Era's unit roster List the base ids of an Era's units from the game data eraDefinition collection: its eraUnitId roster plus its journey unit, with the units it loans to players added on request (include_loaned=True). With no era_id the Era active now is read, its window taken as half-open since consecutive Eras share their boundary millisecond. Co-Authored-By: Claude Opus 5.5 --- docs/api/helpers.md | 38 ++++++++++++ src/swgoh_comlink/helpers/__init__.py | 3 + src/swgoh_comlink/helpers/_eras.py | 80 ++++++++++++++++++++++++++ tests/unit/test_helpers.py | 83 +++++++++++++++++++++++++++ tests/unit/test_helpers_mocked.py | 32 ++++++++++- 5 files changed, 235 insertions(+), 1 deletion(-) create mode 100644 src/swgoh_comlink/helpers/_eras.py diff --git a/docs/api/helpers.md b/docs/api/helpers.md index d408e21..c264677 100644 --- a/docs/api/helpers.md +++ b/docs/api/helpers.md @@ -372,6 +372,44 @@ displayed. --- +## Era Helpers + +Functions for reading the game data `eraDefinition` collection. These do not call +comlink themselves. + +### get_era_units + +Lists the base ids of an Era's units. With no Era named, it reads the one active now. + +```python +from swgoh_comlink import SwgohComlink +from swgoh_comlink.helpers import DataItems, get_era_units + +comlink = SwgohComlink() +game_data = comlink.get_game_data(items=DataItems.ERA_DEFINITION, include_pve_units=False) + +print(get_era_units(game_data)) +# ['ADELPHIRANGER', 'B1MERCDROID'] +print(get_era_units(game_data, "ERA09_MYTHS")) +# ['DARTHJARJAR', 'THERONIN', 'STORMTROOPERCONCEPT', 'LUKESTARKILLER', 'YODADSV', 'JAXXON', 'MARAJADESKYWALKER'] +``` + +Game data usually carries the previous, current and next Era. An Era's own units are +its `eraUnitId` roster plus its journey unit; pass `include_loaned=True` to also list +the older units it lends to players for its episodes. + +!!! note + Consecutive Eras share their boundary: one ends at the millisecond the next starts. + At that moment the new Era is the active one. When no Era is active, the default + call returns an empty list. + +::: swgoh_comlink.helpers._eras.get_era_units + options: + show_root_heading: true + show_root_full_path: false + +--- + ## Guild Helpers ### Sync diff --git a/src/swgoh_comlink/helpers/__init__.py b/src/swgoh_comlink/helpers/__init__.py index f3f4d0c..1039e5d 100644 --- a/src/swgoh_comlink/helpers/__init__.py +++ b/src/swgoh_comlink/helpers/__init__.py @@ -13,6 +13,7 @@ from ._constants import Constants from ._data_items import DataItems from ._decorators import func_debug_logger, func_timer +from ._eras import get_era_units from ._events import ScheduledEvent, get_event_schedule from ._gac import ( async_get_current_gac_event, @@ -180,6 +181,8 @@ # Events "ScheduledEvent", "get_event_schedule", + # Eras + "get_era_units", # Game configuration "get_game_config", "get_game_config_int", diff --git a/src/swgoh_comlink/helpers/_eras.py b/src/swgoh_comlink/helpers/_eras.py new file mode 100644 index 0000000..df68506 --- /dev/null +++ b/src/swgoh_comlink/helpers/_eras.py @@ -0,0 +1,80 @@ +# coding=utf-8 +"""Era helper functions over the game data ``eraDefinition`` collection.""" + +from __future__ import annotations + +from datetime import datetime, timezone +from typing import Any + +from ..exceptions import SwgohComlinkValueError +from ._utils import get_function_name +from ._wire import as_int, as_list, base_id + + +def get_era_units( + eras: dict[str, Any] | list[dict[str, Any]], + era_id: str | None = None, + *, + now: datetime | None = None, + include_loaned: bool = False, +) -> list[str]: + """List the base ids of the units that belong to an Era. + + Game data carries a few ``eraDefinition`` entries at once, typically the previous, current and next Era, + each open from ``startTimeMs`` until ``endTimeMs`` (epoch milliseconds, sent as strings). Consecutive Eras + share their boundary, so the window is read as half-open: an Era ending at the moment the next begins is no + longer active. + + An Era's own units are its ``eraUnitId`` list plus its ``journeyUnit``, which is normally in that list + already. Eras also lend older units for their episodes (``loanedUnit``); those are added after the Era's + own units when ``include_loaned`` is set. + + Args: + eras: The response from ``SwgohComlink.get_game_data()`` (request ``DataItems.ERA_DEFINITION``), or + its ``eraDefinition`` list. + era_id: The ``id`` of the Era to read, e.g. ``"ERA09_MYTHS"``. Defaults to the Era active at ``now``. + now: The time to compare against when ``era_id`` is omitted. A naive value is read as local time. + Defaults to the current time. + include_loaned: Also list the units the Era lends to players. [Default: False] + + Returns: + Unit base ids in game data order, each listed once. An empty list when ``era_id`` is omitted and no + Era is active at ``now``. + + Raises: + SwgohComlinkValueError: If ``eras`` is not a ``get_game_data()`` response or ``eraDefinition`` list, + or ``era_id`` names no Era in it. + + Examples: + >>> game_data = comlink.get_game_data(items=DataItems.ERA_DEFINITION) # doctest: +SKIP + >>> get_era_units(game_data, "ERA09_MYTHS") # doctest: +SKIP + ['DARTHJARJAR', 'THERONIN', 'STORMTROOPERCONCEPT', 'LUKESTARKILLER', 'YODADSV', 'JAXXON', ...] + """ + definitions = eras.get("eraDefinition") if isinstance(eras, dict) else eras + if not isinstance(definitions, list): + raise SwgohComlinkValueError( + f"{get_function_name()}: 'eras' must be a get_game_data() response or its 'eraDefinition' list, " + f"not {type(eras)}" + ) + definitions = [era for era in definitions if isinstance(era, dict)] + + if era_id is None: + now = datetime.now(timezone.utc) if now is None else now.astimezone(timezone.utc) + now_ms = now.timestamp() * 1000 + era = next( + (e for e in definitions if as_int(e.get("startTimeMs")) <= now_ms < as_int(e.get("endTimeMs"))), None + ) + if era is None: + return [] + else: + era = next((e for e in definitions if e.get("id") == era_id), None) + if era is None: + raise SwgohComlinkValueError( + f"{get_function_name()}: no Era with id {era_id!r} (available: {[e.get('id') for e in definitions]})" + ) + + unit_ids = [*as_list(era.get("eraUnitId")), era.get("journeyUnit")] + if include_loaned: + unit_ids += [loan.get("id") for loan in as_list(era.get("loanedUnit")) if isinstance(loan, dict)] + # dict.fromkeys drops repeats (the journey unit, a loan of an Era unit) and keeps first-seen order + return list(dict.fromkeys(unit for unit in map(base_id, unit_ids) if unit)) diff --git a/tests/unit/test_helpers.py b/tests/unit/test_helpers.py index 252472c..1e7c0bc 100644 --- a/tests/unit/test_helpers.py +++ b/tests/unit/test_helpers.py @@ -36,6 +36,7 @@ get_ability_upgrade_costs, get_conquest_feats, get_data_disc_names, + get_era_units, get_event_schedule, get_game_config, get_game_config_int, @@ -950,6 +951,88 @@ def test_invalid_localization_raises(self): get_event_schedule(_GAME_EVENTS, localization) +# ── _eras ──────────────────────────────────────────────────────────────── + +# Trimmed from live game data: consecutive Eras share their boundary millisecond. +_ERA09_START_MS = 1785261600000 # 2026-07-28 18:00 UTC +_ERA10_START_MS = 1792519200000 # 2026-10-20 18:00 UTC +_ERA_DATA: dict[str, Any] = { + "eraDefinition": [ + { + "id": "ERA09_MYTHS", + "startTimeMs": str(_ERA09_START_MS), + "endTimeMs": str(_ERA10_START_MS), + "eraUnitId": ["DARTHJARJAR", "THERONIN", "JAXXON"], + "journeyUnit": "DARTHJARJAR", + "loanedUnit": [{"expireTime": str(_ERA10_START_MS), "id": "MAUL"}], + }, + { + "id": "ERA10_MANDALORIANANDGROGU", + "startTimeMs": str(_ERA10_START_MS), + "endTimeMs": "1794942000000", + "eraUnitId": ["ADELPHIRANGER", "B1MERCDROID"], + "journeyUnit": "", + "loanedUnit": [ + {"expireTime": "1794942000000", "id": "CARSONTEVA"}, + {"expireTime": "1794942000000", "id": "B1MERCDROID"}, + ], + }, + ] +} + + +def _era_moment(millis: int) -> datetime: + return datetime.fromtimestamp(millis / 1000, tz=timezone.utc) + + +class TestGetEraUnits: + def test_defaults_to_the_active_era(self): + + now = _era_moment(_ERA10_START_MS) + timedelta(days=2) + assert get_era_units(_ERA_DATA, now=now) == ["ADELPHIRANGER", "B1MERCDROID"] + + def test_era_boundary_belongs_to_the_next_era(self): + + assert get_era_units(_ERA_DATA, now=_era_moment(_ERA10_START_MS - 1)) == ["DARTHJARJAR", "THERONIN", "JAXXON"] + assert get_era_units(_ERA_DATA, now=_era_moment(_ERA10_START_MS)) == ["ADELPHIRANGER", "B1MERCDROID"] + + def test_named_era_ignores_now(self): + + units = get_era_units(_ERA_DATA["eraDefinition"], "ERA09_MYTHS", now=_era_moment(0)) + assert units == ["DARTHJARJAR", "THERONIN", "JAXXON"] + + def test_include_loaned_appends_loans_once(self): + + units = get_era_units(_ERA_DATA, "ERA10_MANDALORIANANDGROGU", include_loaned=True) + assert units == ["ADELPHIRANGER", "B1MERCDROID", "CARSONTEVA"] + + def test_journey_unit_missing_from_roster_is_added(self): + + eras = [{"id": "E", "eraUnitId": ["A"], "journeyUnit": "J:SEVEN_STAR", "loanedUnit": ["bad"]}] + assert get_era_units(eras, "E", include_loaned=True) == ["A", "J"] + + def test_no_active_era_returns_empty_list(self): + + assert get_era_units(_ERA_DATA, now=_era_moment(_ERA09_START_MS - 1)) == [] + assert get_era_units({"eraDefinition": []}) == [] + + def test_naive_now_is_local_time(self): + + now = _era_moment(_ERA09_START_MS).astimezone().replace(tzinfo=None) + assert get_era_units(_ERA_DATA, now=now) == ["DARTHJARJAR", "THERONIN", "JAXXON"] + + def test_unknown_era_id_raises(self): + + with pytest.raises(SwgohComlinkValueError, match="ERA99.*ERA09_MYTHS"): + get_era_units(_ERA_DATA, "ERA99") + + @pytest.mark.parametrize("eras", [None, "eras", {"units": []}]) + def test_invalid_eras_raises(self, eras: Any): + + with pytest.raises(SwgohComlinkValueError, match="get_era_units"): + get_era_units(eras) + + # ── _guild (pure functions) ───────────────────────────────────────────── diff --git a/tests/unit/test_helpers_mocked.py b/tests/unit/test_helpers_mocked.py index b5f2f84..c23a4fb 100644 --- a/tests/unit/test_helpers_mocked.py +++ b/tests/unit/test_helpers_mocked.py @@ -25,7 +25,14 @@ import swgoh_comlink.helpers as helpers from swgoh_comlink import SwgohComlink, SwgohComlinkAsync from swgoh_comlink.exceptions import SwgohComlinkValueError -from swgoh_comlink.helpers import get_event_schedule, get_game_config, get_game_config_int, get_guild_activity +from swgoh_comlink.helpers import ( + DataItems, + get_era_units, + get_event_schedule, + get_game_config, + get_game_config_int, + get_guild_activity, +) URL = "http://localhost:3000" LEADERBOARD_URL = f"{URL}/getLeaderboard" @@ -338,6 +345,10 @@ def _journey_summary(events: dict[str, Any]) -> list[tuple[Any, ...]]: return [(e["name"], e["title"], e["subtitle"], e["status"], e["end"]) for e in schedule] +def _era_summary(game_data: dict[str, Any]) -> list[str]: + return get_era_units(game_data, now=datetime(2026, 10, 22, tzinfo=timezone.utc), include_loaned=True) + + def _config_summary(metadata: dict[str, Any]) -> tuple[Any, ...]: return get_game_config(metadata), get_game_config_int(metadata, "max-datacron-currency") @@ -377,6 +388,25 @@ def _config_summary(metadata: dict[str, Any]) -> tuple[Any, ...]: [("THE MANDALORIAN - Hero's Journey", "THE MANDALORIAN", "Hero's Journey", "live", None)], id="event-schedule", ), + pytest.param( + "get_game_data", + {"version": "g1", "items": DataItems.ERA_DEFINITION}, + { + "eraDefinition": [ + { + "id": "ERA10_MANDALORIANANDGROGU", + "startTimeMs": "1792519200000", + "endTimeMs": "1794942000000", + "eraUnitId": ["ADELPHIRANGER", "B1MERCDROID"], + "journeyUnit": "", + "loanedUnit": [{"expireTime": "1794942000000", "id": "CARSONTEVA"}], + } + ] + }, + _era_summary, + ["ADELPHIRANGER", "B1MERCDROID", "CARSONTEVA"], + id="era-units", + ), pytest.param( "get_game_metadata", {},