@@ -10,8 +10,8 @@ import desert
|
||||
import marshmallow
|
||||
|
||||
|
||||
DataSub = typing.TypeVar("DataSub")
|
||||
DataDict = typing.NewType("DataDict", dict[str, typing.Any])
|
||||
DataSub = typing.TypeVar("DataSub", bound="BaseData")
|
||||
DataDict = typing.TypeVar("DataDict")
|
||||
|
||||
|
||||
class DataMeta(type):
|
||||
@@ -38,11 +38,12 @@ class DataMeta(type):
|
||||
data class in dictionary form.
|
||||
|
||||
Returns:
|
||||
DataDict: The TypedDict for this object.
|
||||
dict[str, object]: The TypedDict for this object.
|
||||
"""
|
||||
fields = {
|
||||
name: type_ if type_ is not datetime.datetime else str
|
||||
name: str if type_ is datetime.datetime else type_
|
||||
for name, type_ in cls.__annotations__.items()
|
||||
if name != "id"
|
||||
}
|
||||
return typing.TypedDict( # type: ignore [no-any-return,operator]
|
||||
f"{cls.__name__}Dict",
|
||||
@@ -51,17 +52,17 @@ class DataMeta(type):
|
||||
|
||||
|
||||
@dataclasses.dataclass
|
||||
class BaseData(typing.Generic[DataSub], metaclass=DataMeta):
|
||||
class BaseData(metaclass=DataMeta):
|
||||
"""Base class for data classes in the application."""
|
||||
|
||||
__sa_dataclass_metadata_key__ = "sa"
|
||||
|
||||
@classmethod
|
||||
def load(cls: type[BaseData[DataSub]], data: DataDict) -> DataSub:
|
||||
def load(cls: type[BaseData], data: DataDict) -> DataSub:
|
||||
"""Load the given data dictionary into a class instance."""
|
||||
return typing.cast(DataSub, cls.Schema.load(data))
|
||||
|
||||
def dump(self: BaseData[DataSub]) -> DataDict:
|
||||
def dump(self: BaseData) -> DataDict:
|
||||
"""Dump the class instance into a data dictionary."""
|
||||
cls = type(self)
|
||||
return typing.cast(DataDict, cls.Schema.dump(self))
|
||||
|
||||
@@ -12,7 +12,7 @@ mapper_registry = sqlalchemy.orm.registry()
|
||||
|
||||
@mapper_registry.mapped
|
||||
@dataclasses.dataclass
|
||||
class Track(base.BaseData["Track"]):
|
||||
class Track(base.BaseData):
|
||||
"""Model defining a Track object."""
|
||||
|
||||
__tablename__ = "tracks"
|
||||
|
||||
@@ -7,17 +7,19 @@ import dataclasses
|
||||
import datetime
|
||||
import functools
|
||||
import pathlib
|
||||
import typing
|
||||
|
||||
import yaml
|
||||
|
||||
from playlist.data import base
|
||||
from playlist.data import const
|
||||
|
||||
|
||||
__all__ = ["get", "modify"]
|
||||
|
||||
|
||||
@dataclasses.dataclass
|
||||
class CredentialSettings(base.BaseData["CredentialSettings"]):
|
||||
class CredentialSettings(base.BaseData):
|
||||
"""Credentials component for Settings object."""
|
||||
|
||||
baseurl: str
|
||||
@@ -28,11 +30,11 @@ class CredentialSettings(base.BaseData["CredentialSettings"]):
|
||||
from playlist.plex import server
|
||||
|
||||
data = server.get_creds()
|
||||
return cls.load(data)
|
||||
return typing.cast(CredentialSettings, cls.load(data))
|
||||
|
||||
|
||||
@dataclasses.dataclass
|
||||
class DownloadSettings(base.BaseData["DownloadSettings"]):
|
||||
class DownloadSettings(base.BaseData):
|
||||
"""DownloadSettingser options for Settings object."""
|
||||
|
||||
batch_size: int
|
||||
@@ -47,7 +49,7 @@ class DownloadSettings(base.BaseData["DownloadSettings"]):
|
||||
|
||||
|
||||
@dataclasses.dataclass
|
||||
class TrackSettings(base.BaseData["TrackSettings"]):
|
||||
class TrackSettings(base.BaseData):
|
||||
"""Track options for Settings object."""
|
||||
|
||||
duration: float
|
||||
@@ -58,7 +60,7 @@ class TrackSettings(base.BaseData["TrackSettings"]):
|
||||
|
||||
|
||||
@dataclasses.dataclass
|
||||
class PlaylistSettings(base.BaseData["PlaylistSettings"]):
|
||||
class PlaylistSettings(base.BaseData):
|
||||
"""Playlist options for Settings object."""
|
||||
|
||||
playtime: datetime.timedelta
|
||||
@@ -73,7 +75,7 @@ class PlaylistSettings(base.BaseData["PlaylistSettings"]):
|
||||
|
||||
|
||||
@dataclasses.dataclass
|
||||
class Settings(base.BaseData["Settings"]):
|
||||
class Settings(base.BaseData):
|
||||
"""Settings object, loaded from settings.yaml file."""
|
||||
|
||||
creds: CredentialSettings = dataclasses.field(
|
||||
@@ -95,7 +97,7 @@ class Settings(base.BaseData["Settings"]):
|
||||
"""Read the given YAML file and convert it into an object."""
|
||||
with filepath.open() as fp:
|
||||
data = yaml.safe_load(fp)
|
||||
return cls.load(data)
|
||||
return typing.cast(Settings, cls.load(data))
|
||||
|
||||
def yaml_write(self: Settings, filepath: pathlib.Path) -> None:
|
||||
"""Write this object as the given YAML file."""
|
||||
@@ -117,8 +119,8 @@ class Settings(base.BaseData["Settings"]):
|
||||
@functools.cache
|
||||
def get() -> Settings: # pragma: no cover
|
||||
"""Get the Settings object instance."""
|
||||
instance: Settings
|
||||
filepath = const.PATHS.CONFIG / const.SETTINGS_FILENAME
|
||||
instance: Settings
|
||||
try:
|
||||
instance = Settings.yaml_read(filepath)
|
||||
except FileNotFoundError:
|
||||
|
||||
@@ -33,10 +33,10 @@ def calc_delay(times: list[float], weights: list[float]) -> None:
|
||||
with settings.modify() as s:
|
||||
avg_time_per_batch = utils.harmonic_mean(times, weights=weights)
|
||||
delay = avg_time_per_batch / const.MAX_PROCESSES
|
||||
s.download.process_delay = round(delay, ndigits=6)
|
||||
s.download.process_delay = round(delay, ndigits=3)
|
||||
|
||||
|
||||
def calc_duration(durations: list[int]) -> None:
|
||||
def calc_duration(durations: list[float]) -> None:
|
||||
"""Calculate the average duration and max tracks to play each day.
|
||||
|
||||
This is based on the geometric mean of the durations of every track loaded from
|
||||
@@ -48,7 +48,7 @@ def calc_duration(durations: list[int]) -> None:
|
||||
"""
|
||||
with settings.modify() as s:
|
||||
avg_track_duration = statistics.geometric_mean(durations)
|
||||
s.track.duration = round(avg_track_duration, ndigits=6)
|
||||
s.track.duration = round(avg_track_duration / 1000, ndigits=3)
|
||||
playtime_seconds = s.playlist.playtime.total_seconds()
|
||||
s.playlist.max_tracks = int(playtime_seconds / avg_track_duration)
|
||||
|
||||
@@ -120,7 +120,7 @@ def get_creds() -> settings.CredentialSettings.Dict: # type: ignore [name-defin
|
||||
|
||||
account = plexapi.myplex.MyPlexAccount(username, password)
|
||||
plex = account.resource(server).connect()
|
||||
return {"baseurl": plex._baseurl, "token": plex._token}
|
||||
return settings.CredentialSettings.Dict(baseurl=plex._baseurl, token=plex._token)
|
||||
|
||||
|
||||
async def _downloader(
|
||||
@@ -192,7 +192,7 @@ def _track_dump(track: plexapi.audio.Track) -> models.Track.Dict: # type: ignor
|
||||
album_num=track.parentIndex,
|
||||
album=track.parentTitle,
|
||||
album_artist=track.grandparentTitle,
|
||||
duration=round(track.duration / 1000, ndigits=6),
|
||||
duration=track.duration,
|
||||
rating=track.userRating,
|
||||
comments=track.summary,
|
||||
added=str(track.addedAt),
|
||||
|
||||
Reference in New Issue
Block a user