cleanup and type hints

.gitignore

@@ -1,3 +1,4 @@
 *.pyc
 __pycache__
 config.py
+*.socket

grzegorz/api.py

@@ -1,8 +1,7 @@
-import asyncio
-from sanic import Blueprint, response
+from sanic import Request, Blueprint, response
 from sanic_openapi import doc
 from functools import wraps
-from . import mpv
+from .mpv import MPVControl
 from .playlist_data import PlaylistDataCache
 
 bp = Blueprint("grzegorz-api", strict_slashes=True)
@@ -37,15 +36,17 @@ def response_text(func):
         return response.text(body)
     return newfunc
 
-class APIError(Exception): pass
+class APIError(Exception):
+    pass
 
+# singleton
 PLAYLIST_DATA_CACHE = PlaylistDataCache(auto_fetch_data=True)
 
 #routes:
 @bp.get("")
 @doc.exclude(True)
 @response_text
-async def root(request):
+async def root(request: Request):
     return "Hello friend, I hope you're having a lovely day"
 
 @bp.post("/load")
@@ -53,7 +54,7 @@ async def root(request):
 @doc.consumes({"path": doc.String("Link to the resource to enqueue")}, required=True)
 @doc.consumes({"body":doc.Dictionary(description="Any data you want stored with the queued item")}, location="body")
 @response_json
-async def loadfile(request, mpv_control):
+async def loadfile(request: Request, mpv_control: MPVControl):
     if "path" not in request.args:
         raise APIError("No query parameter \"path\" provided")
     if request.json:
@@ -64,7 +65,7 @@ async def loadfile(request, mpv_control):
 @bp.get("/play")
 @doc.summary("Check whether the player is paused or playing")
 @response_json
-async def play_get(request, mpv_control):
+async def play_get(request: Request, mpv_control: MPVControl):
     value = await mpv_control.pause_get() == False
     return locals()
 
@@ -72,7 +73,7 @@ async def play_get(request, mpv_control):
 @doc.summary("Set whether the player is paused or playing")
 @doc.consumes({"play": doc.Boolean("Whether to be playing or not")})
 @response_json
-async def play_set(request, mpv_control):
+async def play_set(request: Request, mpv_control: MPVControl):
     if "play" not in request.args:
         raise APIError("No query parameter \"play\" provided")
     success = await mpv_control \
@@ -82,7 +83,7 @@ async def play_set(request, mpv_control):
 @bp.get("/volume")
 @doc.summary("Get the current player volume")
 @response_json
-async def volume_get(request, mpv_control):
+async def volume_get(request: Request, mpv_control: MPVControl):
     value = await mpv_control.volume_get()
     return locals()
 
@@ -90,7 +91,7 @@ async def volume_get(request, mpv_control):
 @doc.summary("Set the player volume")
 @doc.consumes({"volume": doc.Integer("A number between 0 and 100")})
 @response_json
-async def volume_set(request, mpv_control):
+async def volume_set(request: Request, mpv_control: MPVControl):
     if "volume" not in request.args:
         raise APIError("No query parameter \"volume\" provided")
     success = await mpv_control \
@@ -100,7 +101,7 @@ async def volume_set(request, mpv_control):
 @bp.get("/time")
 @doc.summary("Get current playback position")
 @response_json
-async def time_get(request, mpv_control):
+async def time_get(request: Request, mpv_control: MPVControl):
     value = {
         "current": await mpv_control.time_pos_get(),
         "left": await mpv_control.time_remaining_get(),
@@ -112,7 +113,7 @@ async def time_get(request, mpv_control):
 @doc.summary("Set playback position")
 @doc.consumes({"pos": doc.Float("Seconds to seek to"), "pos": doc.Integer("Percent to seek to")})
 @response_json
-async def time_set(request, mpv_control):
+async def time_set(request: Request, mpv_control: MPVControl):
     if "pos" in request.args:
         success = await mpv_control.seek_absolute(float(request.args["pos"][0]))
     elif "percent" in request.args:
@@ -124,7 +125,7 @@ async def time_set(request, mpv_control):
 @bp.get("/playlist")
 @doc.summary("Get the current playlist")
 @response_json
-async def playlist_get(request, mpv_control):
+async def playlist_get(request: Request, mpv_control: MPVControl):
     value = await mpv_control.playlist_get()
     value = list(PLAYLIST_DATA_CACHE.add_data_to_playlist(value))
     for i, v in enumerate(value):
@@ -137,14 +138,14 @@ async def playlist_get(request, mpv_control):
 @bp.post("/playlist/next")
 @doc.summary("Skip to the next item in the playlist")
 @response_json
-async def playlist_next(request, mpv_control):
+async def playlist_next(request: Request, mpv_control: MPVControl):
     success = await mpv_control.playlist_next()
     return locals()
 
 @bp.post("/playlist/previous")
 @doc.summary("Go back to the previous item in the playlist")
 @response_json
-async def playlist_previous(request, mpv_control):
+async def playlist_previous(request: Request, mpv_control: MPVControl):
     success = await mpv_control.playlist_prev()
     return locals()
 
@@ -152,7 +153,7 @@ async def playlist_previous(request, mpv_control):
 @doc.summary("Go chosen item in the playlist")
 @doc.consumes({"index": doc.Integer("The 0 indexed playlist item to go to")}, required=True)
 @response_json
-async def playlist_goto(request, mpv_control):
+async def playlist_goto(request: Request, mpv_control: MPVControl):
     if "index" not in request.args:
         raise APIError("Missing the required parameter: \"index\"")
     success = await mpv_control.playlist_goto(
@@ -163,7 +164,7 @@ async def playlist_goto(request, mpv_control):
 @doc.summary("Clears single item or whole playlist")
 @doc.consumes({"index": doc.Integer("Index to item in playlist to remove. If unset, the whole playlist is cleared")})
 @response_json
-async def playlist_remove_or_clear(request, mpv_control):
+async def playlist_remove_or_clear(request: Request, mpv_control: MPVControl):
     if "index" in request.args:
         success = await mpv_control.playlist_remove(int(request.args["index"][0]))
         action = f"remove #{request.args['index'][0]}"
@@ -183,7 +184,7 @@ async def playlist_remove_or_clear(request, mpv_control):
 @doc.consumes({"index2": int}, required=True)
 @doc.consumes({"index1": int}, required=True)
 @response_json
-async def playlist_move(request, mpv_control):
+async def playlist_move(request: Request, mpv_control: MPVControl):
     if "index1" not in request.args or "index2" not in request.args:
         raise APIError(
             "Missing at least one of the required query "
@@ -196,14 +197,14 @@ async def playlist_move(request, mpv_control):
 @bp.post("/playlist/shuffle")
 @doc.summary("Clears single item or whole playlist")
 @response_json
-async def playlist_shuffle(request, mpv_control):
+async def playlist_shuffle(request: Request, mpv_control: MPVControl):
     success = await mpv_control.playlist_shuffle()
     return locals()
 
 @bp.get("/playlist/loop")
 @doc.summary("See whether it loops the playlist or not")
 @response_json
-async def playlist_get_looping(request, mpv_control):
+async def playlist_get_looping(request: Request, mpv_control: MPVControl):
     value = await mpv_control.playlist_get_looping()
     return locals()
 
@@ -211,7 +212,7 @@ async def playlist_get_looping(request, mpv_control):
 @doc.summary("Sets whether to loop the playlist or not")
 @doc.consumes({"loop": doc.Boolean("Whether to be looping or not")}, required=True)
 @response_json
-async def playlist_set_looping(request, mpv_control):
+async def playlist_set_looping(request: Request, mpv_control: MPVControl):
     if "loop" not in request.args:
         raise APIError("Missing the required parameter: \"loop\"")
     success = await mpv_control.playlist_set_looping(

grzegorz/mpv.py

@@ -2,9 +2,11 @@ import os
 import asyncio
 import json
 from shlex import quote
+from typing import List, Optional, Union
 
 from . import nyasync
 
+
 class MPV:
     _ipc_endpoint = 'mpv_ipc.socket'
 
@@ -69,7 +71,8 @@ class MPV:
             else: # response
                 self.responses.put_nowait(msg)
 
-class MPVError(Exception): pass
+class MPVError(Exception):
+    pass
 
 class MPVControl:
     def __init__(self):
@@ -84,6 +87,7 @@ class MPVControl:
 
     async def process_events(self):
         async for event in self.mpv.events:
+            # TODO: print?
             pass
 
     async def send_request(self, msg):
@@ -96,6 +100,8 @@ class MPVControl:
 
     #other commands:
     async def wake_screen(self):
+        # TODO: use this
+        # TODO: wayland counterpart
         p = await asyncio.create_subprocess_exec(
             "xset",
             "-display",
@@ -107,73 +113,97 @@ class MPVControl:
         code = await process.wait()
 
     #Shorthand command requests:
-    async def loadfile(self, file):#appends to playlist and start playback if paused
+
+    async def loadfile(self, file: Union[str, Path]):
+        "appends to playlist and start playback if paused"
+
+        if isinstance(file, Path):
+            file = str(file)
         resp = await self.send_request({"command":["loadfile", file, "append-play"]})
         return resp["error"] == "success"
+
     async def pause_get(self):
         resp = await self.send_request({"command":["get_property", "pause"]})
         if "error" in resp and resp["error"] != "success":
             raise MPVError("Unable to get whether paused or not: " + resp["error"])
         return resp["data"] if "data" in resp else None
-    async def pause_set(self, state):
+
+    async def pause_set(self, state: bool):
         resp = await self.send_request({"command":["set_property", "pause", bool(state)]})
         return resp["error"] == "success"
+
     async def volume_get(self):
         resp = await self.send_request({"command":["get_property", "volume"]})
         if "error" in resp and resp["error"] != "success":
             raise MPVError("Unable to get volume! " + resp["error"])
         return resp["data"] if "data" in resp else None
-    async def volume_set(self, volume):
+
+    async def volume_set(self, volume: int):
         resp = await self.send_request({"command":["set_property", "volume", volume]})
         return resp["error"] == "success"
+
     async def time_pos_get(self):
         resp = await self.send_request({"command":["get_property", "time-pos"]})
         if "error" in resp and resp["error"] != "success":
             raise MPVError("Unable to get time pos: " + resp["error"])
         return resp["data"] if "data" in resp else None
+
     async def time_remaining_get(self):
         resp = await self.send_request({"command":["get_property", "time-remaining"]})
         if "error" in resp and resp["error"] != "success":
             raise MPVError("Unable to get time left:" + resp["error"])
         return resp["data"] if "data" in resp else None
-    async def seek_absolute(self, seconds):
+
+    async def seek_absolute(self, seconds: float):
         resp = await self.send_request({"command":["seek", seconds, "absolute"]})
         return resp["data"] if "data" in resp else None
-    async def seek_relative(self, seconds):
+
+    async def seek_relative(self, seconds: float):
         resp = await self.send_request({"command":["seek", seconds, "relative"]})
         return resp["data"] if "data" in resp else None
-    async def seek_percent(self, percent):
+
+    async def seek_percent(self, percent: float):
         resp = await self.send_request({"command":["seek", percent, "absolute-percent"]})
         return resp["data"] if "data" in resp else None
+
     async def playlist_get(self):
         resp = await self.send_request({"command":["get_property", "playlist"]})
         if "error" in resp and resp["error"] != "success":
             raise MPVError("Unable to get playlist:" + resp["error"])
         return resp["data"] if "data" in resp else None
+
     async def playlist_next(self):
         resp = await self.send_request({"command":["playlist-next", "weak"]})
         return resp["error"] == "success"
+
     async def playlist_prev(self):
         resp = await self.send_request({"command":["playlist-prev", "weak"]})
         return resp["error"] == "success"
+
     async def playlist_goto(self, index):
         resp = await self.send_request({"command":["set_property", "playlist-pos", index]})
         return resp["error"] == "success"
+
     async def playlist_clear(self):
         resp = await self.send_request({"command":["playlist-clear"]})
         return resp["error"] == "success"
-    async def playlist_remove(self, index=None):
+
+    async def playlist_remove(self, index: Optional[int] = None):
         resp = await self.send_request({"command":["playlist-remove", "current" if index==None else index]})
         return resp["error"] == "success"
-    async def playlist_move(self, index1, index2):
+
+    async def playlist_move(self, index1: int, index2: int):
         resp = await self.send_request({"command":["playlist-move", index1, index2]})
         return resp["error"] == "success"
+
     async def playlist_shuffle(self):
         resp = await self.send_request({"command":["playlist-shuffle"]})
         return resp["error"] == "success"
+
     async def playlist_get_looping(self):
         resp = await self.send_request({"command":["get_property", "loop-playlist"]})
         return resp["data"] == "inf" if "data" in resp else False
-    async def playlist_set_looping(self, value):
+
+    async def playlist_set_looping(self, value: bool):
         resp = await self.send_request({"command":["set_property", "loop-playlist", "inf" if value else "no"]})
         return resp["error"] == "success"

grzegorz/nyasync.py

@@ -1,4 +1,5 @@
 import asyncio
+from asyncio.streams import StreamReader, StreamWriter
 
 def ify(func):
     """Decorate func to run async in default executor"""
@@ -57,9 +58,9 @@ async def unix_connection(path):
     return UnixConnection(*endpoints)
 
 class UnixConnection:
-    def __init__(self, reader, writer):
-        self.reader = reader
-        self.writer = writer
+    def __init__(self, reader: StreamReader, writer: StreamWriter):
+        self.reader: StreamReader = reader
+        self.writer: StreamWriter = writer
 
     def __aiter__(self):
         return self.reader.__aiter__()

grzegorz/playlist_data.py

@@ -1,4 +1,3 @@
-import asyncio
 from .metadatafetch import get_metadata
 from . import nyasync
 
@@ -8,9 +7,11 @@ class PlaylistDataCache:
         self.filepath_data_map = {}
         self.auto_fetch_data = auto_fetch_data
         self.jobs = None
+
     def add_data(self, filepath, data=None):
         if data:
             self.filepath_data_map[filepath] = data
+
     async def run(self):
         if not self.auto_fetch_data: return
 
@@ -22,6 +23,7 @@ class PlaylistDataCache:
             if filename in self.filepath_data_map:
                 self.filepath_data_map[filename].update(data)
                 del self.filepath_data_map[filename]["fetching"]
+
     def add_data_to_playlist(self, playlist):
         seen = set()
 
@@ -45,4 +47,3 @@ class PlaylistDataCache:
         not_seen = set(self.filepath_data_map.keys()) - seen
         for name in not_seen:
             del self.filepath_data_map[name]
-