feat: add anticipated sensors (#111)

README.md

@@ -175,6 +175,43 @@ There are three parameters for each sensor:
 - `exclude` should be a list of shows you'd like to exclude, since it's based on your watched history. To find keys to put there, go on trakt.tv, search for a show, click on it, notice the url slug, copy/paste it. So, if I want to hide "Friends", I'll do the steps mentioned above, then land on https://trakt.tv/shows/friends, I'll just have to copy/paste the last part, `friends`, that's it
   You can also use the Trakt.tv "hidden" function to hide a show from [your calendar](https://trakt.tv/calendars/my/shows) or the [progress page](https://trakt.tv/users/<username>/progress)
 
+##### Anticipated Sensors
+
+- `sensor.trakt_anticipated_shows`: This sensor displays the most anticipated TV shows on Trakt.
+- `sensor.trakt_anticipated_movies`: This sensor displays the most anticipated movies on Trakt.
+
+### Example Configuration
+
+To enable these sensors, you can use the following configuration:
+
+```yaml
+trakt_tv:
+  language: en
+  timezone: America/Los_Angeles
+  sensors:
+    anticipated:
+      movie:
+        max_medias: 10
+      show:
+        max_medias: 10
+```
+
+You can also exclude collected items from these sensors:
+
+```yaml
+trakt_tv:
+  language: en
+  timezone: America/Los_Angeles
+  sensors:
+    anticipated:
+      movie:
+        exclude_collected: true
+        max_medias: 10
+      show:
+        exclude_collected: true
+        max_medias: 10
+```
+
 ##### Stats sensors
 
 Creates individual sensors giving all of your stats about the movies, shows, and episodes you have watched, collected, and rated.

custom_components/trakt_tv/apis/trakt.py

@@ -41,6 +41,7 @@ def __init__(
     def cache(self) -> Dict[str, Any]:
         return self.hass.data[DOMAIN].get("cache", {})
 
+
     async def async_get_access_token(self) -> str:
         """Return a valid access token."""
         if not self.oauth_session.valid_token:
@@ -405,6 +406,50 @@ async def fetch_stats(self):
 
         return stats
 
+    async def fetch_anticipated(self, path: str, limit: int, ignore_collected: bool):
+        return await self.request(
+            "get", f"{path}/anticipated?limit={limit}&ignore_collected={ignore_collected}"
+        )
+
+    async def fetch_anticipated_medias(self, configured_kinds: list[TraktKind]):
+        from ..models.kind import ANTICIPATED_KINDS
+
+        kinds = []
+        for kind in configured_kinds:
+            if kind in ANTICIPATED_KINDS:
+                kinds.append(kind)
+            else:
+                LOGGER.warn(
+                    f"Anticipated doesn't support {kind}, you should remove it from the configuration."
+                )
+
+        configuration = Configuration(data=self.hass.data)
+        language = configuration.get_language()
+        data = await gather(
+            *[
+                self.fetch_anticipated(
+                    kind.value.path,
+                    configuration.get_anticipated_max_medias(kind.value.identifier),
+                    configuration.anticipated_exclude_collected(kind.value.identifier)
+                )
+                for kind in kinds
+            ]
+        )
+
+        res = {}
+
+        for trakt_kind, raw_medias in zip(kinds, data):
+            if raw_medias is not None:
+                medias = [
+                    trakt_kind.value.model.from_trakt(media[trakt_kind.value.identifier]) for media in raw_medias
+                ]
+                await gather(
+                    *[media.get_more_information(language) for media in medias]
+                )
+                res[trakt_kind] = Medias(medias)
+
+        return res
+
     async def retrieve_data(self):
         async with timeout(1800):
             configuration = Configuration(data=self.hass.data)
@@ -424,6 +469,9 @@ async def retrieve_data(self):
                 "recommendation": lambda kinds: self.fetch_recommendations(
                     configured_kinds=kinds,
                 ),
+                "anticipated": lambda kinds: self.fetch_anticipated_medias(
+                    configured_kinds=kinds,
+                ),
                 "all": lambda: self.fetch_next_to_watch(
                     configured_kind=TraktKind.NEXT_TO_WATCH_ALL,
                 ),
@@ -443,6 +491,7 @@ async def retrieve_data(self):
                 "upcoming",
                 "all_upcoming",
                 "recommendation",
+                "anticipated",
             ]:
                 if configuration.source_exists(source):
                     sources.append(source)

custom_components/trakt_tv/configuration.py

@@ -78,6 +78,18 @@ def recommendation_identifier_exists(self, identifier: str) -> bool:
     def get_recommendation_max_medias(self, identifier: str) -> int:
         return self.get_max_medias(identifier, "recommendation")
 
+    def anticipated_identifier_exists(self, identifier: str) -> bool:
+        return self.identifier_exists(identifier, "anticipated")
+
+    def get_anticipated_max_medias(self, identifier: str) -> int:
+        return self.get_max_medias(identifier, "anticipated")
+
+    def anticipated_exclude_collected(self, identifier: str) -> bool:
+        try:
+            return self.conf["sensors"]["anticipated"][identifier]["exclude_collected"]
+        except KeyError:
+            return False
+
     def stats_key_exists(self, key: str) -> bool:
         return key in self.conf["sensors"]["stats"]
 

custom_components/trakt_tv/models/kind.py

@@ -23,6 +23,8 @@ class TraktKind(Enum):
     NEXT_TO_WATCH_UPCOMING = CalendarInformation(
         "only_upcoming", "Only Upcoming", "shows", Show
     )
+    ANTICIPATED_MOVIE = CalendarInformation("movie", "Movies", "movies", Movie)
+    ANTICIPATED_SHOW = CalendarInformation("show", "Shows", "shows", Show)
 
     @classmethod
     def from_string(cls, string):
@@ -50,3 +52,8 @@ def from_string(cls, string):
     TraktKind.NEXT_TO_WATCH_AIRED,
     TraktKind.NEXT_TO_WATCH_UPCOMING,
 ]
+
+ANTICIPATED_KINDS = [
+    TraktKind.ANTICIPATED_MOVIE,
+    TraktKind.ANTICIPATED_SHOW,
+]

custom_components/trakt_tv/schema.py

@@ -7,7 +7,7 @@
 from voluptuous import ALLOW_EXTRA, PREVENT_EXTRA, In, Required, Schema
 
 from .const import DOMAIN, LANGUAGE_CODES
-from .models.kind import BASIC_KINDS, NEXT_TO_WATCH_KINDS, TraktKind
+from .models.kind import BASIC_KINDS, NEXT_TO_WATCH_KINDS, TraktKind, ANTICIPATED_KINDS
 
 
 def dictionary_to_schema(
@@ -41,6 +41,7 @@ def sensors_schema() -> Dict[str, Any]:
         "all_upcoming": upcoming_schema(),
         "next_to_watch": next_to_watch_schema(),
         "recommendation": recommendation_schema(),
+        "anticipated": anticipated_schema(),
         "stats": Schema(stats_schema()),
     }
 
@@ -77,6 +78,16 @@ def recommendation_schema() -> Dict[str, Any]:
     return subschemas
 
 
+def anticipated_schema() -> Dict[str, Any]:
+    subschemas = {}
+    for trakt_kind in ANTICIPATED_KINDS:
+        subschemas[trakt_kind.value.identifier] = {
+            Required("max_medias", default=3): cv.positive_int,
+            Required("exclude_collected", default=False): cv.boolean,
+        }
+
+    return subschemas
+
 def stats_schema() -> list[str]:
     return [
         "all",

custom_components/trakt_tv/sensor.py

@@ -7,7 +7,7 @@
 
 from .configuration import Configuration
 from .const import DOMAIN
-from .models.kind import BASIC_KINDS, NEXT_TO_WATCH_KINDS, TraktKind
+from .models.kind import BASIC_KINDS, NEXT_TO_WATCH_KINDS, TraktKind, ANTICIPATED_KINDS
 
 LOGGER = logging.getLogger(__name__)
 
@@ -51,6 +51,21 @@ async def async_setup_entry(hass, config_entry, async_add_entities):
             )
             sensors.append(sensor)
 
+    for trakt_kind in TraktKind:
+        if trakt_kind in ANTICIPATED_KINDS:
+            identifier = trakt_kind.value.identifier
+            if configuration.anticipated_identifier_exists(identifier):
+                sensor = TraktSensor(
+                    hass=hass,
+                    config_entry=config_entry,
+                    coordinator=coordinator,
+                    trakt_kind=trakt_kind,
+                    source="anticipated",
+                    prefix="Trakt Anticipated",
+                    mdi_icon="mdi:movie" if trakt_kind == TraktKind.ANTICIPATED_MOVIE else "mdi:television",
+                )
+                sensors.append(sensor)
+
     for trakt_kind in TraktKind:
         if trakt_kind not in NEXT_TO_WATCH_KINDS:
             continue