feat: implement user stats sensors (#108)

README.md

@@ -175,6 +175,51 @@ 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)
 
+##### Stats sensors
+
+Creates individual sensors giving all of your stats about the movies, shows, and episodes you have watched, collected, and rated.
+Add `sensors` > `stats` with a list of the sensors you want to enable. You can enable all of them instead by adding `all` to the list.
+
+The available stats are available:
+ - `movies_plays`
+ - `movies_watched`
+ - `movies_minutes`
+ - `movies_collected`
+ - `movies_ratings`
+ - `movies_comments`
+ - `shows_watched`
+ - `shows_collected`
+ - `shows_ratings`
+ - `shows_comments`
+ - `seasons_ratings`
+ - `seasons_comments`
+ - `episodes_plays`
+ - `episodes_watched`
+ - `episodes_minutes`
+ - `episodes_collected`
+ - `episodes_ratings`
+ - `episodes_comments`
+ - `network_friends`
+ - `network_followers`
+ - `network_following`
+ - `ratings_total`
+
+###### Stats Example
+```yaml
+trakt_tv:
+  sensors:
+    # Create sensors for all available stats
+    stats:
+      - all
+    
+    # OR
+    
+    # Create sensors for specific stats (see available stats above)
+    stats:
+      - episodes_plays
+      - movies_minutes
+```
+
 #### Example
 
 For example, adding only the following to `configuration.yaml` will create two sensors.

custom_components/trakt_tv/apis/trakt.py

@@ -390,6 +390,21 @@ async def fetch_recommendations(self, configured_kinds: list[TraktKind]):
 
         return res
 
+    async def fetch_stats(self):
+        # Load data
+        data = await self.request("get", f"users/me/stats")
+
+        # Flatten data dictionary
+        stats = {}
+        for key, value in data.items():
+            if isinstance(value, dict):
+                for sub_key, sub_value in value.items():
+                    stats[f"{key}_{sub_key}"] = sub_value
+            else:
+                stats[key] = value
+
+        return stats
+
     async def retrieve_data(self):
         async with timeout(1800):
             configuration = Configuration(data=self.hass.data)
@@ -420,6 +435,7 @@ async def retrieve_data(self):
                     configured_kind=TraktKind.NEXT_TO_WATCH_UPCOMING,
                     only_upcoming=True,
                 ),
+                "stats": lambda: self.fetch_stats(),
             }
 
             """First, let's configure which sensors we need depending on configuration"""
@@ -443,6 +459,11 @@ async def retrieve_data(self):
                     sources.append(sub_source)
                     coroutine_sources_data.append(source_function.get(sub_source)())
 
+            """ Load user stats """
+            if configuration.source_exists("stats"):
+                sources.append("stats")
+                coroutine_sources_data.append(source_function.get("stats")())
+
             sources_data = await gather(*coroutine_sources_data)
 
             return {

custom_components/trakt_tv/configuration.py

@@ -78,6 +78,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 stats_key_exists(self, key: str) -> bool:
+        return key in self.conf["sensors"]["stats"]
+
     def source_exists(self, source: str) -> bool:
         try:
             self.conf["sensors"][source]

custom_components/trakt_tv/schema.py

@@ -1,5 +1,5 @@
 from datetime import datetime
-from typing import Any, Dict
+from typing import Any, Dict, List
 
 import pytz
 from dateutil.tz import tzlocal
@@ -41,6 +41,7 @@ def sensors_schema() -> Dict[str, Any]:
         "all_upcoming": upcoming_schema(),
         "next_to_watch": next_to_watch_schema(),
         "recommendation": recommendation_schema(),
+        "stats": Schema(stats_schema()),
     }
 
 
@@ -76,4 +77,32 @@ def recommendation_schema() -> Dict[str, Any]:
     return subschemas
 
 
+def stats_schema() -> list[str]:
+    return [
+        "all",
+        "movies_plays",
+        "movies_watched",
+        "movies_minutes",
+        "movies_collected",
+        "movies_ratings",
+        "movies_comments",
+        "shows_watched",
+        "shows_collected",
+        "shows_ratings",
+        "shows_comments",
+        "seasons_ratings",
+        "seasons_comments",
+        "episodes_plays",
+        "episodes_watched",
+        "episodes_minutes",
+        "episodes_collected",
+        "episodes_ratings",
+        "episodes_comments",
+        "network_friends",
+        "network_followers",
+        "network_following",
+        "ratings_total",
+    ]
+
+
 configuration_schema = dictionary_to_schema(domain_schema(), extra=ALLOW_EXTRA)

custom_components/trakt_tv/sensor.py

@@ -69,6 +69,40 @@ async def async_setup_entry(hass, config_entry, async_add_entities):
             )
             sensors.append(sensor)
 
+    # Add sensors for stats
+    if configuration.source_exists("stats"):
+        stats = {}
+        # Check if the coordinator has data
+        if coordinator.data:
+            stats = coordinator.data.get("stats", {})
+
+        # Check if all stats are allowed
+        allow_all = configuration.stats_key_exists("all")
+
+        # Create a sensor for each key in the stats
+        for key, value in stats.items():
+            # Skip the key if it is not a valid state (e.g. rating distribution dict)
+            if isinstance(value, dict):
+                continue
+
+            # Skip if not allowed in config
+            if not allow_all and not configuration.stats_key_exists(key):
+                continue
+
+            # Transform the key to a more readable format
+            title = key.replace("_", " ").title()
+
+            # Create the sensor
+            sensor = TraktStateSensor(
+                hass=hass,
+                coordinator=coordinator,
+                prefix="Trakt Stats",
+                mdi_icon="mdi:chart-line",
+                title=title,
+                state=value,
+            )
+            sensors.append(sensor)
+
     async_add_entities(sensors)
 
 
@@ -151,3 +185,55 @@ def has_entity_name(self) -> bool:
     async def async_update(self):
         """Request coordinator to update data."""
         await self.coordinator.async_request_refresh()
+
+
+class TraktStateSensor(Entity):
+    """Trakt sensor to show data as state"""
+
+    _attr_has_entity_name = True
+
+    def __init__(
+        self,
+        hass,
+        coordinator,
+        prefix: str,
+        mdi_icon: str,
+        title: str,
+        state: str,
+    ):
+        """Initialize the sensor."""
+        self.hass = hass
+        self.coordinator = coordinator
+        self.prefix = prefix
+        self.icon = mdi_icon
+        self.title = title
+        self.state = state
+
+    @property
+    def name(self):
+        """Return the name of the sensor."""
+        return f"{self.prefix} {self.title}"
+
+    @property
+    def unit_of_measurement(self):
+        """Return the unit of measurement of this entity"""
+        units = [
+            "ratings",
+            "minutes",
+            "comments",
+            "friends",
+            "followers",
+            "following",
+            "episodes",
+            "movies",
+            "shows",
+            "seasons",
+        ]
+        for unit in units:
+            if unit in self.title.lower():
+                return unit
+        return None
+
+    async def async_update(self):
+        """Request coordinator to update data."""
+        await self.coordinator.async_request_refresh()