Scripting Crestron AirMedia configuration via API

I wrote a Python script to configure Crestron AirMedia devices, via the REST api interface on each device. The intention for this script is to use the command line to configure many dozens of AirMedia devices at once, by typing “python AirMediaAPI.py” (or, use the filename you save the python as).

Github repo

https://github.com/fttdatasolutions/AirMediaAPI_ConfigTool

Code

# Crestron AirMedia API Config 0.9.2

# Required libraries: os, requests, urllib3, pandas (all are installed with native python, but you still need import into script)
import os
import requests
import urllib3
import pandas as pd

# Disable error message for unsecure connection
urllib3.disable_warnings(urllib3.exceptions.InsecureRequestWarning)

# Get username from system and create filepath variables
user = os.environ.get('USER', os.environ.get('USERNAME'))
data_filepath = f"C:/Users/{user}/Downloads/am_api_template.csv"
puf_filepath = f"C:/Users/{user}/Downloads/am3xxx_3.0600.0079.puf"


# Read am_api_template.csv into python as a list of key-value pairs
d = pd.read_csv(data_filepath, header=0)
data = []
for row in d.index:
    i = {}
    for col in d.columns:
        i[col] = d.loc[row,col]
    data.append(i)
del(d)

# Config - loops through each entry in data variable's key-value pairs and configures the parameters listed in "iterator" variable
for am in data:
    target_ip = am["target_ip"]
    index_am = "https://" + target_ip + "/index_airmedia.html"
    index_banner = "https://" + target_ip + "/index_banner.html"
    url_create_user = "https://" + target_ip + "/createUser.html"
    url_login = "https://" + target_ip + "/userlogin.html"
    room_name = am["room_name"]
    am_name = room_name + "-AIRMEDIA"
    intended_dns = am["intended_dns"]
    is_room_conference = am["is_conference"]
    intended_ip = am["vlan_ip"]
    intended_subnet = am["vlan_subnet"]
    intended_gateway = am["vlan_gateway"]

    # iterator is a list of tuples. each tuple consists of (0) a directory entry and (1) a json object with the data for that entry
    iterator = [
    ("/Device/AirMedia/ConnectionDisplayOptions/ShowConnectionInfo", {"Device": {"AirMedia": {"ConnectionDisplayOptions": {"ShowConnectionInfo": False}}}}),
    ("/Device/AirMedia/ConnectionDisplayOptions/ShowConnectionInfo", {"Device": {"AirMedia": {"ConnectionDisplayOptions": {"PersistConnectionInfo": False}}}}),
    ("/Device/AirMedia/ConnectionDisplayOptions/ShowAirplayInfo", {"Device": {"AirMedia": {"ConnectionDisplayOptions": {"ShowAirplayInfo": False}}}}),
    ("/Device/AirMedia/ConnectionDisplayOptions/ShowMiracastInfo", {"Device": {"AirMedia": {"ConnectionDisplayOptions": {"ShowMiracastInfo": False}}}}),
    ("/Device/AutoUpdateMaster/IsEnabled", {"Device": {"AutoUpdateMaster": {"IsEnabled": False}}}),
    ("/Device/App/Config/General/RoomName", {"Device": {"App": {"Config": {"General": {"RoomName": room_name}}}}}),
    ("/Device/App/Config/RuntimeSettings/ShowOverlay", {"Device": {"App": {"Config": {"RuntimeSettings": {"ShowOverlay": False}}}}}),
    ("/Device/App/Config/RuntimeSettings/PinPointUXFontSize", {"Device": {"App": {"Config": {"RuntimeSettings": {"PinPointUXFontSize": "Large"}}}}}),
    ("/Device/NetworkAdapters/HostName", {"Device": {"NetworkAdapters": {"HostName": am_name}}}),
    ("/Device/NetworkAdapters/DnsSettings/IPv4/StaticDns", {"Device": {"NetworkAdapters": {"DnsSettings": {"IPv4": {"StaticDns": [intended_dns]}}}}}),
    ("/Device/SystemClock/TimeZone", {"Device": {"SystemClock": {"TimeZone": "010"}}}),
    ("/Device/IpTableV2", {"Device": {"IpTableV2": {"EntriesCurrentKeyList": ["<example>"], "Entries": {"<example>": {"Address": "<example>", "IpId": "<example>", "Port": 41794, "ProgramInstanceId": ""}}}}}),
    ("/Device/SystemClock/Ntp/Server1/Address", {"Device": {"SystemClock": {"Ntp": {"Servers": {"Server1": {"Address": "<example>"}}}}}}),
    ("/Device/NetworkAdapters/Adapters", {"Device": {"NetworkAdapters": {"Adapters": {"EthernetLan": {"IPv4": {"StaticAddresses": [{"Address": intended_ip, "SubnetMask": [intended_subnet]}], "StaticDefaultGateway": intended_gateway, "IsDhcpEnabled": False}}}}}})
    ]

    # Set first-time admin user, if needed
    try:
        r = requests.Session()
        r.get("https://" + target_ip, verify=False)
        init_track_id = requests.utils.dict_from_cookiejar(r.cookies)
        auth_headers = {"Cookie": init_track_id['TRACKID'],
                        "Host": target_ip,
                        "Referer": "https://" + target_ip + "/"}
        r.get(index_am, verify=False, headers=auth_headers)
        auth_headers['Referer'] = index_am
        r.get(index_banner, verify=False, headers=auth_headers)
        auth_headers['Referer'] = index_banner
        r.get(url_create_user, verify=False, headers=auth_headers)
        auth_headers['Referer'] = url_create_user
        auth_headers['Origin'] = "https://" + target_ip
        r.post("https://" + target_ip + "/Device/Authentication", verify=False, headers=auth_headers, json={"Device": {"Authentication": {"AuthenticationState": {"AdminPassword": "<example>", "AdminUsername": "<example>", "IsEnabled": True}}}})
        r.close()
        print(f"created user for {target_ip}")

    except:
        r.close()
        print(f"{target_ip} already had a user")

    # Settings Config
    s = requests.Session()

    s.get(url_login, verify=False)
    init_track_id = requests.utils.dict_from_cookiejar(s.cookies)
   
    post_headers = {"Cookie": next(iter(init_track_id)) + "=" + init_track_id['TRACKID'],
                    "Origin": target_ip,
                    "Referer": target_ip + "/userlogin.html"}
    post_body = "login=<example>&&passwd=<example>"
    auth = s.post(url_login, verify=False, headers=post_headers, data=post_body)
    print("Configuring " + room_name)

    get_token = {"CREST-XSRF-TOKEN": auth.headers["CREST-XSRF-TOKEN"]} 
    post_token = {"X-CREST-XSRF-TOKEN": auth.headers["CREST-XSRF-TOKEN"]}

    for i in iterator:
        s.post("https://" + target_ip + i[0], json=i[1], verify=False, headers=post_token)
    s.close()

    # Powershell script to handle firmware update and restart. It will trigger a possible firmware upgrade, if the version in your downloads folder is newer than that on your AirMedia. It will not check for completion of the upgrade.
    os.system(f"./AirMediaAPI_Powershell.ps1 -target_ip {target_ip} -filepath {puf_filepath}")

    print("Completed " + room_name)
print("All done. Wait for firmware update, if you think an AirMedia updated firmware (this tool passes the update-puf command, but does not check for completion).")

The python script executes a powershell script that covers firmware updates and device reboot. You can do this through the api, but I found powershell to be effective.

import-module pscrestron -usewindowspowershell

# set target_ip and filepath as mandatory parameters when invoking this script

param(
    [Parameter(Mandatory=$true)]
    [string]$target_ip
)
param(
    [Parameter(Mandatory=$true)]
    [string]$filepath
)

# Update Firmware
update-puf -device $target_ip -path $filepath -secure -username "example" -password "example" -ShowProgress;

# Reboot
invoke-crestroncommand -device $target_ip -secure -username "example" -password "example" -command "reboot";

Future changes

One change that might be implemented for full release is adding user/password fields to the .csv, to allow different access credentials for each room. Our deployments are on a secure network and use standardized credentials (which I removed in this example), but for more modern and secure deployments, one probably wants to give each AirMedia its own credentials.

Python package for interacting with AirMedia API: forthcoming

Scripting Crestron AirMedia configuration via API

Github repo

Code

Future changes

Share this: