Python realtime Chromecast application detect script
Posted: Sunday 14 January 2018 23:10
Hello,
Don't know if anybody wrote something like this already but I like to share a python script I wrote for detecting application changes for Chromecast devices and update a text device in Domoticz to match the application connected in realtime.
My previous script was a python script running avahi-browse requesting information from all network connected Chromecast devices and parse the information to grab the connected application (Netflix, Spotify etc.). Had this scheduled to run every X minutes in a cron schedule. Having a Domoticz light scene to switch some lights when watching Netflix I got a bit annoyed with the delay. I’m watching Netflix, and depending on the check delay the lights will change while I was watching already.
This got me thinking, since Chromecast is using Multicast DNS (mDNS) and the avahi-browse command I was using is seen by any device connected to your network, including the answer from the Chromecast device itself. Why not monitor the network and trigger on the network event? Now I’ve seen other posts from people using pychromecast module for this, don’t know if they trigger as fast as this or are using a cron schedule as well.
To get this running you need a few python modules extra from the default installed. You need:
I’ve checked and they should all be available on Raspberry PI and any other Debian like systems. Install those, copy the script change some settings to match your configuration and Chromecast devices. I put my script in /usr/local/bin and made it executable. If you do that you can use the systemctl service config file listed below and put it in /etc/systemd/system/ as chromecast_detect.service and have that startup on boot.
And finally the chromecast_detect.py script itself below. Curious what you think. Just to note, there is an odd count variable in the script. Each detection for a Chromecast packet gave me 2 matches. So a quick fix was just if count 1, do stuff and if count 2 ignore. I don’t consider myself to be a real programmer, but I get the job done 
Wrote the script this weekend, it’s running almost 24 hours now. Bugs? probably, it's the first version. I have it configured to log to a file in /var/log/chromecast.log. It should log python errors as well. To give you an idea on the logging:
chromecast_detect.py:
Don't know if anybody wrote something like this already but I like to share a python script I wrote for detecting application changes for Chromecast devices and update a text device in Domoticz to match the application connected in realtime.
My previous script was a python script running avahi-browse requesting information from all network connected Chromecast devices and parse the information to grab the connected application (Netflix, Spotify etc.). Had this scheduled to run every X minutes in a cron schedule. Having a Domoticz light scene to switch some lights when watching Netflix I got a bit annoyed with the delay. I’m watching Netflix, and depending on the check delay the lights will change while I was watching already.
This got me thinking, since Chromecast is using Multicast DNS (mDNS) and the avahi-browse command I was using is seen by any device connected to your network, including the answer from the Chromecast device itself. Why not monitor the network and trigger on the network event? Now I’ve seen other posts from people using pychromecast module for this, don’t know if they trigger as fast as this or are using a cron schedule as well.
To get this running you need a few python modules extra from the default installed. You need:
Code: Select all
python-pcapy, python-dpkt, python-requests and python-jsonCode: Select all
[Unit]
Description=Chromecast Domoticz detect Service
After=multi-user.target
[Service]
User=root
Type=idle
ExecStart=/usr/local/bin/chromecast_detect.py
[Install]
WantedBy=multi-user.targetWrote the script this weekend, it’s running almost 24 hours now. Bugs? probably, it's the first version. I have it configured to log to a file in /var/log/chromecast.log. It should log python errors as well. To give you an idea on the logging:
Code: Select all
01/14/2018 09:34:13 PM Starting Chromecast Domoticz detect
01/14/2018 09:34:21 PM Chromecast Slaapkamer now casting: Netflix
01/14/2018 09:34:21 PM Starting new HTTP connection (1): 127.0.0.1
01/14/2018 09:34:21 PM http://127.0.0.1:8080 "GET /json.htm?type=devices&rid=185 HTTP/1.1" 200 489
01/14/2018 09:34:21 PM Updating domoticz information for chromecast: Slaapkamer
01/14/2018 09:34:21 PM Starting new HTTP connection (1): 127.0.0.1
01/14/2018 09:34:21 PM http://127.0.0.1:8080 "GET /json.htm?type=command¶m=udevice&idx=185&nvalue=0&svalue=Netflix HTTP/1.1" 200 53
01/14/2018 09:36:06 PM Chromecast Slaapkamer now casting: Nothing
01/14/2018 09:36:06 PM Starting new HTTP connection (1): 127.0.0.1
01/14/2018 09:36:06 PM http://127.0.0.1:8080 "GET /json.htm?type=devices&rid=185 HTTP/1.1" 200 488
01/14/2018 09:36:06 PM Updating domoticz information for chromecast: Slaapkamer
01/14/2018 09:36:06 PM Starting new HTTP connection (1): 127.0.0.1
01/14/2018 09:36:06 PM http://127.0.0.1:8080 "GET /json.htm?type=command¶m=udevice&idx=185&nvalue=0&svalue=Nothing HTTP/1.1" 200 53
Code: Select all
#!/usr/bin/env python
# File : chromecast_detect.py
# Author: Triton
# Date: 14-Jan-2018
# Description : Monitor network for Multicast DNS _googlecast._tcp.local, update devices with application playing
# information Domoticz accordingly.
# Version : 1.0
#
# Required in Domoticz: text devices for each chromecast, set IDX below.
#
# Configuration (Change here):
DOMOTICZ_HOST = 'http://127.0.0.1:8080'
DOMOTICZ_USER = ''
DOMOTICZ_PASS = ''
URL_GET = '/json.htm?type=devices&rid=PARAM_IDX'
URL_POST = '/json.htm?type=command¶m=udevice&idx=PARAM_IDX&nvalue=0&svalue=PARAM_STR'
logOutFilename = '/var/log/chromecast.log' # Disable for console output
# Change Chromecast name and matching IDX for Domoticz:
CHROMECAST_IDX = {
'<YOUR_CHROMECAST_1>': 1,
'<YOUR_CHROMECAST_2>': 2,
}
# Nothing to edit under this line ##############################################################
import pcapy
import dpkt
import logging
import sys
import requests
import json
class StreamToLogger(object):
"""
Fake file-like stream object that redirects writes to a logger instance.
"""
def __init__(self, logger, log_level=logging.INFO):
self.logger = logger
self.log_level = log_level
self.linebuf = ''
def write(self, buf):
for line in buf.rstrip().splitlines():
self.logger.log(self.log_level, line.rstrip())
class StdLogger:
def __init__(self,
filename,
level=logging.DEBUG,
formattting='%(asctime)s %(message)s',
datefmt='%m/%d/%Y %I:%M:%S %p',
filemode='a'
):
logging.basicConfig(
level=level,
format=formattting,
filename=filename,
datefmt=datefmt,
filemode=filemode,
)
stdout_logger = logging.getLogger('STDOUT')
sl = StreamToLogger(stdout_logger, logging.INFO)
sys.stdout = sl
stderr_logger = logging.getLogger('STDERR')
sl = StreamToLogger(stderr_logger, logging.ERROR)
sys.stderr = sl
class NetworkMonitor:
def __init__(self):
self.count = 1
pass
def start(self):
dev = pcapy.findalldevs()[0]
p = pcapy.open_live(dev, 262144, False, 1)
p.setfilter('port 5353')
p.loop(-1, self.handle_packet)
def handle_packet(self, header, data):
eth = dpkt.ethernet.Ethernet(data)
if eth.type == dpkt.ethernet.ETH_TYPE_IP:
ip = eth.data
ip_data = ip.data
if isinstance(ip_data, dpkt.udp.UDP):
udp = ip_data
if udp.dport == 5353:
mdns = dpkt.dns.DNS(udp.data)
if mdns.an:
for answer in mdns.an:
if answer.type == 16 and '_googlecast._tcp.local' in answer.name:
# Getting the same packet twice, need to filter that with a count
if self.count == 1:
# print("PTR request", answer.name, "response", answer)
result = answer.text
chromecast_name = result[6].split('=')[1]
status = result[11].split('=')[1] or 'Nothing'
print('Chromecast %s now casting: %s' % (chromecast_name, status))
url_get = DOMOTICZ_HOST + URL_GET
url_get = url_get.replace('PARAM_IDX', str(CHROMECAST_IDX[chromecast_name]))
data = requests.get(url_get, auth=(DOMOTICZ_USER, DOMOTICZ_PASS))
old_status = json.loads(data.text)['result'][0]['Data']
if old_status != status:
print('Updating domoticz information for chromecast: %s' % chromecast_name)
url_post = DOMOTICZ_HOST + URL_POST
url_post = url_post.replace('PARAM_IDX', str(CHROMECAST_IDX[chromecast_name]))
url_post = url_post.replace('PARAM_STR', str(status))
requests.get(url_post, auth=(DOMOTICZ_USER, DOMOTICZ_PASS))
else:
print('Domoticz matching chromecast, no need to update')
self.count = 2
else:
self.count = 1
def main():
try:
if 'logOutFilename' in globals():
StdLogger(logOutFilename)
print('Starting Chromecast Domoticz detect')
NetworkMonitor().start()
except KeyboardInterrupt:
sys.exit(0)
if __name__ == "__main__":
main()
.
(dimm the lights when streaming)