1import base64 

2import io 

3import logging 

4import mimetypes 

5import multiprocessing 

6import os 

7import socket 

8from datetime import timedelta 

9from typing import Any, Callable, cast, Dict, Optional, Set, Tuple, Type, Union 

10from urllib.parse import urlparse 

11from uuid import UUID 

12 

13from gi.repository import GLib 

14 

15from .base import Player, PlayerDeviceEvent, PlayerEvent 

16from ..adapters import AdapterManager 

17from ..adapters.api_objects import Song 

18 

19try: 

20 import pychromecast 

21 

22 chromecast_imported = True 

23except Exception: 

24 chromecast_imported = False 

25 

26try: 

27 import bottle 

28 

29 bottle_imported = True 

30except Exception: 

31 bottle_imported = False 

32 

33SERVE_FILES_KEY = "Serve Local Files to Chromecasts on the LAN" 

34LAN_PORT_KEY = "LAN Server Port Number" 

35 

36 

37class ChromecastPlayer(Player): 

38 name = "Chromecast" 

39 can_start_playing_with_no_latency = False 

40 

41 @property 

42 def enabled(self) -> bool: 

43 return chromecast_imported 

44 

45 @staticmethod 

46 def get_configuration_options() -> Dict[str, Union[Type, Tuple[str, ...]]]: 

47 if not bottle_imported: 

48 return {} 

49 return {SERVE_FILES_KEY: bool, LAN_PORT_KEY: int} 

50 

51 @property 

52 def supported_schemes(self) -> Set[str]: 

53 schemes = {"http", "https"} 

54 if bottle_imported and self.config.get(SERVE_FILES_KEY): 

55 schemes.add("file") 

56 return schemes 

57 

58 _timepos = 0.0 

59 

60 def __init__( 

61 self, 

62 on_timepos_change: Callable[[Optional[float]], None], 

63 on_track_end: Callable[[], None], 

64 on_player_event: Callable[[PlayerEvent], None], 

65 player_device_change_callback: Callable[[PlayerDeviceEvent], None], 

66 config: Dict[str, Union[str, int, bool]], 

67 ): 

68 self.server_process: Optional[multiprocessing.Process] = None 

69 self.on_timepos_change = on_timepos_change 

70 self.on_track_end = on_track_end 

71 self.on_player_event = on_player_event 

72 self.player_device_change_callback = player_device_change_callback 

73 

74 self.change_settings(config) 

75 

76 if chromecast_imported: 

77 self._chromecasts: Dict[UUID, pychromecast.Chromecast] = {} 

78 self._current_chromecast: Optional[pychromecast.Chromecast] = None 

79 

80 self.stop_get_chromecasts = None 

81 self.refresh_players() 

82 

83 def chromecast_discovered_callback(self, chromecast: Any): 

84 chromecast = cast(pychromecast.Chromecast, chromecast) 

85 self._chromecasts[chromecast.device.uuid] = chromecast 

86 self.player_device_change_callback( 

87 PlayerDeviceEvent( 

88 PlayerDeviceEvent.Delta.ADD, 

89 type(self), 

90 str(chromecast.device.uuid), 

91 chromecast.device.friendly_name, 

92 ) 

93 ) 

94 

95 def change_settings(self, config: Dict[str, Union[str, int, bool]]): 

96 if not chromecast_imported: 

97 return 

98 

99 self.config = config 

100 if bottle_imported and self.config.get(SERVE_FILES_KEY): 

101 # Try and terminate the existing process if it exists. 

102 if self.server_process is not None: 

103 try: 

104 self.server_process.terminate() 

105 except Exception: 

106 pass 

107 

108 self.server_process = multiprocessing.Process( 

109 target=self._run_server_process, 

110 args=("0.0.0.0", self.config.get(LAN_PORT_KEY)), 

111 ) 

112 self.server_process.start() 

113 

114 def refresh_players(self): 

115 if not chromecast_imported: 

116 return 

117 

118 if self.stop_get_chromecasts is not None: 

119 self.stop_get_chromecasts.cancel() 

120 

121 for id_, chromecast in self._chromecasts.items(): 

122 self.player_device_change_callback( 

123 PlayerDeviceEvent( 

124 PlayerDeviceEvent.Delta.REMOVE, 

125 type(self), 

126 str(id_), 

127 chromecast.device.friendly_name, 

128 ) 

129 ) 

130 

131 self._chromecasts = {} 

132 

133 self.stop_get_chromecasts = pychromecast.get_chromecasts( 

134 blocking=False, callback=self.chromecast_discovered_callback 

135 ) 

136 

137 def set_current_device_id(self, device_id: str): 

138 self._current_chromecast = self._chromecasts[UUID(device_id)] 

139 self._current_chromecast.media_controller.register_status_listener(self) 

140 self._current_chromecast.register_status_listener(self) 

141 self._current_chromecast.wait() 

142 

143 def new_cast_status(self, status: Any): 

144 assert self._current_chromecast 

145 self.on_player_event( 

146 PlayerEvent( 

147 PlayerEvent.EventType.VOLUME_CHANGE, 

148 str(self._current_chromecast.device.uuid), 

149 volume=(status.volume_level * 100 if not status.volume_muted else 0), 

150 ) 

151 ) 

152 

153 # This happens when the Chromecast is first activated or after "Stop Casting" is 

154 # pressed in the Google Home app. Reset `song_loaded` so that it calls 

155 # `play_media` the next time around rather than trying to toggle the play state. 

156 if status.session_id is None: 

157 self.song_loaded = False 

158 

159 time_increment_order_token = 0 

160 

161 def new_media_status(self, status: Any): 

162 # Detect the end of a track and go to the next one. 

163 if ( 

164 status.idle_reason == "FINISHED" 

165 and status.player_state == "IDLE" 

166 and self._timepos > 0 

167 ): 

168 logging.debug("Chromecast track ended") 

169 self.on_track_end() 

170 return 

171 

172 self.song_loaded = True 

173 

174 self._timepos = status.current_time 

175 

176 assert self._current_chromecast 

177 self.on_player_event( 

178 PlayerEvent( 

179 PlayerEvent.EventType.PLAY_STATE_CHANGE, 

180 str(self._current_chromecast.device.uuid), 

181 playing=(status.player_state in ("PLAYING", "BUFFERING")), 

182 ) 

183 ) 

184 

185 def increment_time(order_token: int): 

186 if self.time_increment_order_token != order_token or not self.playing: 

187 return 

188 

189 self._timepos += 0.5 

190 self.on_timepos_change(self._timepos) 

191 GLib.timeout_add(500, increment_time, order_token) 

192 

193 self.time_increment_order_token += 1 

194 GLib.timeout_add(500, increment_time, self.time_increment_order_token) 

195 

196 def shutdown(self): 

197 if self.server_process: 

198 self.server_process.terminate() 

199 

200 try: 

201 self._current_chromecast.quit_app() 

202 except Exception: 

203 pass 

204 

205 _serving_song_id = multiprocessing.Array("c", 1024) # huge buffer, just in case 

206 _serving_token = multiprocessing.Array("c", 16) 

207 

208 def _run_server_process(self, host: str, port: int): 

209 app = bottle.Bottle() 

210 

211 @app.route("/") 

212 def index() -> str: 

213 return """ 

214 <h1>Sublime Music Local Music Server</h1> 

215 <p> 

216 Sublime Music uses this port as a server for serving music to 

217 Chromecasts on the same LAN. 

218 </p> 

219 """ 

220 

221 @app.route("/s/<token>") 

222 def stream_song(token: str) -> bytes: 

223 if token != self._serving_token.value.decode(): 

224 raise bottle.HTTPError(status=401, body="Invalid token.") 

225 

226 song = AdapterManager.get_song_details( 

227 self._serving_song_id.value.decode() 

228 ).result() 

229 filename = AdapterManager.get_song_file_uri(song) 

230 with open(filename[7:], "rb") as fin: 

231 song_buffer = io.BytesIO(fin.read()) 

232 

233 content_type = mimetypes.guess_type(filename)[0] 

234 bottle.response.set_header("Content-Type", content_type) 

235 bottle.response.set_header("Accept-Ranges", "bytes") 

236 return song_buffer.read() 

237 

238 bottle.run(app, host=host, port=port) 

239 

240 @property 

241 def playing(self) -> bool: 

242 if ( 

243 not self._current_chromecast 

244 or not self._current_chromecast.media_controller 

245 ): 

246 return False 

247 return self._current_chromecast.media_controller.status.player_is_playing 

248 

249 def get_volume(self) -> float: 

250 if self._current_chromecast: 

251 # The volume is in the range [0, 1]. Multiply by 100 to get to [0, 100]. 

252 return self._current_chromecast.status.volume_level * 100 

253 else: 

254 return 100 

255 

256 def set_volume(self, volume: float): 

257 if self._current_chromecast: 

258 # volume value is in [0, 100]. Convert to [0, 1] for Chromecast. 

259 self._current_chromecast.set_volume(volume / 100) 

260 

261 def get_is_muted(self) -> bool: 

262 if not self._current_chromecast: 

263 return False 

264 return self._current_chromecast.volume_muted 

265 

266 def set_muted(self, muted: bool): 

267 if not self._current_chromecast: 

268 return 

269 self._current_chromecast.set_volume_muted(muted) 

270 

271 def play_media(self, uri: str, progress: timedelta, song: Song): 

272 assert self._current_chromecast 

273 scheme = urlparse(uri).scheme 

274 if scheme == "file": 

275 token = base64.b16encode(os.urandom(8)) 

276 self._serving_token.value = token 

277 self._serving_song_id.value = song.id.encode() 

278 

279 # If this fails, then we are basically screwed, so don't care if it blows 

280 # up. 

281 # TODO (#129): this does not work properly when on VPNs when the DNS is 

282 # piped over the VPN tunnel. 

283 s = socket.socket(socket.AF_INET, socket.SOCK_DGRAM) 

284 s.connect(("8.8.8.8", 80)) 

285 host_ip = s.getsockname()[0] 

286 s.close() 

287 

288 uri = f"http://{host_ip}:{self.config.get(LAN_PORT_KEY)}/s/{token.decode()}" 

289 logging.info("Serving {song.name} at {uri}") 

290 

291 assert AdapterManager._instance 

292 networked_scheme_priority = ("https", "http") 

293 scheme = sorted( 

294 AdapterManager._instance.ground_truth_adapter.supported_schemes, 

295 key=lambda s: networked_scheme_priority.index(s), 

296 )[0] 

297 cover_art_url = AdapterManager.get_cover_art_uri( 

298 song.cover_art, scheme, size=1000 

299 ).result() 

300 self._current_chromecast.media_controller.play_media( 

301 uri, 

302 # Just pretend that whatever we send it is mp3, even if it isn't. 

303 "audio/mp3", 

304 current_time=progress.total_seconds(), 

305 title=song.title, 

306 thumb=cover_art_url, 

307 metadata={ 

308 "metadataType": 3, 

309 "albumName": song.album.name if song.album else None, 

310 "artist": song.artist.name if song.artist else None, 

311 "trackNumber": song.track, 

312 }, 

313 ) 

314 

315 # Make sure to clear out the cache duration state. 

316 self.on_player_event( 

317 PlayerEvent( 

318 PlayerEvent.EventType.STREAM_CACHE_PROGRESS_CHANGE, 

319 str(self._current_chromecast.device.uuid), 

320 stream_cache_duration=0, 

321 ) 

322 ) 

323 self._timepos = progress.total_seconds() 

324 

325 def pause(self): 

326 if self._current_chromecast and self._current_chromecast.media_controller: 

327 self._current_chromecast.media_controller.pause() 

328 

329 def play(self): 

330 if self._current_chromecast and self._current_chromecast.media_controller: 

331 self._current_chromecast.media_controller.play() 

332 

333 def seek(self, position: timedelta): 

334 if not self._current_chromecast: 

335 return 

336 

337 do_pause = not self.playing 

338 self._current_chromecast.media_controller.seek(position.total_seconds()) 

339 if do_pause: 

340 self.pause() 

341 

342 def _wait_for_playing(self): 

343 pass