colin/sublime-music
1
0
Fork 0
Code Issues Pull Requests Packages Projects Releases Wiki Activity

Removed CacheManager from the face of the earth

Browse Source
This commit is contained in:
Sumner Evans
2020-05-13 22:00:35 -06:00
parent d54dcdbd8b
commit 8f07d1ec48
8 changed files with 36 additions and 2640 deletions
Download Patch File Download Diff File Expand all files Collapse all files

2
sublime/app.py
View File

@@ -41,7 +41,6 @@ except Exception:
from .adapters import AdapterManager, AlbumSearchQuery, Result
from .adapters.api_objects import Playlist, PlayQueue, Song
from .cache_manager import CacheManager
from .config import AppConfiguration, ReplayGainType
from .dbus import dbus_propagate, DBusManager
from .players import ChromecastPlayer, MPVPlayer, PlayerEvent
@@ -880,7 +879,6 @@ class SublimeMusicApp(Gtk.Application):
self.save_play_queue()
if self.dbus_manager:
self.dbus_manager.shutdown()
CacheManager.shutdown()
AdapterManager.shutdown()
# ########## HELPER METHODS ########## #

398
sublime/cache_manager.py
View File

@@ -1,398 +0,0 @@
import json
import logging
import os
import re
import shutil
import threading
from collections import defaultdict
from concurrent.futures import Future, ThreadPoolExecutor
from datetime import datetime
from enum import EnumMeta
from pathlib import Path
from typing import (
Any,
Callable,
DefaultDict,
Dict,
Generic,
List,
Optional,
Set,
TypeVar,
Union,
)
try:
import gi
gi.require_version("NM", "1.0")
from gi.repository import NM
networkmanager_imported = True
except Exception:
# I really don't care what kind of exception it is, all that matters is the
# import failed for some reason.
logging.warning(
"Unable to import NM from GLib. Detection of SSID will be disabled."
)
networkmanager_imported = False
from .config import AppConfiguration
from .server import Server
from .server.api_object import APIObject
from .server.api_objects import (
AlbumWithSongsID3,
Artist,
ArtistID3,
ArtistInfo2,
ArtistWithAlbumsID3,
Child,
Directory,
)
class Singleton(type):
"""
Metaclass for :class:`CacheManager` so that it can be used like a
singleton.
"""
def __getattr__(cls, name: str) -> Any:
if not CacheManager._instance:
return None
# If the cache has a function to do the thing we want, use it. If
# not, then go directly to the server (this is useful for things
# that just send data to the server.)
if hasattr(CacheManager._instance, name):
return getattr(CacheManager._instance, name)
else:
return getattr(CacheManager._instance.server, name)
return None
T = TypeVar("T")
class CacheManager(metaclass=Singleton):
"""
Handles everything related to caching metadata and song files.
"""
executor: ThreadPoolExecutor = ThreadPoolExecutor(max_workers=50)
should_exit: bool = False
class Result(Generic[T]):
# This needs to accept some way of:
# 1. getting data from the server to fulfill the request
# 2. coercing the data to the schema of the cachedb
# 3. queries for retriving the data from the cachedb
# All results should be retrieved using select statements from the DB
"""
A result from a CacheManager function. This is effectively a wrapper
around a Future, but it can also resolve immediately if the data
already exists.
"""
data: Optional[T] = None
future: Optional[Future] = None
on_cancel: Optional[Callable[[], None]] = None
@staticmethod
def from_data(data: T) -> "CacheManager.Result[T]":
result: "CacheManager.Result[T]" = CacheManager.Result()
result.data = data
return result
@staticmethod
def from_server(
download_fn: Callable[[], T],
before_download: Callable[[], Any] = None,
after_download: Callable[[T], Any] = None,
on_cancel: Callable[[], Any] = None,
) -> "CacheManager.Result[T]":
result: "CacheManager.Result[T]" = CacheManager.Result()
def future_fn() -> T:
if before_download:
before_download()
return download_fn()
result.future = CacheManager.executor.submit(future_fn)
result.on_cancel = on_cancel
if after_download is not None:
result.future.add_done_callback(
lambda f: after_download and after_download(f.result())
)
return result
def result(self) -> T:
if self.data is not None:
return self.data
if self.future is not None:
return self.future.result()
raise Exception(
"CacheManager.Result did not have either a data or future " "member."
)
def add_done_callback(self, fn: Callable, *args):
if self.future is not None:
self.future.add_done_callback(fn, *args)
else:
# Run the function immediately if it's not a future.
fn(self, *args)
def cancel(self) -> bool:
if self.on_cancel is not None:
self.on_cancel()
if self.future is not None:
return self.future.cancel()
return True
@property
def is_future(self) -> bool:
return self.future is not None
@staticmethod
def ready() -> bool:
return CacheManager._instance is not None
@staticmethod
def shutdown():
logging.info("CacheManager shutdown start")
CacheManager.should_exit = True
CacheManager.executor.shutdown()
CacheManager._instance.save_cache_info()
logging.info("CacheManager shutdown complete")
class CacheEncoder(json.JSONEncoder):
def default(self, obj: Any) -> Optional[Union[int, List, Dict]]:
"""
Encodes Python objects to JSON.
- ``datetime`` objects are converted to UNIX timestamps (``int``)
- ``set`` objects are converted to ``list`` objects
- ``APIObject`` objects are recursively encoded
- ``EnumMeta`` objects are ignored
- everything else is encoded using the default encoder
"""
if type(obj) == datetime:
return int(obj.timestamp() * 1000)
elif type(obj) == set:
return list(obj)
elif isinstance(obj, APIObject):
return {k: v for k, v in obj.__dict__.items() if v is not None}
elif isinstance(obj, EnumMeta):
return None
return json.JSONEncoder.default(self, obj)
class __CacheManagerInternal:
# Thread lock for preventing threads from overriding the state while
# it's being saved.
cache_lock = threading.Lock()
cache: DefaultDict[str, Any] = defaultdict(dict)
permanently_cached_paths: Set[str] = set()
# The server instance.
server: Server
# TODO (#56): need to split out the song downloads and make them higher
# priority I think. Maybe even need to just make this a priority queue.
download_set_lock = threading.Lock()
current_downloads: Set[str] = set()
def __init__(self, app_config: AppConfiguration):
self.app_config = app_config
assert self.app_config.server is not None
self.app_config.server
# If connected to the "Local Network SSID", use the "Local Network
# Address" instead of the "Server Address".
hostname = self.app_config.server.server_address
if self.app_config.server.local_network_ssid in self.current_ssids:
hostname = self.app_config.server.local_network_address
self.server = Server(
name=self.app_config.server.name,
hostname=hostname,
username=self.app_config.server.username,
password=self.app_config.server.password,
disable_cert_verify=self.app_config.server.disable_cert_verify,
)
self.download_limiter_semaphore = threading.Semaphore(
self.app_config.concurrent_download_limit
)
self.load_cache_info()
@property
def current_ssids(self) -> Set[str]:
if not networkmanager_imported:
return set()
self.networkmanager_client = NM.Client.new()
self.nmclient_initialized = False
self._current_ssids: Set[str] = set()
if not self.nmclient_initialized:
# Only look at the active WiFi connections.
for ac in self.networkmanager_client.get_active_connections():
if ac.get_connection_type() != "802-11-wireless":
continue
devs = ac.get_devices()
if len(devs) != 1:
continue
if devs[0].get_device_type() != NM.DeviceType.WIFI:
continue
self._current_ssids.add(ac.get_id())
return self._current_ssids
def load_cache_info(self):
cache_meta_file = self.calculate_abs_path(".cache_meta")
meta_json = {}
if cache_meta_file.exists():
with open(cache_meta_file, "r") as f:
try:
meta_json = json.load(f)
except json.decoder.JSONDecodeError:
# Just continue with the default meta_json.
logging.warning("Unable to load cache", stack_info=True)
cache_version = meta_json.get("version", 0)
if cache_version < 1:
logging.info("Migrating cache to version 1.")
cover_art_re = re.compile(r"(\d+)_(\d+)")
abs_path = self.calculate_abs_path("cover_art/")
abs_path.mkdir(parents=True, exist_ok=True)
for cover_art_file in abs_path.iterdir():
match = cover_art_re.match(cover_art_file.name)
if match:
art_id, dimensions = map(int, match.groups())
if dimensions == 1000:
no_dimens = cover_art_file.parent.joinpath("{art_id}")
logging.info(f"Moving {cover_art_file} to {no_dimens}")
shutil.move(cover_art_file, no_dimens)
else:
logging.info(f"Deleting {cover_art_file}")
cover_art_file.unlink()
self.cache["version"] = 1
cache_configs = [
("song_details", Child, dict),
# Non-ID3 caches
("music_directories", Directory, dict),
("indexes", Artist, list),
# ID3 caches
("albums", AlbumWithSongsID3, "dict-list"),
("album_details", AlbumWithSongsID3, dict),
("artists", ArtistID3, list),
("artist_details", ArtistWithAlbumsID3, dict),
("artist_infos", ArtistInfo2, dict),
]
for name, type_name, default in cache_configs:
if default == list:
self.cache[name] = [
type_name.from_json(x) for x in meta_json.get(name) or []
]
elif default == dict:
self.cache[name] = {
id: type_name.from_json(x)
for id, x in (meta_json.get(name) or {}).items()
}
elif default == "dict-list":
self.cache[name] = {
n: [type_name.from_json(x) for x in xs]
for n, xs in (meta_json.get(name) or {}).items()
}
def save_cache_info(self):
os.makedirs(self.app_config.cache_location, exist_ok=True)
cache_meta_file = self.calculate_abs_path(".cache_meta")
os.makedirs(os.path.dirname(cache_meta_file), exist_ok=True)
with open(cache_meta_file, "w+") as f, self.cache_lock:
f.write(json.dumps(self.cache, indent=2, cls=CacheManager.CacheEncoder))
def save_file(self, absolute_path: Path, data: bytes):
# Make the necessary directories and write to file.
os.makedirs(absolute_path.parent, exist_ok=True)
with open(absolute_path, "wb+") as f:
f.write(data)
def calculate_abs_path(self, *relative_paths) -> Path:
assert self.app_config.server is not None
return Path(self.app_config.cache_location).joinpath(
self.app_config.server.strhash(), *relative_paths
)
@staticmethod
def create_future(fn: Callable, *args) -> Future:
"""Creates a future on the CacheManager's executor."""
return CacheManager.executor.submit(fn, *args)
def get_indexes(
self,
before_download: Callable[[], None] = lambda: None,
force: bool = False,
) -> "CacheManager.Result[List[Artist]]":
cache_name = "indexes"
if self.cache.get(cache_name) and not force:
return CacheManager.Result.from_data(self.cache[cache_name])
def download_fn() -> List[Artist]:
artists: List[Artist] = []
for index in self.server.get_indexes().index:
artists.extend(index.artist)
return artists
def after_download(artists: List[Artist]):
with self.cache_lock:
self.cache[cache_name] = artists
self.save_cache_info()
return CacheManager.Result.from_server(
download_fn,
before_download=before_download,
after_download=after_download,
)
def get_music_directory(
self,
id: int,
before_download: Callable[[], None] = lambda: None,
force: bool = False,
) -> "CacheManager.Result[Directory]":
cache_name = "music_directories"
if id in self.cache.get(cache_name, {}) and not force:
return CacheManager.Result.from_data(self.cache[cache_name][id])
def after_download(directory: Directory):
with self.cache_lock:
self.cache[cache_name][id] = directory
self.save_cache_info()
return CacheManager.Result.from_server(
lambda: self.server.get_music_directory(id),
before_download=before_download,
after_download=after_download,
)
_instance: Optional[__CacheManagerInternal] = None
def __init__(self):
raise Exception("Do not instantiate the CacheManager.")
@staticmethod
def reset(app_config: AppConfiguration):
CacheManager._instance = CacheManager.__CacheManagerInternal(app_config)

2
sublime/config.py
View File

@@ -153,10 +153,8 @@ class AppConfiguration:
self._state = UIState()
# Do the import in the function to avoid circular imports.
from sublime.cache_manager import CacheManager
from sublime.adapters import AdapterManager
CacheManager.reset(self)
AdapterManager.reset(self)
@property

6
sublime/server/__init__.py
View File

@@ -1,6 +0,0 @@
"""
This module defines a stateless server which interops with the Subsonic API.
"""
from .server import Server
__all__ = ("Server",)

20
sublime/server/api_object.py
View File

@@ -1,20 +0,0 @@
# from dataclasses import Field, fields
from typing import Any, Dict
from sublime.from_json import from_json as _from_json
class APIObject:
"""Defines the base class for objects coming from the Subsonic API."""
@classmethod
def from_json(cls, data: Dict[str, Any]) -> Any:
"""
Creates an :class:`APIObject` by taking the ``data`` and passing it to the class
constructor and then recursively calling ``from_json`` on all of the fields.
``data`` just has to be a well-formed :class:`dict`, so it can come from the
JSON or XML APIs.
:param data: a Python dictionary representation of the data to deserialize
"""
return _from_json(cls, data)

936
sublime/server/api_objects.py
View File

@@ -1,936 +0,0 @@
"""
WARNING: AUTOGENERATED FILE
This file was generated by the api_object_generator.py
script. Do not modify this file directly, rather modify the
script or run it on a new API version.
"""
from dataclasses import dataclass, field
from datetime import datetime
from enum import Enum
from typing import Any, List, Optional
from sublime.server.api_object import APIObject
@dataclass(frozen=True)
class AlbumInfo(APIObject):
notes: List[str] = field(default_factory=list)
musicBrainzId: List[str] = field(default_factory=list)
lastFmUrl: List[str] = field(default_factory=list)
smallImageUrl: List[str] = field(default_factory=list)
mediumImageUrl: List[str] = field(default_factory=list)
largeImageUrl: List[str] = field(default_factory=list)
value: Optional[str] = None
def get(self, key: str, default: Any = None) -> Any:
return getattr(self, key, default)
@dataclass(frozen=True)
class AverageRating(APIObject, float):
pass
class MediaType(APIObject, Enum):
MUSIC = "music"
PODCAST = "podcast"
AUDIOBOOK = "audiobook"
VIDEO = "video"
def get(self, key: str, default: Any = None) -> Any:
return getattr(self, key, default)
@dataclass(frozen=True)
class UserRating(APIObject, int):
pass
@dataclass(frozen=True)
class Child(APIObject):
id: str
isDir: bool
title: str
value: Optional[str] = None
parent: Optional[str] = None
album: Optional[str] = None
artist: Optional[str] = None
track: Optional[int] = None
year: Optional[int] = None
genre: Optional[str] = None
coverArt: Optional[str] = None
size: Optional[int] = None
contentType: Optional[str] = None
suffix: Optional[str] = None
transcodedContentType: Optional[str] = None
transcodedSuffix: Optional[str] = None
duration: Optional[int] = None
bitRate: Optional[int] = None
path: Optional[str] = None
isVideo: Optional[bool] = None
userRating: Optional[UserRating] = None
averageRating: Optional[AverageRating] = None
playCount: Optional[int] = None
discNumber: Optional[int] = None
created: Optional[datetime] = None
starred: Optional[datetime] = None
albumId: Optional[str] = None
artistId: Optional[str] = None
type: Optional[MediaType] = None
bookmarkPosition: Optional[int] = None
originalWidth: Optional[int] = None
originalHeight: Optional[int] = None
def get(self, key: str, default: Any = None) -> Any:
return getattr(self, key, default)
@dataclass(frozen=True)
class AlbumList(APIObject):
album: List[Child] = field(default_factory=list)
value: Optional[str] = None
def get(self, key: str, default: Any = None) -> Any:
return getattr(self, key, default)
@dataclass(frozen=True)
class AlbumID3(APIObject):
id: str
name: str
songCount: int
duration: int
created: datetime
value: Optional[str] = None
artist: Optional[str] = None
artistId: Optional[str] = None
coverArt: Optional[str] = None
playCount: Optional[int] = None
starred: Optional[datetime] = None
year: Optional[int] = None
genre: Optional[str] = None
def get(self, key: str, default: Any = None) -> Any:
return getattr(self, key, default)
@dataclass(frozen=True)
class AlbumList2(APIObject):
album: List[AlbumID3] = field(default_factory=list)
value: Optional[str] = None
def get(self, key: str, default: Any = None) -> Any:
return getattr(self, key, default)
@dataclass(frozen=True)
class AlbumWithSongsID3(APIObject):
id: str
name: str
songCount: int
duration: int
created: datetime
song: List[Child] = field(default_factory=list)
value: Optional[str] = None
artist: Optional[str] = None
artistId: Optional[str] = None
coverArt: Optional[str] = None
playCount: Optional[int] = None
starred: Optional[datetime] = None
year: Optional[int] = None
genre: Optional[str] = None
def get(self, key: str, default: Any = None) -> Any:
return getattr(self, key, default)
@dataclass(frozen=True)
class Artist(APIObject):
id: str
name: str
value: Optional[str] = None
artistImageUrl: Optional[str] = None
starred: Optional[datetime] = None
userRating: Optional[UserRating] = None
averageRating: Optional[AverageRating] = None
def get(self, key: str, default: Any = None) -> Any:
return getattr(self, key, default)
@dataclass(frozen=True)
class ArtistInfoBase(APIObject):
biography: List[str] = field(default_factory=list)
musicBrainzId: List[str] = field(default_factory=list)
lastFmUrl: List[str] = field(default_factory=list)
smallImageUrl: List[str] = field(default_factory=list)
mediumImageUrl: List[str] = field(default_factory=list)
largeImageUrl: List[str] = field(default_factory=list)
value: Optional[str] = None
def get(self, key: str, default: Any = None) -> Any:
return getattr(self, key, default)
@dataclass(frozen=True)
class ArtistInfo(APIObject):
similarArtist: List[Artist] = field(default_factory=list)
value: Optional[str] = None
biography: List[str] = field(default_factory=list)
musicBrainzId: List[str] = field(default_factory=list)
lastFmUrl: List[str] = field(default_factory=list)
smallImageUrl: List[str] = field(default_factory=list)
mediumImageUrl: List[str] = field(default_factory=list)
largeImageUrl: List[str] = field(default_factory=list)
def get(self, key: str, default: Any = None) -> Any:
return getattr(self, key, default)
@dataclass(frozen=True)
class ArtistID3(APIObject):
id: str
name: str
albumCount: int
value: Optional[str] = None
coverArt: Optional[str] = None
artistImageUrl: Optional[str] = None
starred: Optional[datetime] = None
def get(self, key: str, default: Any = None) -> Any:
return getattr(self, key, default)
@dataclass(frozen=True)
class ArtistInfo2(APIObject):
similarArtist: List[ArtistID3] = field(default_factory=list)
value: Optional[str] = None
biography: List[str] = field(default_factory=list)
musicBrainzId: List[str] = field(default_factory=list)
lastFmUrl: List[str] = field(default_factory=list)
smallImageUrl: List[str] = field(default_factory=list)
mediumImageUrl: List[str] = field(default_factory=list)
largeImageUrl: List[str] = field(default_factory=list)
def get(self, key: str, default: Any = None) -> Any:
return getattr(self, key, default)
@dataclass(frozen=True)
class ArtistWithAlbumsID3(APIObject):
id: str
name: str
albumCount: int
album: List[AlbumID3] = field(default_factory=list)
value: Optional[str] = None
coverArt: Optional[str] = None
artistImageUrl: Optional[str] = None
starred: Optional[datetime] = None
def get(self, key: str, default: Any = None) -> Any:
return getattr(self, key, default)
@dataclass(frozen=True)
class IndexID3(APIObject):
name: str
artist: List[ArtistID3] = field(default_factory=list)
value: Optional[str] = None
def get(self, key: str, default: Any = None) -> Any:
return getattr(self, key, default)
@dataclass(frozen=True)
class ArtistsID3(APIObject):
ignoredArticles: str
index: List[IndexID3] = field(default_factory=list)
value: Optional[str] = None
def get(self, key: str, default: Any = None) -> Any:
return getattr(self, key, default)
@dataclass(frozen=True)
class Bookmark(APIObject):
position: int
username: str
created: datetime
changed: datetime
entry: List[Child] = field(default_factory=list)
value: Optional[str] = None
comment: Optional[str] = None
def get(self, key: str, default: Any = None) -> Any:
return getattr(self, key, default)
@dataclass(frozen=True)
class Bookmarks(APIObject):
bookmark: List[Bookmark] = field(default_factory=list)
value: Optional[str] = None
def get(self, key: str, default: Any = None) -> Any:
return getattr(self, key, default)
@dataclass(frozen=True)
class ChatMessage(APIObject):
username: str
time: int
message: str
value: Optional[str] = None
def get(self, key: str, default: Any = None) -> Any:
return getattr(self, key, default)
@dataclass(frozen=True)
class ChatMessages(APIObject):
chatMessage: List[ChatMessage] = field(default_factory=list)
value: Optional[str] = None
def get(self, key: str, default: Any = None) -> Any:
return getattr(self, key, default)
@dataclass(frozen=True)
class Directory(APIObject):
id: str
name: str
child: List[Child] = field(default_factory=list)
value: Optional[str] = None
parent: Optional[str] = None
starred: Optional[datetime] = None
userRating: Optional[UserRating] = None
averageRating: Optional[AverageRating] = None
playCount: Optional[int] = None
def get(self, key: str, default: Any = None) -> Any:
return getattr(self, key, default)
@dataclass(frozen=True)
class Error(APIObject):
code: int
value: Optional[str] = None
message: Optional[str] = None
def get(self, key: str, default: Any = None) -> Any:
return getattr(self, key, default)
@dataclass(frozen=True)
class Genre(APIObject):
songCount: int
albumCount: int
value: Optional[str] = None
def get(self, key: str, default: Any = None) -> Any:
return getattr(self, key, default)
@dataclass(frozen=True)
class Genres(APIObject):
genre: List[Genre] = field(default_factory=list)
value: Optional[str] = None
def get(self, key: str, default: Any = None) -> Any:
return getattr(self, key, default)
@dataclass(frozen=True)
class Index(APIObject):
name: str
artist: List[Artist] = field(default_factory=list)
value: Optional[str] = None
def get(self, key: str, default: Any = None) -> Any:
return getattr(self, key, default)
@dataclass(frozen=True)
class Indexes(APIObject):
lastModified: int
ignoredArticles: str
shortcut: List[Artist] = field(default_factory=list)
index: List[Index] = field(default_factory=list)
child: List[Child] = field(default_factory=list)
value: Optional[str] = None
def get(self, key: str, default: Any = None) -> Any:
return getattr(self, key, default)
@dataclass(frozen=True)
class InternetRadioStation(APIObject):
id: str
name: str
streamUrl: str
value: Optional[str] = None
homePageUrl: Optional[str] = None
def get(self, key: str, default: Any = None) -> Any:
return getattr(self, key, default)
@dataclass(frozen=True)
class InternetRadioStations(APIObject):
internetRadioStation: List[InternetRadioStation] = field(default_factory=list)
value: Optional[str] = None
def get(self, key: str, default: Any = None) -> Any:
return getattr(self, key, default)
@dataclass(frozen=True)
class JukeboxStatus(APIObject):
currentIndex: int
playing: bool
gain: float
value: Optional[str] = None
position: Optional[int] = None
def get(self, key: str, default: Any = None) -> Any:
return getattr(self, key, default)
@dataclass(frozen=True)
class JukeboxPlaylist(APIObject):
currentIndex: int
playing: bool
gain: float
entry: List[Child] = field(default_factory=list)
value: Optional[str] = None
position: Optional[int] = None
def get(self, key: str, default: Any = None) -> Any:
return getattr(self, key, default)
@dataclass(frozen=True)
class License(APIObject):
valid: bool
value: Optional[str] = None
email: Optional[str] = None
licenseExpires: Optional[datetime] = None
trialExpires: Optional[datetime] = None
def get(self, key: str, default: Any = None) -> Any:
return getattr(self, key, default)
@dataclass(frozen=True)
class Lyrics(APIObject):
artist: Optional[str] = None
value: Optional[str] = None
title: Optional[str] = None
def get(self, key: str, default: Any = None) -> Any:
return getattr(self, key, default)
@dataclass(frozen=True)
class MusicFolder(APIObject):
id: int
value: Optional[str] = None
name: Optional[str] = None
def get(self, key: str, default: Any = None) -> Any:
return getattr(self, key, default)
@dataclass(frozen=True)
class MusicFolders(APIObject):
musicFolder: List[MusicFolder] = field(default_factory=list)
value: Optional[str] = None
def get(self, key: str, default: Any = None) -> Any:
return getattr(self, key, default)
class PodcastStatus(APIObject, Enum):
NEW = "new"
DOWNLOADING = "downloading"
COMPLETED = "completed"
ERROR = "error"
DELETED = "deleted"
SKIPPED = "skipped"
def get(self, key: str, default: Any = None) -> Any:
return getattr(self, key, default)
@dataclass(frozen=True)
class PodcastEpisode(APIObject):
channelId: str
status: PodcastStatus
id: str
isDir: bool
title: str
streamId: Optional[str] = None
description: Optional[str] = None
publishDate: Optional[datetime] = None
value: Optional[str] = None
parent: Optional[str] = None
album: Optional[str] = None
artist: Optional[str] = None
track: Optional[int] = None
year: Optional[int] = None
genre: Optional[str] = None
coverArt: Optional[str] = None
size: Optional[int] = None
contentType: Optional[str] = None
suffix: Optional[str] = None
transcodedContentType: Optional[str] = None
transcodedSuffix: Optional[str] = None
duration: Optional[int] = None
bitRate: Optional[int] = None
path: Optional[str] = None
isVideo: Optional[bool] = None
userRating: Optional[UserRating] = None
averageRating: Optional[AverageRating] = None
playCount: Optional[int] = None
discNumber: Optional[int] = None
created: Optional[datetime] = None
starred: Optional[datetime] = None
albumId: Optional[str] = None
artistId: Optional[str] = None
type: Optional[MediaType] = None
bookmarkPosition: Optional[int] = None
originalWidth: Optional[int] = None
originalHeight: Optional[int] = None
def get(self, key: str, default: Any = None) -> Any:
return getattr(self, key, default)
@dataclass(frozen=True)
class NewestPodcasts(APIObject):
episode: List[PodcastEpisode] = field(default_factory=list)
value: Optional[str] = None
def get(self, key: str, default: Any = None) -> Any:
return getattr(self, key, default)
@dataclass(frozen=True)
class NowPlayingEntry(APIObject):
username: str
minutesAgo: int
playerId: int
id: str
isDir: bool
title: str
playerName: Optional[str] = None
value: Optional[str] = None
parent: Optional[str] = None
album: Optional[str] = None
artist: Optional[str] = None
track: Optional[int] = None
year: Optional[int] = None
genre: Optional[str] = None
coverArt: Optional[str] = None
size: Optional[int] = None
contentType: Optional[str] = None
suffix: Optional[str] = None
transcodedContentType: Optional[str] = None
transcodedSuffix: Optional[str] = None
duration: Optional[int] = None
bitRate: Optional[int] = None
path: Optional[str] = None
isVideo: Optional[bool] = None
userRating: Optional[UserRating] = None
averageRating: Optional[AverageRating] = None
playCount: Optional[int] = None
discNumber: Optional[int] = None
created: Optional[datetime] = None
starred: Optional[datetime] = None
albumId: Optional[str] = None
artistId: Optional[str] = None
type: Optional[MediaType] = None
bookmarkPosition: Optional[int] = None
originalWidth: Optional[int] = None
originalHeight: Optional[int] = None
def get(self, key: str, default: Any = None) -> Any:
return getattr(self, key, default)
@dataclass(frozen=True)
class NowPlaying(APIObject):
entry: List[NowPlayingEntry] = field(default_factory=list)
value: Optional[str] = None
def get(self, key: str, default: Any = None) -> Any:
return getattr(self, key, default)
@dataclass(frozen=True)
class PlayQueue(APIObject):
username: str
changed: datetime
changedBy: str
entry: List[Child] = field(default_factory=list)
value: Optional[str] = None
current: Optional[int] = None
position: Optional[int] = None
def get(self, key: str, default: Any = None) -> Any:
return getattr(self, key, default)
@dataclass(frozen=True)
class Playlist(APIObject):
id: str
name: str
songCount: int
duration: int
created: datetime
changed: datetime
allowedUser: List[str] = field(default_factory=list)
value: Optional[str] = None
comment: Optional[str] = None
owner: Optional[str] = None
public: Optional[bool] = None
coverArt: Optional[str] = None
def get(self, key: str, default: Any = None) -> Any:
return getattr(self, key, default)
@dataclass(frozen=True)
class PlaylistWithSongs(APIObject):
id: str
name: str
songCount: int
duration: int
created: datetime
changed: datetime
entry: List[Child] = field(default_factory=list)
value: Optional[str] = None
allowedUser: List[str] = field(default_factory=list)
comment: Optional[str] = None
owner: Optional[str] = None
public: Optional[bool] = None
coverArt: Optional[str] = None
def get(self, key: str, default: Any = None) -> Any:
return getattr(self, key, default)
@dataclass(frozen=True)
class Playlists(APIObject):
playlist: List[Playlist] = field(default_factory=list)
value: Optional[str] = None
def get(self, key: str, default: Any = None) -> Any:
return getattr(self, key, default)
@dataclass(frozen=True)
class PodcastChannel(APIObject):
id: str
url: str
status: PodcastStatus
episode: List[PodcastEpisode] = field(default_factory=list)
value: Optional[str] = None
title: Optional[str] = None
description: Optional[str] = None
coverArt: Optional[str] = None
originalImageUrl: Optional[str] = None
errorMessage: Optional[str] = None
def get(self, key: str, default: Any = None) -> Any:
return getattr(self, key, default)
@dataclass(frozen=True)
class Podcasts(APIObject):
channel: List[PodcastChannel] = field(default_factory=list)
value: Optional[str] = None
def get(self, key: str, default: Any = None) -> Any:
return getattr(self, key, default)
class ResponseStatus(APIObject, Enum):
OK = "ok"
FAILED = "failed"
def get(self, key: str, default: Any = None) -> Any:
return getattr(self, key, default)
@dataclass(frozen=True)
class ScanStatus(APIObject):
scanning: bool
value: Optional[str] = None
count: Optional[int] = None
def get(self, key: str, default: Any = None) -> Any:
return getattr(self, key, default)
@dataclass(frozen=True)
class SearchResult(APIObject):
offset: int
totalHits: int
match: List[Child] = field(default_factory=list)
value: Optional[str] = None
def get(self, key: str, default: Any = None) -> Any:
return getattr(self, key, default)
@dataclass(frozen=True)
class SearchResult2(APIObject):
artist: List[Artist] = field(default_factory=list)
album: List[Child] = field(default_factory=list)
song: List[Child] = field(default_factory=list)
value: Optional[str] = None
def get(self, key: str, default: Any = None) -> Any:
return getattr(self, key, default)
@dataclass(frozen=True)
class SearchResult3(APIObject):
artist: List[ArtistID3] = field(default_factory=list)
album: List[AlbumID3] = field(default_factory=list)
song: List[Child] = field(default_factory=list)
value: Optional[str] = None
def get(self, key: str, default: Any = None) -> Any:
return getattr(self, key, default)
@dataclass(frozen=True)
class Share(APIObject):
id: str
url: str
username: str
created: datetime
visitCount: int
entry: List[Child] = field(default_factory=list)
value: Optional[str] = None
description: Optional[str] = None
expires: Optional[datetime] = None
lastVisited: Optional[datetime] = None
def get(self, key: str, default: Any = None) -> Any:
return getattr(self, key, default)
@dataclass(frozen=True)
class Shares(APIObject):
share: List[Share] = field(default_factory=list)
value: Optional[str] = None
def get(self, key: str, default: Any = None) -> Any:
return getattr(self, key, default)
@dataclass(frozen=True)
class SimilarSongs(APIObject):
song: List[Child] = field(default_factory=list)
value: Optional[str] = None
def get(self, key: str, default: Any = None) -> Any:
return getattr(self, key, default)
@dataclass(frozen=True)
class SimilarSongs2(APIObject):
song: List[Child] = field(default_factory=list)
value: Optional[str] = None
def get(self, key: str, default: Any = None) -> Any:
return getattr(self, key, default)
@dataclass(frozen=True)
class Songs(APIObject):
song: List[Child] = field(default_factory=list)
value: Optional[str] = None
def get(self, key: str, default: Any = None) -> Any:
return getattr(self, key, default)
@dataclass(frozen=True)
class Starred(APIObject):
artist: List[Artist] = field(default_factory=list)
album: List[Child] = field(default_factory=list)
song: List[Child] = field(default_factory=list)
value: Optional[str] = None
def get(self, key: str, default: Any = None) -> Any:
return getattr(self, key, default)
@dataclass(frozen=True)
class Starred2(APIObject):
artist: List[ArtistID3] = field(default_factory=list)
album: List[AlbumID3] = field(default_factory=list)
song: List[Child] = field(default_factory=list)
value: Optional[str] = None
def get(self, key: str, default: Any = None) -> Any:
return getattr(self, key, default)
@dataclass(frozen=True)
class TopSongs(APIObject):
song: List[Child] = field(default_factory=list)
value: Optional[str] = None
def get(self, key: str, default: Any = None) -> Any:
return getattr(self, key, default)
@dataclass(frozen=True)
class User(APIObject):
username: str
scrobblingEnabled: bool
adminRole: bool
settingsRole: bool
downloadRole: bool
uploadRole: bool
playlistRole: bool
coverArtRole: bool
commentRole: bool
podcastRole: bool
streamRole: bool
jukeboxRole: bool
shareRole: bool
videoConversionRole: bool
folder: List[int] = field(default_factory=list)
value: Optional[str] = None
email: Optional[str] = None
maxBitRate: Optional[int] = None
avatarLastChanged: Optional[datetime] = None
def get(self, key: str, default: Any = None) -> Any:
return getattr(self, key, default)
@dataclass(frozen=True)
class Users(APIObject):
user: List[User] = field(default_factory=list)
value: Optional[str] = None
def get(self, key: str, default: Any = None) -> Any:
return getattr(self, key, default)
@dataclass(frozen=True)
class Version(APIObject, str):
pass
@dataclass(frozen=True)
class AudioTrack(APIObject):
id: str
value: Optional[str] = None
name: Optional[str] = None
languageCode: Optional[str] = None
def get(self, key: str, default: Any = None) -> Any:
return getattr(self, key, default)
@dataclass(frozen=True)
class Captions(APIObject):
id: str
value: Optional[str] = None
name: Optional[str] = None
def get(self, key: str, default: Any = None) -> Any:
return getattr(self, key, default)
@dataclass(frozen=True)
class VideoConversion(APIObject):
id: str
value: Optional[str] = None
bitRate: Optional[int] = None
audioTrackId: Optional[int] = None
def get(self, key: str, default: Any = None) -> Any:
return getattr(self, key, default)
@dataclass(frozen=True)
class VideoInfo(APIObject):
id: str
captions: List[Captions] = field(default_factory=list)
audioTrack: List[AudioTrack] = field(default_factory=list)
conversion: List[VideoConversion] = field(default_factory=list)
value: Optional[str] = None
def get(self, key: str, default: Any = None) -> Any:
return getattr(self, key, default)
@dataclass(frozen=True)
class Videos(APIObject):
video: List[Child] = field(default_factory=list)
value: Optional[str] = None
def get(self, key: str, default: Any = None) -> Any:
return getattr(self, key, default)
@dataclass(frozen=True)
class Response(APIObject):
musicFolders: Optional[MusicFolders] = None
indexes: Optional[Indexes] = None
directory: Optional[Directory] = None
genres: Optional[Genres] = None
artists: Optional[ArtistsID3] = None
artist: Optional[ArtistWithAlbumsID3] = None
album: Optional[AlbumWithSongsID3] = None
song: Optional[Child] = None
videos: Optional[Videos] = None
videoInfo: Optional[VideoInfo] = None
nowPlaying: Optional[NowPlaying] = None
searchResult: Optional[SearchResult] = None
searchResult2: Optional[SearchResult2] = None
searchResult3: Optional[SearchResult3] = None
playlists: Optional[Playlists] = None
playlist: Optional[PlaylistWithSongs] = None
jukeboxStatus: Optional[JukeboxStatus] = None
jukeboxPlaylist: Optional[JukeboxPlaylist] = None
license: Optional[License] = None
users: Optional[Users] = None
user: Optional[User] = None
chatMessages: Optional[ChatMessages] = None
albumList: Optional[AlbumList] = None
albumList2: Optional[AlbumList2] = None
randomSongs: Optional[Songs] = None
songsByGenre: Optional[Songs] = None
lyrics: Optional[Lyrics] = None
podcasts: Optional[Podcasts] = None
newestPodcasts: Optional[NewestPodcasts] = None
internetRadioStations: Optional[InternetRadioStations] = None
bookmarks: Optional[Bookmarks] = None
playQueue: Optional[PlayQueue] = None
shares: Optional[Shares] = None
starred: Optional[Starred] = None
starred2: Optional[Starred2] = None
albumInfo: Optional[AlbumInfo] = None
artistInfo: Optional[ArtistInfo] = None
artistInfo2: Optional[ArtistInfo2] = None
similarSongs: Optional[SimilarSongs] = None
similarSongs2: Optional[SimilarSongs2] = None
topSongs: Optional[TopSongs] = None
scanStatus: Optional[ScanStatus] = None
error: Optional[Error] = None
value: Optional[str] = None
status: Optional[ResponseStatus] = None
version: Optional[Version] = None
def get(self, key: str, default: Any = None) -> Any:
return getattr(self, key, default)

1239
sublime/server/server.py
View File

File diff suppressed because it is too large Load Diff

73
sublime/ui/configure_servers.py
View File

@@ -4,7 +4,6 @@ from typing import Any
from gi.repository import GObject, Gtk
from sublime.config import AppConfiguration, ServerConfiguration
from sublime.server import Server
from sublime.ui.common import EditFormDialog, IconButton
@@ -26,7 +25,7 @@ class EditServerDialog(EditFormDialog):
def __init__(self, *args, **kwargs):
test_server = Gtk.Button(label="Test Connection to Server")
test_server.connect("clicked", self.on_test_server_clicked)
# test_server.connect("clicked", self.on_test_server_clicked)
open_in_browser = Gtk.Button(label="Open in Browser")
open_in_browser.connect("clicked", self.on_open_in_browser_clicked)
@@ -35,43 +34,43 @@ class EditServerDialog(EditFormDialog):
super().__init__(*args, **kwargs)
def on_test_server_clicked(self, event: Any):
# Instantiate the server.
server_address = self.data["server_address"].get_text()
server = Server(
name=self.data["name"].get_text(),
hostname=server_address,
username=self.data["username"].get_text(),
password=self.data["password"].get_text(),
disable_cert_verify=self.data["disable_cert_verify"].get_active(),
)
# def on_test_server_clicked(self, event: Any):
# # Instantiate the server.
# server_address = self.data["server_address"].get_text()
# server = Server(
# name=self.data["name"].get_text(),
# hostname=server_address,
# username=self.data["username"].get_text(),
# password=self.data["password"].get_text(),
# disable_cert_verify=self.data["disable_cert_verify"].get_active(),
# )
# Try to ping, and show a message box with whether or not it worked.
try:
server.ping()
dialog = Gtk.MessageDialog(
transient_for=self,
message_type=Gtk.MessageType.INFO,
buttons=Gtk.ButtonsType.OK,
text="Connection to server successful.",
)
dialog.format_secondary_markup(
f"Connection to {server_address} successful."
)
except Exception as err:
dialog = Gtk.MessageDialog(
transient_for=self,
message_type=Gtk.MessageType.ERROR,
buttons=Gtk.ButtonsType.OK,
text="Connection to server unsuccessful.",
)
dialog.format_secondary_markup(
f"Connection to {server_address} resulted in the following "
f"error:\n\n{err}"
)
# # Try to ping, and show a message box with whether or not it worked.
# try:
# server.ping()
# dialog = Gtk.MessageDialog(
# transient_for=self,
# message_type=Gtk.MessageType.INFO,
# buttons=Gtk.ButtonsType.OK,
# text="Connection to server successful.",
# )
# dialog.format_secondary_markup(
# f"Connection to {server_address} successful."
# )
# except Exception as err:
# dialog = Gtk.MessageDialog(
# transient_for=self,
# message_type=Gtk.MessageType.ERROR,
# buttons=Gtk.ButtonsType.OK,
# text="Connection to server unsuccessful.",
# )
# dialog.format_secondary_markup(
# f"Connection to {server_address} resulted in the following "
# f"error:\n\n{err}"
# )
dialog.run()
dialog.destroy()
# dialog.run()
# dialog.destroy()
def on_open_in_browser_clicked(self, event: Any):
subprocess.call(["xdg-open", self.data["server_address"].get_text()])