[Feature] Trakt List Sensor

.devcontainer/configuration.yaml

@@ -23,3 +23,7 @@ trakt_tv:
         max_medias: 10
       show:
         max_medias: 2
+    list:
+      - id: 3837211
+        sort_by: added
+        sort_order: desc

README.md

@@ -175,6 +175,19 @@ 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)
 
+##### List sensors
+
+List sensors are sensors gather media from a specified list.
+
+You can have as many list sensors as you would like
+
+There are one parameter for each sensor:
+
+- `id` should be a positive number for the id of the trakt list
+- `sort_by` should be a left blank or set to 'rank', 'added' or 'released'
+- `sort_order` should be left blank or set to either 'asc'or 'desc'
+- `max_medias` should be a positive number for how many items to grab - default 30
+
 #### Example
 
 For example, adding only the following to `configuration.yaml` will create two sensors.
@@ -196,6 +209,11 @@ trakt_tv:
         max_medias: 3
       movie:
         max_medias: 3
+    list:
+      - id: 3837211
+        sort_by: added
+        sort_order: desc
+        max_medias: 20
 ```
 
 ### 3. Restart Home Assistant

custom_components/trakt_tv/apis/trakt.py

@@ -293,12 +293,36 @@ async def fetch_recommendations(self):
             res[trakt_kind] = Medias(medias)
         return res
 
+    async def fetch_list(self, id: int, type: str = ""):
+        return await self.request(
+            "get", f"lists/{id}/items/{type}"
+        )
+
+    async def fetch_lists(self):
+        configuration = Configuration(data=self.hass.data)
+        language = configuration.get_language()
+        res = []
+        for list in configuration.get_list_configs():
+            new_list = []
+            type = list['type'] if list.get('type') and list['type'] != 'all' else ''
+            data = await self.fetch_list(list['id'], type)
+            for media in data:
+                if media['type'] == 'movie':
+                    new_list.append(BASIC_KINDS[1].value.model.from_trakt(media))
+                else:
+                    new_list.append(BASIC_KINDS[0].value.model.from_trakt(media))
+            await gather(*[media.get_more_information(language) for media in new_list])
+            res.append(Medias(new_list))
+
+        return res
+
     async def retrieve_data(self):
         async with timeout(1800):
             titles = [
                 "upcoming",
                 "all_upcoming",
                 "recommendation",
+                "list",
                 "all",
                 "only_aired",
                 "only_upcoming",
@@ -308,6 +332,7 @@ async def retrieve_data(self):
                     self.fetch_upcomings(all_medias=False),
                     self.fetch_upcomings(all_medias=True),
                     self.fetch_recommendations(),
+                    self.fetch_lists(),
                     self.fetch_next_to_watch(),
                     self.fetch_next_to_watch(only_aired=True),
                     self.fetch_next_to_watch(only_upcoming=True),

custom_components/trakt_tv/configuration.py

@@ -76,3 +76,9 @@ 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 get_list_configs(self) -> int:
+        try:
+            return self.conf["sensors"]["list"]
+        except KeyError:
+            return []

custom_components/trakt_tv/models/media.py

@@ -45,6 +45,8 @@ def from_trakt(data) -> "Identifiers":
 class Media(ABC):
     name: str
     ids: Identifiers
+    rank: int
+    added: datetime
 
     @abstractstaticmethod
     def from_trakt(data) -> "Media":
@@ -117,6 +119,8 @@ def from_trakt(data) -> "Movie":
             name=movie["title"],
             released=released,
             ids=Identifiers.from_trakt(movie),
+            rank=data["rank"] if data.get("rank") else None,
+            added=data["listed_at"] if data.get("listed_at") else None,
         )
 
     async def get_more_information(self, language):
@@ -172,6 +176,8 @@ class Episode:
     season: int
     title: str
     ids: Identifiers
+    rank: int
+    added: datetime
 
     @staticmethod
     def from_trakt(data) -> "Episode":
@@ -185,6 +191,8 @@ def from_trakt(data) -> "Episode":
             season=episode["season"],
             title=episode["title"],
             ids=Identifiers.from_trakt(episode),
+            rank=data["rank"] if data.get("rank") else None,
+            added=data["listed_at"] if data.get("listed_at") else None,
         )
 
 
@@ -220,6 +228,8 @@ def from_trakt(data) -> "Show":
             ids=Identifiers.from_trakt(show),
             released=released,
             episode=episode,
+            rank=data["rank"] if data.get("rank") else None,
+            added=data["listed_at"] if data.get("listed_at") else None,
         )
 
     def update_common_information(self, data: Dict[str, Any]):
@@ -283,13 +293,24 @@ def to_homeassistant(self) -> Dict[str, Any]:
 class Medias:
     items: List[Media]
 
-    def to_homeassistant(self) -> Dict[str, Any]:
+    def to_homeassistant(self, config={}) -> Dict[str, Any]:
         """
-        Convert the List of medias to recommendation data.
+        Sort the list then convert the medias to list data.
 
         :return: The dictionary containing all necessary information for upcoming media
                  card
         """
-        medias = sorted(self.items, key=lambda media: media.released)
+        medias = self.items
+        if "sort_by" in config:
+            if config["sort_by"] == "released":
+                medias = sorted(medias, key=lambda media: media.released)
+            if config["sort_by"] == "rank":
+                medias = sorted(medias, key=lambda media: media.rank)
+            if config["sort_by"] == "added":
+                medias = sorted(medias, key=lambda media: media.added)
+        if "sort_order" in config and config["sort_order"] == "desc":
+            medias.reverse()
+        max_medias = config["max_medias"] if "max_medias" in config else 30
+        medias = medias[0 : max_medias]
         medias = [media.to_homeassistant() for media in medias]
         return [first_item] + medias

custom_components/trakt_tv/schema.py

@@ -1,10 +1,10 @@
 from datetime import datetime
-from typing import Any, Dict
+from typing import Any, Dict, List
 
 import pytz
 from dateutil.tz import tzlocal
 from homeassistant.helpers import config_validation as cv
-from voluptuous import ALLOW_EXTRA, PREVENT_EXTRA, In, Required, Schema
+from voluptuous import ALLOW_EXTRA, PREVENT_EXTRA, In, Required, Optional, Schema
 
 from .const import DOMAIN, LANGUAGE_CODES
 from .models.kind import BASIC_KINDS, NEXT_TO_WATCH_KINDS, TraktKind
@@ -41,6 +41,7 @@ def sensors_schema() -> Dict[str, Any]:
         "all_upcoming": upcoming_schema(),
         "next_to_watch": next_to_watch_schema(),
         "recommendation": recommendation_schema(),
+        "list": lists_schema(),
     }
 
 
@@ -72,8 +73,21 @@ def recommendation_schema() -> Dict[str, Any]:
         subschemas[trakt_kind.value.identifier] = {
             Required("max_medias", default=3): cv.positive_int,
         }
-
     return subschemas
 
 
+def lists_schema() -> List:
+    return [list_schema()]
+
+
+def list_schema() -> Dict[str, Any]:
+    return {
+        Required("id"): cv.positive_int,
+        Optional("sort_by"): In(["rank", "added", "released"]),
+        Optional("sort_order"): In(["asc", "desc"]),
+        Optional("type"): In(["shows", "movies", "episodes", "all"]),
+        Optional("max_medias"): cv.positive_int,
+    }
+
+
 configuration_schema = dictionary_to_schema(domain_schema(), extra=ALLOW_EXTRA)

custom_components/trakt_tv/sensor.py

@@ -68,10 +68,23 @@ async def async_setup_entry(hass, config_entry, async_add_entities):
             )
             sensors.append(sensor)
 
+    for index, list in enumerate(configuration.get_list_configs()):
+        sensor = ListSensor(
+            hass=hass,
+            config_entry=config_entry,
+            coordinator=coordinator,
+            config=list,
+            source="list",
+            index=index,
+            prefix="Trakt List",
+            mdi_icon="mdi:format-list-group",
+        )
+        sensors.append(sensor)
+
     async_add_entities(sensors)
 
 
-class TraktSensor(Entity):
+class BaseSensor(Entity):
     """Representation of a trakt sensor."""
 
     _attr_has_entity_name = True
@@ -81,7 +94,6 @@ def __init__(
         hass,
         config_entry,
         coordinator,
-        trakt_kind: TraktKind,
         source: str,
         prefix: str,
         mdi_icon: str,
@@ -90,11 +102,52 @@ def __init__(
         self.hass = hass
         self.config_entry = config_entry
         self.coordinator = coordinator
-        self.trakt_kind = trakt_kind
         self.source = source
         self.prefix = prefix
         self.mdi_icon = mdi_icon
 
+    @property
+    def state(self):
+        """Return the state of the sensor."""
+        return max([len(self.data) - 1, 0])
+
+    @property
+    def icon(self):
+        """Return the icon to use in the frontend, if any."""
+        return self.mdi_icon
+
+    @property
+    def extra_state_attributes(self):
+        """Return the state attributes of the sensor."""
+        return {"data": self.data}
+
+    @property
+    def has_entity_name(self) -> bool:
+        """Return if the name of the entity is describing only the entity itself."""
+        return True
+
+    async def async_update(self):
+        """Request coordinator to update data."""
+        await self.coordinator.async_request_refresh()
+
+
+class TraktSensor(BaseSensor):
+    """Representation of a trakt sensor."""
+
+    def __init__(
+        self,
+        hass,
+        config_entry,
+        coordinator,
+        trakt_kind: TraktKind,
+        source: str,
+        prefix: str,
+        mdi_icon: str,
+    ):
+        """Initialize the sensor."""
+        super().__init__(hass, config_entry, coordinator, source, prefix, mdi_icon)
+        self.trakt_kind = trakt_kind
+
     @property
     def name(self):
         """Return the name of the sensor."""
@@ -119,34 +172,57 @@ def configuration(self):
     def data(self):
         if self.medias:
             max_medias = self.configuration["max_medias"]
-            return self.medias.to_homeassistant()[0 : max_medias + 1]
+            return self.medias.to_homeassistant({"sort_by": 'released'})[0 : max_medias + 1]
         return []
 
     @property
-    def state(self):
-        """Return the state of the sensor."""
-        return max([len(self.data) - 1, 0])
+    def unit_of_measurement(self):
+        """Return the unit of measurement of this entity, if any."""
+        return self.trakt_kind.value.path.split("/")[0]
+
+
+class ListSensor(BaseSensor):
+    """Representation of a trakt sensor."""
+
+    def __init__(
+        self,
+        hass,
+        config_entry,
+        coordinator,
+        config: dict,
+        source: str,
+        index: int,
+        prefix: str,
+        mdi_icon: str,
+    ):
+        """Initialize the sensor."""
+        super().__init__(hass, config_entry, coordinator, source, prefix, mdi_icon)
+        self.config = config
+        self.index = index
 
     @property
-    def icon(self):
-        """Return the icon to use in the frontend, if any."""
-        return self.mdi_icon
+    def name(self):
+        """Return the name of the sensor."""
+        return f"{self.prefix} {self.config['id']}"
 
     @property
-    def unit_of_measurement(self):
-        """Return the unit of measurement of this entity, if any."""
-        return self.trakt_kind.value.path.split("/")[0]
+    def medias(self):
+        if self.coordinator.data:
+            return self.coordinator.data.get(self.source, [])[self.index]
+        return None
 
     @property
-    def extra_state_attributes(self):
-        """Return the state attributes of the sensor."""
-        return {"data": self.data}
+    def configuration(self):
+        data = self.hass.data[DOMAIN]
+        return data["configuration"]["sensors"]["list"][self.config["id"]]
 
     @property
-    def has_entity_name(self) -> bool:
-        """Return if the name of the entity is describing only the entity itself."""
-        return True
+    def data(self):
+        if self.medias:
+            return self.medias.to_homeassistant(self.config)
+        return []
 
-    async def async_update(self):
-        """Request coordinator to update data."""
-        await self.coordinator.async_request_refresh()
+    @property
+    def unit_of_measurement(self):
+        """Return the unit of measurement of this entity, if any."""
+        return "media"