Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
38 changes: 38 additions & 0 deletions docs/api/helpers.md
Original file line number Diff line number Diff line change
Expand Up @@ -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
Expand Down
3 changes: 3 additions & 0 deletions src/swgoh_comlink/helpers/__init__.py
Original file line number Diff line number Diff line change
Expand Up @@ -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,
Expand Down Expand Up @@ -180,6 +181,8 @@
# Events
"ScheduledEvent",
"get_event_schedule",
# Eras
"get_era_units",
# Game configuration
"get_game_config",
"get_game_config_int",
Expand Down
80 changes: 80 additions & 0 deletions src/swgoh_comlink/helpers/_eras.py
Original file line number Diff line number Diff line change
@@ -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))
83 changes: 83 additions & 0 deletions tests/unit/test_helpers.py
Original file line number Diff line number Diff line change
Expand Up @@ -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,
Expand Down Expand Up @@ -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) ─────────────────────────────────────────────


Expand Down
32 changes: 31 additions & 1 deletion tests/unit/test_helpers_mocked.py
Original file line number Diff line number Diff line change
Expand Up @@ -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"
Expand Down Expand Up @@ -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")

Expand Down Expand Up @@ -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",
{},
Expand Down
Loading