#!/usr/bin/python3
"""
Play streams on a Chromecast Audio, using PyChromecast.
"""
from pychromecast import get_chromecasts, _get_chromecast_from_host, Chromecast
from pychromecast.error import PyChromecastError
from inspect import signature
import logging
import time
from typing import Optional, Union, List, Tuple, Dict, Any
class Discovery:
def __init__(self) -> None:
self._tries: int = 3
self._retry_wait: int = 5
self._timeout: float = 20.0
self._logger = logging.getLogger(self.__class__.__name__)
def _wait_for_status(self, cc: Chromecast) -> bool:
cc.wait(timeout=self._timeout)
if cc.status is not None:
self._logger.info(f"Connected chromecast '{cc.name}'")
return True
else:
cc.disconnect(blocking=True, timeout=self._timeout)
return False
def find_by_name(self, friendly_name: str,
host: Optional[str] = None) -> Optional[Chromecast]:
self._logger.debug(f"Trying to discover chromecast '{friendly_name}'")
try:
extra_args: Dict[str, Any] = {}
if host is not None and 'known_hosts' in \
signature(get_chromecasts).parameters.keys():
extra_args = {'known_hosts': [host]}
discovered: Union[Tuple, List] = get_chromecasts(
blocking=True,
tries=self._tries,
retry_wait=self._retry_wait,
timeout=self._timeout,
**extra_args,
)
if isinstance(discovered, tuple):
# discovered[1].stop_discovery() # keep running
discovered = discovered[0]
cc: Chromecast
for cc in discovered:
self._logger.debug(f"Discovered chromecast '{cc.name}'")
if cc.name == friendly_name:
return cc if self._wait_for_status(cc) else None
return None
except PyChromecastError as e:
self._logger.warning(str(e))
return None
def find_by_host(self, host: str, port: int) -> Optional[Chromecast]:
self._logger.debug(f"Trying to connect chromecast '{host}:{port}'")
try:
# ip_address, port, uuid, model_name, friendly_name
host_info = (host, port, None, "Unknown Model", host)
cc = _get_chromecast_from_host(host_info,
tries=self._tries,
retry_wait=self._retry_wait,
timeout=self._timeout)
return cc if self._wait_for_status(cc) else None
except PyChromecastError as e:
self._logger.warning(str(e))
return None
def discover(self, name: Optional[str],
host: Optional[str], port: int) -> Optional[Chromecast]:
"""
Tries first to discover the chromecast with the given name via mDNS.
As fallback, a direct connection to the given IP address is tried.
"""
cc: Optional[Chromecast] = None
try:
if name is not None:
cc = self.find_by_name(name, host)
if cc is None and host is not None:
cc = self.find_by_host(host, port)
except KeyboardInterrupt:
logger.info("Shutting down")
return None
return cc
class ChromecastStream:
def __init__(self, cc: Chromecast) -> None:
self._cc: Chromecast = cc
self._timeout: float = 20.0
self._logger = logging.getLogger(self.__class__.__name__)
@property
def chromecast(self) -> Chromecast:
return self._cc
def disconnect(self) -> None:
self._logger.debug("Disconnecting")
self._cc.disconnect(blocking=True, timeout=self._timeout)
def stop(self) -> None:
self._logger.debug("Stopping playback")
self._cc.quit_app()
def set_volume(self, volume: int) -> None:
volume = min(max(volume, 0), 100)
self._logger.debug(f"Setting volume to {volume}")
self._cc.set_volume(volume / 100.0) # unmutes
def play_media(self, url: str, content_type: str,
title: str = "PyChromecast") -> None:
self._logger.debug(f"Playing {url}")
self._cc.play_media(url=url,
content_type=content_type,
title=title,
autoplay=True,
stream_type="LIVE")
def play(self) -> None:
self._logger.debug("Starting playback")
self._cc.media_controller.play()
def status(self) -> str:
return str(self._cc.status) if self._cc.status is not None \
else "Status Unknown"
def __str__(self) -> str:
return str(self._cc)
def _stream(logger: logging.Logger,
cc: ChromecastStream,
url: str, content_type: str, volume: Optional[int]) -> bool:
try:
logger.debug("Starting up")
logger.info(str(cc))
logger.info(cc.status())
cc.stop()
if volume is not None:
cc.set_volume(volume)
try:
cc.play_media(url, content_type)
cc.play() # hack trying to minimize buffering delay
logger.info(f"Started '{url}' - Waiting for interrupt")
logger.debug(cc.status())
while True:
time.sleep(60)
logger.debug(cc.status())
except KeyboardInterrupt:
logger.info("Shutting down")
finally:
cc.stop()
cc.disconnect()
logger.debug(cc.status())
return True
except PyChromecastError as e:
logger.warning(str(e))
return False
def _main(logger: logging.Logger,
name: Optional[str], host: Optional[str], port: int,
url: str, content_type: str, volume: Optional[int]) -> bool:
discovery = Discovery()
cc: Optional[Chromecast] = discovery.discover(name, host, port)
if cc is None:
logger.error("Chromecast not found")
return False
return _stream(logger, ChromecastStream(cc), url, content_type, volume)
def _configure_logging(debug: bool) -> logging.Logger:
import logging.config
import os.path
logging.config.dictConfig({
'version': 1,
'formatters': {
'standard': {
'format': '[%(levelname)s] %(name)s: %(message)s',
},
},
'handlers': {
'default': {
'formatter': 'standard',
'class': 'logging.StreamHandler',
'stream': 'ext://sys.stdout',
},
},
'loggers': {
'': {
'handlers': ['default'],
'level': 'DEBUG' if debug else 'INFO',
'propagate': False,
},
},
})
return logging.getLogger(os.path.splitext(os.path.basename(__file__))[0])
if __name__ == "__main__":
from argparse import ArgumentParser, ArgumentDefaultsHelpFormatter
parser = ArgumentParser(description=__doc__,
formatter_class=ArgumentDefaultsHelpFormatter)
parser.add_argument('--debug', action='store_const',
const=True, default=False,
help='enable debug log output')
parser.add_argument('--name', type=str, default=None,
help='chromecast friendly name for discovery')
parser.add_argument('--host', type=str, default=None,
help='chromecast ip address for direct connection')
parser.add_argument('--port', type=int, default=8009,
help='chromecast port for direct connection')
parser.add_argument('--volume', type=int, default=None,
help='volume to set before streaming, 0-100')
parser.add_argument('--ctype', type=str, default='audio/mpeg',
help='stream content type, if not mp3')
parser.add_argument('URL',
help='stream url to play')
args = parser.parse_args()
if args.name is None and args.host is None:
parser.error('either --name or --host is required (or both)')
logger: logging.Logger = _configure_logging(args.debug)
exit(0 if _main(
logger, args.name, args.host, args.port,
args.URL, args.ctype, args.volume
) else 1)
