Created
June 4, 2022 02:33
-
-
Save mekaneck/e6737923d28f01c4c4083bf3a843dc2b to your computer and use it in GitHub Desktop.
Meeting, Mute, and Camera Status monitor for Mutesync on Windows which will post status to AWS Lambda function (or directly to Home Assistant webhook)
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
""" | |
Script to send meeting, mute, and camera status to AWS Lamda service, which will then relay it to a Home Assistant instance. | |
Mutesync software (www.mutesync.com) must be installed on the host machine. This software will determine meeting and mute status. | |
Camera status will be obtained from a Windows registry entry for Microsoft Teams. | |
""" | |
import time | |
import requests | |
import os | |
import os | |
import sys | |
import win32security | |
import winreg | |
# To obtain the Mutesync status, a token must be obtained from the Mutesync application. | |
# 1. Choose mutesync preferences, authentication tab, and check "allow external app" | |
# 2. Open a browser and navigate to http://127.0.0.1:8249/authenticate | |
# 3. Copy the 16-character token and paste it into this next line between the quotes. | |
mutesyncToken = "1234567890123456" | |
#-------------------------- | |
# Mutesync HTTP GET settings | |
#-------------------------- | |
mutesyncAPIVersion = "1" | |
mutesyncURL = "http://127.0.0.1:8249/state" | |
mutesyncTokenText = "Token " + mutesyncToken | |
get_headers = { | |
'Content-type': 'application/json', | |
'Authorization': mutesyncTokenText, | |
'x-mutesync-api-version': mutesyncAPIVersion, | |
} | |
#-------------------------- | |
# AWS Lambda function HTTP POST settings | |
#-------------------------- | |
aws_lambda_urls = { | |
'alias 1':'https://somelettersandnumbers1.lambda-url.somelocation.on.aws', | |
'alias 2':'https://somelettersandnumbers2.lambda-url.somelocation.on.aws', | |
'alias 3':'https://somelettersandnumbers3.lambda-url.somelocation.on.aws', | |
'alias 4':'https://somelettersandnumbers4.lambda-url.somelocation.on.aws', | |
'alias 5':'https://somelettersandnumbers5.lambda-url.somelocation.on.aws', | |
'alias 6':'https://somelettersandnumbers6.lambda-url.somelocation.on.aws', | |
} | |
post_token = "some_secret_authorization_token" | |
post_headers = { | |
'Content-type': 'application/json', | |
'Authorization': post_token, | |
'Connection': 'close', | |
} | |
#-------------------------- | |
# VPN Software settings | |
#-------------------------- | |
# Proxies to use when on VPN | |
vpn_proxies = { | |
'http': 'proxy.company.com:80', | |
'https': 'proxy.company.com:80', | |
} | |
# Proxies to use when not on VPN | |
nonvpn_proxies = { | |
'http': None, | |
'https': None, | |
} | |
#-------------------------- | |
# Global variables | |
#-------------------------- | |
url_index = 0 | |
#-------------------------- | |
# Other Customizeable settings. These can be changed as desired. | |
#-------------------------- | |
minimumUpdatePeriodInSeconds = 7200 # 2 hours | |
checkPeriodInSeconds = 2 | |
#-------------------------- | |
# Get webcam registry key that can be used to determine if webcam is active | |
#-------------------------- | |
user_name = os.getlogin() #get windows login username | |
sid = win32security.LookupAccountName(None, user_name)[0] #get SID associated with username | |
sidstr = win32security.ConvertSidToStringSid(sid) #convert SID to string | |
key_path = sidstr + r'\SOFTWARE\Microsoft\Windows\CurrentVersion\CapabilityAccessManager\ConsentStore\webcam\NonPackaged' | |
key = winreg.OpenKey(winreg.HKEY_USERS, key_path) | |
camera_registry_keys = dict() | |
#Iterate through each subkey inside the \NonPackaged key, and build a dictionary of the 'LastUsed' timestamps | |
for i in range(0, winreg.QueryInfoKey(key)[0]): #tuple index 0 is the number of subkeys | |
subkey_path = key_path + '\\' + winreg.EnumKey(key, i) | |
subkey = winreg.OpenKey(winreg.HKEY_USERS, subkey_path) | |
for j in range(0, winreg.QueryInfoKey(subkey)[1]): #tuple index 1 is the number of values | |
x = winreg.EnumValue(subkey,j)[0] #tuple 0 is the value name | |
if x.count('LastUsedTimeStop') == 1: | |
camera_registry_keys[subkey] = j | |
break | |
if camera_registry_keys == dict(): | |
SystemExit('Error: No webcam status found') | |
#=========================================================== | |
# Start of Functions | |
#=========================================================== | |
def get_status(sess): | |
data = dict() | |
#Get meeting and mute systus from mutesync | |
try: | |
r = sess.get(mutesyncURL, headers=get_headers, timeout=0.5) | |
r.raise_for_status() | |
mutesync_reply = r.json() | |
except requests.exceptions.HTTPError as err: | |
print(post_url, 'Mutesync HTTP Error : ', err) | |
return False | |
except requests.exceptions.RequestException as err: | |
print(post_url, 'Mutesync Exception: ', err) | |
return False | |
if 'data' in mutesync_reply: | |
if 'muted' in mutesync_reply['data']: | |
data['mute_status'] = 'off' if not mutesync_reply['data']['muted'] else 'on' | |
if 'in_meeting' in mutesync_reply['data']: | |
data['meeting_status'] = 'off' if not mutesync_reply['data']['in_meeting'] else 'on' | |
else: | |
print('Mutesync bad reply:') | |
SystemExit(mutesync_reply) | |
#Get camera status from registry keys | |
data['camera_status'] = 'off' | |
for key, key_index in camera_registry_keys.items(): | |
cam_last_used_timestamp = winreg.EnumValue(key, key_index)[1] | |
if cam_last_used_timestamp == 0: | |
data['camera_status'] = 'on' | |
break | |
return data | |
def get_vpn_status(): | |
#Figure out some way to determine if you are connected to VPN. | |
#Option 1: Use command line with 3rd party VPN software to query status | |
#Option 2: Send GET request to http://icanhazip.com/ to see what your external IP address is | |
vpn_status = False #used if VPN is not active | |
return vpn_status | |
def post_to_ha(proxies, payload): | |
global url_index | |
post_url = list(aws_lambda_urls.values())[url_index] | |
url_index +=1 | |
if url_index >= len(aws_lambda_urls): | |
url_index = 0 | |
try: | |
r = requests.post(post_url, headers=post_headers, proxies=proxies, json=payload, timeout=8.0) | |
r.raise_for_status() # response code validation | |
print('AWS reply: ', r.text) | |
return True | |
except requests.exceptions.HTTPError as err: | |
print(post_url, 'AWS HTTP Error : ', err) | |
return False | |
except requests.exceptions.RequestException as err: | |
print(post_url, 'AWS Exception: ', err) | |
return False | |
def run(): | |
data_previous = dict() | |
last_post_time = 0 | |
sess = requests.Session() | |
while True: | |
data = get_status(sess) | |
if not data: | |
#wait for 1 second and try again | |
time.sleep(1) | |
else: | |
# If status has changed, or minimum update time has expired | |
if (data_previous != data or | |
time.perf_counter() - last_post_time > minimumUpdatePeriodInSeconds): | |
# Check if on VPN | |
if get_vpn_status(): | |
proxies = vpn_proxies | |
else: | |
proxies = nonvpn_proxies | |
# Post an update | |
if post_to_ha(proxies, data): | |
#If update is successful reset timer & sleep | |
data_previous = data | |
last_post_time = time.perf_counter() | |
time.sleep(checkPeriodInSeconds) | |
else: | |
# sleep if no update occurred | |
time.sleep(checkPeriodInSeconds) | |
if __name__ == '__main__': | |
run() |
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment