Meeting, Mute, and Camera Status monitor for Mutesync on Windows which will post status to AWS Lambda function (or directly to Home Assistant webhook)

pubWS.py

"""
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()