1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
|
#!/usr/bin/env python3
import subprocess
import time
from pathlib import Path
import threading
import json
import tempfile
import urllib.request
import websocket
# 128kbps Opus
URL="https://radio.animebits.moe/stream/stream128.ogg"
# 256kbps Opus
#URL="https://radio.animebits.moe/stream/stream256.ogg"
# ~148kbps VBR AAC
#URL="https://radio.animebits.moe/stream/stream128.aac"
# ~192kbps VBR MP3
#URL="https://radio.animebits.moe/stream/stream192.mp3"
# Lossless FLAC
#URL="https://radio.animebits.moe/stream/stream.flac"
#config file & pid
CONF = Path.home() / ".config" / "radiorc"
PID = Path.home() / ".config" / "radio-pid"
with open(CONF) as f:
VOL = f.read().strip()
def notify(title, image_url=None):
icon_path = None
if image_url:
try:
tmp = tempfile.NamedTemporaryFile(delete=False, suffix=".png")
urllib.request.urlretrieve(image_url, tmp.name)
icon_path = tmp.name
except:
icon_path = None
args = ["notify-send", title]
if icon_path:
args += ["-i", icon_path]
subprocess.run(args)
def ws_listener():
def on_message(ws, message):
data = json.loads(message)
if "data" in data and "title" in data["data"]:
title = data["data"]["title"]
image = data["data"].get("album_cover")
notify(title, image)
def on_error(ws, error):
pass
def on_close(ws, close_status_code, close_msg):
time.sleep(1)
run_ws()
def on_open(ws):
pass
def run_ws():
ws = websocket.WebSocketApp("wss://radio.animebits.moe/api/events/basic",
on_message=on_message,
on_error=on_error,
on_close=on_close,
on_open=on_open)
ws.run_forever()
run_ws()
threading.Thread(target=ws_listener, daemon=True).start()
while True:
try:
print("Playing now...")
subprocess.run(["notify-send", "Playing now..."])
proc = subprocess.Popen(["mpv", "--no-video", f"--volume={VOL}", URL])
with open(PID, "w") as f:
f.write(str(proc.pid))
proc.wait()
except KeyboardInterrupt:
subprocess.run(["notify-send", "Stopping now..."])
print("\nStopping now...")
break
except Exception as e:
print(f"Something broke... trying again... Error: {e}")
time.sleep(2)
|
