Centralize conversions from DTO to model instances

Refs: #143

argos/controllers/albums.py

@@ -6,7 +6,6 @@
     from argos.app import Application
 
 from argos.controllers.base import ControllerBase
-from argos.controllers.utils import parse_tracks
 from argos.controllers.visitors import AlbumMetadataCollector, LengthAcc
 from argos.info import InformationService
 from argos.message import Message, MessageType, consume
@@ -51,7 +50,7 @@ async def complete_album_description(self, message: Message) -> None:
 
         length_acc = LengthAcc()
         metadata_collector = AlbumMetadataCollector()
-        parsed_tracks = parse_tracks(
+        parsed_tracks = self.helper.parse_tracks(
             tracks_dto, visitors=[length_acc, metadata_collector]
         ).get(album_uri, [])
 

argos/controllers/base.py

@@ -6,6 +6,7 @@
 if TYPE_CHECKING:
     from argos.app import Application
 
+from argos.controllers.helper import ModelHelper
 from argos.http import MopidyHTTPClient
 from argos.message import Message, MessageType
 from argos.model import Model
@@ -35,6 +36,8 @@ def __init__(
         self._notifier: Notifier = application.props.notifier
         self._settings: Gio.Settings = application.props.settings
 
+        self.helper = ModelHelper()
+
     def send_message(
         self, message_type: MessageType, data: Optional[Dict[str, Any]] = None
     ) -> None:

argos/controllers/helper.py

@@ -0,0 +1,65 @@
+from collections import defaultdict
+from typing import Callable, Dict, List, Mapping, Optional, Sequence
+
+from argos.dto import ArtistDTO, TlTrackDTO, TrackDTO
+from argos.model import TracklistTrackModel, TrackModel
+
+
+class ModelHelper:
+    """Help convert DTOs to model instances.
+
+    Some conversions can be tracked through visitors.
+
+    """
+
+    def convert_track(self, track_dto: TrackDTO) -> TrackModel:
+        track = TrackModel(
+            uri=track_dto.uri,
+            name=track_dto.name,
+            track_no=track_dto.track_no if track_dto.track_no is not None else -1,
+            disc_no=track_dto.disc_no if track_dto.disc_no is not None else 1,
+            length=track_dto.length if track_dto.length is not None else -1,
+            artist_name=track_dto.artists[0].name if len(track_dto.artists) > 0 else "",
+            album_name=track_dto.album.name if track_dto.album is not None else "",
+            last_modified=track_dto.last_modified
+            if track_dto.last_modified is not None
+            else -1,
+        )
+        return track
+
+    def convert_tl_track(self, tl_track_dto: TlTrackDTO) -> TracklistTrackModel:
+        track = self.convert_track(tl_track_dto.track)
+        tl_track = TracklistTrackModel(tlid=tl_track_dto.tlid, track=track)
+        return tl_track
+
+    def parse_tracks(
+        self,
+        tracks_dto: Mapping[str, Sequence[TrackDTO]],
+        *,
+        visitors: Optional[Sequence[Callable[[str, TrackDTO], None]]] = None,
+    ) -> Dict[str, List[TrackModel]]:
+        """Parse a track list.
+
+        Keys in ``track_dto`` can be album URIs or track URIs (when fetching
+        details of playlist tracks).
+
+        Args:
+            tracks_dto: Track data transfer objects to parse.
+
+            visitors: An optional list of callable to be called on each
+                visited track.
+
+        Returns:
+            Dict of list of ``TrackModel``.
+
+        """
+        parsed_tracks: Dict[str, List[TrackModel]] = defaultdict(list)
+        for uri in tracks_dto:
+            for track_dto in tracks_dto[uri]:
+                if visitors is not None:
+                    for visitor in visitors:
+                        visitor(uri, track_dto)
+
+                parsed_tracks[uri].append(self.convert_track(track_dto))
+
+        return parsed_tracks

argos/controllers/library.py

@@ -14,7 +14,7 @@
     DirectoryCompletionProgressNotifier,
     ProgressNotifierProtocol,
 )
-from argos.controllers.utils import call_by_slice, parse_tracks
+from argos.controllers.utils import call_by_slice
 from argos.controllers.visitors import AlbumMetadataCollector, LengthAcc
 from argos.download import ImageDownloader
 from argos.dto import RefDTO, RefType
@@ -315,7 +315,7 @@ async def _complete_albums(
             )
 
             LOGGER.debug("Parsing albums tracks")
-            parsed_tracks = parse_tracks(
+            parsed_tracks = self.helper.parse_tracks(
                 directory_tracks_dto, visitors=[length_acc, metadata_collector]
             )
 
@@ -438,7 +438,7 @@ async def _complete_tracks(
 
         LOGGER.debug("Parsing tracks")
         parsed_tracks: List[TrackModel] = []
-        for tracks in parse_tracks(directory_tracks_dto).values():
+        for tracks in self.helper.parse_tracks(directory_tracks_dto).values():
             for track in tracks:
                 track_uri = track.uri
                 if images is not None and len(images.get(track_uri, [])) > 0:

argos/controllers/playlists.py

@@ -11,7 +11,7 @@
     from argos.app import Application
 
 from argos.controllers.base import ControllerBase
-from argos.controllers.utils import call_by_slice, parse_tracks
+from argos.controllers.utils import call_by_slice
 from argos.controllers.visitors import PlaylistTrackNameFix
 from argos.dto import PlaylistDTO
 from argos.message import Message, MessageType, consume
@@ -206,7 +206,7 @@ async def _complete_playlist_from(
                 params=track_uris,
             )
             parsed_tracks: List[TrackModel] = []
-            for tracks in parse_tracks(
+            for tracks in self.helper.parse_tracks(
                 found_tracks_dto, visitors=[PlaylistTrackNameFix(playlist_dto)]
             ).values():
                 parsed_tracks += tracks
@@ -269,7 +269,7 @@ async def __complete_history_playlist(self) -> None:
         )
 
         parsed_history_tracks_with_duplicates: List[TrackModel] = []
-        parsed_history_tracks: Dict[str, List[TrackModel]] = parse_tracks(
+        parsed_history_tracks: Dict[str, List[TrackModel]] = self.helper.parse_tracks(
             history_tracks_dto
         )
         for history_item in history:

argos/controllers/tracklist.py

@@ -81,7 +81,7 @@ async def get_tracklist(self, message: Message) -> None:
 
         tl_tracks = (
             [
-                TracklistTrackModel.factory(tl_track_dto)
+                self.helper.convert_tl_track(tl_track_dto)
                 for tl_track_dto in tl_tracks_dto
             ]
             if tl_tracks_dto is not None

argos/controllers/utils.py

@@ -1,10 +1,7 @@
 import logging
-from collections import defaultdict
-from typing import Any, Callable, Coroutine, Dict, List, Mapping, Optional, Sequence
+from typing import Any, Callable, Coroutine, Dict, List, Mapping, Optional
 
 from argos.controllers.progress import ProgressNotifierProtocol
-from argos.dto import TrackDTO
-from argos.model import TrackModel
 
 LOGGER = logging.getLogger(__name__)
 
@@ -50,35 +47,3 @@ async def call_by_slice(
             break
         result.update(ith_result)
     return result
-
-
-def parse_tracks(
-    tracks_dto: Mapping[str, Sequence[TrackDTO]],
-    *,
-    visitors: Optional[Sequence[Callable[[str, TrackDTO], None]]] = None,
-) -> Dict[str, List[TrackModel]]:
-    """Parse a track list.
-
-    Keys in ``track_dto`` can be album URIs or track URIs (when fetching
-    details of playlist tracks).
-
-    Args:
-        tracks_dto: Track data transfer objects to parse.
-
-        visitors: An optional list of callable to be called on each
-            visited track.
-
-    Returns:
-        Dict of list of ``TrackModel``.
-
-    """
-    parsed_tracks: Dict[str, List[TrackModel]] = defaultdict(list)
-    for uri in tracks_dto:
-        for track_dto in tracks_dto[uri]:
-            if visitors is not None:
-                for visitor in visitors:
-                    visitor(uri, track_dto)
-
-            parsed_tracks[uri].append(TrackModel.factory(track_dto))
-
-    return parsed_tracks

argos/controllers/visitors.py

@@ -1,3 +1,8 @@
+"""Visitors for tracks DTOs.
+
+See ``argos.controllers.helper.ModelHelper.parse_tracks()``.
+
+"""
 import logging
 from collections import Counter, defaultdict
 from typing import Dict, List, Optional
@@ -8,11 +13,7 @@
 
 
 class LengthAcc:
-    """Visitor accumulating track length by uri.
-
-    See ``argos.utils.parse_tracks()``.
-
-    """
+    """Visitor accumulating track length by uri."""
 
     def __init__(self):
         self.length: Dict[str, int] = defaultdict(int)
@@ -35,7 +36,7 @@ class AlbumMetadataCollector:
     the names of the artists of the album tracks are collected and the
     most common name is returned.
 
-    See ``argos.utils.parse_tracks()``."""
+    """
 
     def __init__(self):
         self._name: Dict[str, str] = {}

argos/info.py

@@ -2,7 +2,7 @@
 import logging
 import urllib.parse
 from enum import Enum
-from typing import TYPE_CHECKING, Dict, List, Optional, Tuple
+from typing import TYPE_CHECKING, Dict, List, Mapping, Optional, Sequence, Tuple
 
 import aiohttp
 from gi.repository import GLib, GObject
@@ -132,7 +132,7 @@ async def _get_sitelinks_from_wikidata(
         return sitelinks
 
     def _build_preferred_abstract_url(
-        self, sitelinks: Dict[str, Dict[str, str]]
+        self, sitelinks: Mapping[str, Mapping[str, str]]
     ) -> Optional[str]:
         language_names = [
             lang
@@ -241,7 +241,7 @@ async def _get_album_abstract(
     async def _get_artist_abstract(
         self,
         session: aiohttp.ClientSession,
-        artist_mbids: List[str],
+        artist_mbids: Sequence[str],
     ) -> Optional[str]:
         raw_abstracts = []
         for artist_mbid in artist_mbids:
@@ -326,3 +326,32 @@ async def get_album_information(
                 )
 
         return album_abstract, artist_abstract
+
+    async def get_artist_information(self, artist_mbid: str) -> Optional[str]:
+        """Return short text for artist with given artist MBID.
+
+        The text is the abstract of the Wikipedia page dedicated to the artist.
+
+        Wikidata API is used to search for Wikipedia pages associated to MBIDs.
+
+        Finally, Wikipedia API is used to retrieve pages abstracts.
+
+        Page selection is expected to match current locale language or
+        English.
+
+        """
+        if not artist_mbid:
+            return None
+
+        artist_abstract = None
+        async with self._http_session_manager.get_session() as session:
+            try:
+                artist_abstract = await self._get_artist_abstract(
+                    session, [artist_mbid]
+                )
+            except aiohttp.ClientError as err:
+                LOGGER.error(
+                    f"Failed to request abstract for artist MBID {artist_mbid!r}, {err}"
+                )
+
+        return artist_abstract

argos/model/track.py

@@ -2,8 +2,6 @@
 
 from gi.repository import GObject
 
-from argos.dto import TrackDTO
-
 
 def compare_tracks_by_name_func(
     a: "TrackModel",
@@ -44,17 +42,3 @@ class TrackModel(GObject.Object):
     last_played = GObject.Property(type=GObject.TYPE_DOUBLE, default=-1)
     image_path = GObject.Property(type=str)
     image_uri = GObject.Property(type=str)
-
-    @staticmethod
-    def factory(dto: TrackDTO) -> "TrackModel":
-        track = TrackModel(
-            uri=dto.uri,
-            name=dto.name,
-            track_no=dto.track_no if dto.track_no is not None else -1,
-            disc_no=dto.disc_no if dto.disc_no is not None else 1,
-            length=dto.length if dto.length is not None else -1,
-            artist_name=dto.artists[0].name if len(dto.artists) > 0 else "",
-            album_name=dto.album.name if dto.album is not None else "",
-            last_modified=dto.last_modified if dto.last_modified is not None else -1,
-        )
-        return track

argos/model/tracklist.py

@@ -2,7 +2,6 @@
 
 from gi.repository import Gio, GObject
 
-from argos.dto import TlTrackDTO
 from argos.model.track import TrackModel
 from argos.model.utils import WithThreadSafePropertySetter
 
@@ -13,12 +12,6 @@ class TracklistTrackModel(GObject.Object):
     tlid = GObject.Property(type=int)
     track = GObject.Property(type=TrackModel)
 
-    @staticmethod
-    def factory(dto: TlTrackDTO) -> "TracklistTrackModel":
-        track = TrackModel.factory(dto.track)
-        tl_track = TracklistTrackModel(tlid=dto.tlid, track=track)
-        return tl_track
-
 
 class TracklistModel(WithThreadSafePropertySetter, GObject.Object):
     """Model for the tracklist.

tests/controllers/test_helper.py

@@ -0,0 +1,35 @@
+import json
+import pathlib
+import unittest
+from unittest.mock import Mock, call
+
+from argos.controllers.helper import ModelHelper
+from argos.dto import TrackDTO
+from argos.model.track import TrackModel
+
+
+def load_json_data(filename: str):
+    path = pathlib.Path(__file__).parent.parent / "data" / filename
+    with open(path) as fh:
+        data = json.load(fh)
+    return data
+
+
+class TestModelHelper(unittest.TestCase):
+    def test_parse_tracks(self):
+        track_dto = TrackDTO.factory(load_json_data("track.json"))
+        tracks_dto = {"local:album:md5:ff5c5b8f60a44e4c7d6f1bb53474e17b": [track_dto]}
+        tracks = ModelHelper().parse_tracks(tracks_dto)
+        self.assertListEqual([k for k in tracks_dto.keys()], [k for k in tracks.keys()])
+        self.assertIsInstance(
+            tracks["local:album:md5:ff5c5b8f60a44e4c7d6f1bb53474e17b"][0], TrackModel
+        )
+
+    def test_parse_tracks_with_visitor(self):
+        track_dto = TrackDTO.factory(load_json_data("track.json"))
+        tracks_dto = {"local:album:md5:ff5c5b8f60a44e4c7d6f1bb53474e17b": [track_dto]}
+        visitor = Mock()
+        ModelHelper().parse_tracks(tracks_dto, visitors=[visitor])
+        visitor.assert_called_once_with(
+            "local:album:md5:ff5c5b8f60a44e4c7d6f1bb53474e17b", track_dto
+        )