fix: renamed base_objects to core

examples/worldstate/custom_models.py

@@ -5,7 +5,7 @@
 from msgspec import field  # use this to rename response keys
 
 from warframe.worldstate import WorldstateClient
-from warframe.worldstate.common.base_objects import (
+from warframe.worldstate.common.core import (
     SingleQueryModel,
 )  # this import might change
 
@@ -29,9 +29,9 @@ class CustomCambionDrift(SingleQueryModel):
 
 async def main():
     async with WorldstateClient() as client:
-        arbi = await client.query(CustomCambionDrift)
+        ccd = await client.query(CustomCambionDrift)
 
-        print(arbi)
+        print(ccd)
 
 
 if __name__ == "__main__":

warframe/worldstate/client.py

@@ -108,7 +108,7 @@ async def query(
 
     async def query_list_of(
         self, cls: Type[SupportsMultiQuery], language: Optional[Language] = None
-    ) -> Optional[List[SupportsMultiQuery]]:
+    ) -> List[SupportsMultiQuery]:
         """Queries the model of type `MultiQueryModel` to return its corresponding object.
 
         Args:

warframe/worldstate/common/__init__.py

@@ -1,2 +1,2 @@
 from .types_ import *
-from .base_objects import *
+from .core import *

warframe/worldstate/common/core.py

@@ -0,0 +1,119 @@
+from datetime import datetime, timezone
+from typing import Any, List, Type, TypeVar, ClassVar
+
+import msgspec
+
+
+__all__ = ["MultiQueryModel", "SingleQueryModel", "WorldstateObject", "TimedEvent"]
+
+
+def _decode_hook(type: Type, obj: Any) -> Any:
+    if isinstance(type, datetime) and isinstance(obj, str):
+        return datetime.fromisoformat(obj.strip("Z"))
+
+    return obj
+
+
+def _get_short_format_time_string(dt: datetime) -> str:
+    """Returns a short time formatted string based on the now and the dt.
+
+    Args:
+        dt (datetime): The time the Event XYZ started/ends
+
+    Returns:
+        str: The short time formatted string of the time in between now and when the dt.
+    """
+    now = datetime.now(tz=timezone.utc)
+    time_in_between = now - dt if now > dt else dt - now
+
+    days = time_in_between.days
+    hours, remainder = divmod(time_in_between.seconds, 3600)
+    minutes, seconds = divmod(remainder, 60)
+
+    time_components = [(days, "d"), (hours, "h"), (minutes, "m"), (seconds, "s")]
+    formatted_time = ""
+
+    for t, suffix in time_components:
+        if t != 0:
+            formatted_time += f"{t}{suffix} "
+
+    return formatted_time.strip()
+
+
+class TimedEvent(msgspec.Struct, rename="camel"):
+    activation: datetime
+    "The time the event began"
+
+    expiry: datetime
+    "The time the event ends"
+
+    @property
+    def start_string(self) -> str:
+        "Short-time-formatted duration string representing the start of the event"
+        return f"{_get_short_format_time_string(self.activation)} ago"
+
+    @property
+    def eta(self) -> str:
+        "Short-time-formatted duration string representing the end of the event / cycle"
+        return _get_short_format_time_string(self.expiry)
+
+    @property
+    def expired(self) -> bool:
+        return self.activation >= self.expiry
+
+    @property
+    def active(self) -> bool:
+        return not self.expired
+
+
+class WorldstateObject(msgspec.Struct, rename="camel"):
+    """
+    Base class for every model-related object.
+    """
+
+    pass
+
+
+T = TypeVar("T", bound=WorldstateObject)
+
+
+class MultiQueryModel(WorldstateObject):
+    """
+    Base class for giving models an indicator whether they can only come from a JSON array.
+    """
+
+    __endpoint__: ClassVar[str]
+
+    @classmethod
+    def _from_json(cls: Type[T], response: str) -> List[T]:
+        """Decodes a JSON string to an list of object of T.
+
+        Args:
+            cls (Type[T]): The type T.
+            response (str): The raw JSON as string.
+
+        Returns:
+            List[T]: A list of objects of T.
+        """
+        return msgspec.json.decode(response, type=List[cls], dec_hook=_decode_hook)
+
+
+class SingleQueryModel(WorldstateObject):
+    """
+    Base class for giving models an indicator whether they can only come from a single JSON object.
+    """
+
+    __endpoint__: ClassVar[str]
+
+    @classmethod
+    def _from_json(cls: Type[T], response: str) -> T:
+        """Decodes a JSON string to an object of T.
+
+        Args:
+            cls (Type[T]): The type T.
+            response (str): The raw JSON as string.
+
+        Returns:
+            T: The object of T
+        """
+        return msgspec.json.decode(response, type=cls, dec_hook=_decode_hook)

warframe/worldstate/models/alert.py

@@ -1,37 +1,17 @@
-from datetime import datetime
-from typing import List, Optional
+from typing import List
 
-from ..common import ItemRewardType, MultiQueryModel
+from ..common import ItemRewardType, MultiQueryModel, TimedEvent
 from .mission import Mission
 
 __all__ = ["Alert"]
 
 
-class Alert(MultiQueryModel):
+class Alert(MultiQueryModel, TimedEvent):
     __endpoint__ = "/alerts"
 
     # required
-    activation: datetime
-    "The time the mission began"
-
-    expiry: datetime
-    "The time the mission ends"
-
     mission: Mission
     "The mission that corresponds to this Alert"
 
     reward_types: List[ItemRewardType]
     "A list of reward types"
-
-    # optional
-    start_string: Optional[str] = None
-    "Short-time-formatted duration string representing the start of the alert"
-
-    active: Optional[bool] = None
-    "Whether the alert is still active or not"
-
-    eta: Optional[str] = None
-    "Short-formatted string estimating the time until the Alert is closed"
-
-    expired: Optional[bool] = None
-    "Whether the mission is expired or not"

warframe/worldstate/models/arbitration.py

@@ -1,37 +1,40 @@
-from datetime import datetime
 from typing import Optional
 
 from msgspec import field
 
-from ..common import SingleQueryModel
+from ..common import SingleQueryModel, TimedEvent
 from ..enums import Faction, MissionType
 
 __all__ = ["Arbitration"]
 
 
-class Arbitration(SingleQueryModel):
+class Arbitration(SingleQueryModel, TimedEvent):
+    """
+    UNSTABLE
+    """
+
     __endpoint__ = "/arbitration"
 
     # required
-    activation: datetime
-    "The time the mission began."
-    expiry: datetime
-    "The time the mission ends."
     node: str
+    "The localized plain name for the node the Arbitration is on."
+
+    node_key: str
     "The plain name for the node the Arbitration is on."
+
     faction: Faction = field(
         name="enemy"
     )  # it's not called faction at the endpoint, instead it's called enemy
     "The Faction of the corresponding mission."
+
     archwing_required: bool = field(name="archwing")
     "Whether an archwing is required in order to play the mission"
+
     is_sharkwing: bool = field(name="sharkwing")
     "Whether the mission takes place in a submerssible mission"
 
-    # optional
-    start_string: Optional[str] = None
-    "Short-time-formatted duration string representing the start of the Arbitration."
-    active: Optional[bool] = None
-    "Whether the alert is still active or not."
-    mission_type: Optional[MissionType] = None
+    mission_type: str = field(name="type")
+    "The localized MissionType of the given mission (Capture, Spy, etc.)"
+
+    mission_type_key: MissionType = field(name="typeKey")
     "The MissionType of the given mission (Capture, Spy, etc.)"