mirror of
https://github.com/LukeSpad/BeoGateway.git
synced 2024-12-23 21:51:51 +00:00
Delete MLCONFIG.py
This commit is contained in:
LukeSpad
GitHub
parent
477681868b
commit
6870ada99c
1 changed files with 0 additions and 202 deletions
|
|
@ -1,202 +0,0 @@
|
||||||
import logging
|
|
||||||
import asyncore
|
|
||||||
import json
|
|
||||||
import time
|
|
||||||
import requests
|
|
||||||
from requests.auth import HTTPDigestAuth, HTTPBasicAuth
|
|
||||||
from collections import OrderedDict
|
|
||||||
|
|
||||||
import Resources.CONSTANTS as CONST
|
|
||||||
|
|
||||||
|
|
||||||
class MLConfig:
|
|
||||||
|
|
||||||
def __init__(self, host_address='blgw.local', user='admin', pwd='admin'):
|
|
||||||
self.log = logging.getLogger('Config')
|
|
||||||
self.log.setLevel('INFO')
|
|
||||||
|
|
||||||
self._host = host_address
|
|
||||||
self._user = user
|
|
||||||
self._pwd = pwd
|
|
||||||
|
|
||||||
self._download_data()
|
|
||||||
|
|
||||||
def _download_data(self):
|
|
||||||
try:
|
|
||||||
self.log.info('Downloading configuration data from Gateway...')
|
|
||||||
url = 'http://' + self._host + '/mlgwpservices.json'
|
|
||||||
# try Basic Auth next (this is needed for the BLGW)
|
|
||||||
response = requests.get(url, auth=HTTPBasicAuth(self._user, self._pwd))
|
|
||||||
|
|
||||||
if response.status_code == 401:
|
|
||||||
# try Digest Auth first (this is needed for the MLGW)
|
|
||||||
response = requests.get(url, auth=HTTPDigestAuth(self._user, self._pwd))
|
|
||||||
|
|
||||||
if response.status_code == 401:
|
|
||||||
return
|
|
||||||
else:
|
|
||||||
# Once logged in successfully download and process the configuration data
|
|
||||||
configurationdata = json.loads(response.text)
|
|
||||||
self.log.debug(json.dumps(configurationdata, indent=4))
|
|
||||||
self.configure_mlgw(configurationdata)
|
|
||||||
except ValueError:
|
|
||||||
pass
|
|
||||||
|
|
||||||
def configure_mlgw(self, data):
|
|
||||||
self.log.info('Processing Gateway configuration data...\n')
|
|
||||||
CONST.gateway['Serial_Number'] = data['sn']
|
|
||||||
CONST.gateway['Project'] = data['project']
|
|
||||||
try:
|
|
||||||
CONST.gateway['Installer'] = str(data['installer']['name'])
|
|
||||||
CONST.gateway['Contact'] = str(data['installer']['contact'])
|
|
||||||
gateway_type = 'blgw'
|
|
||||||
except KeyError:
|
|
||||||
gateway_type = 'mlgw'
|
|
||||||
|
|
||||||
for zone in data["zones"]:
|
|
||||||
if int(zone['number']) == 240:
|
|
||||||
continue
|
|
||||||
room = OrderedDict()
|
|
||||||
room['Room_Number'] = zone['number']
|
|
||||||
if gateway_type == 'blgw':
|
|
||||||
# BLGW arranges rooms within zones
|
|
||||||
room['Zone'] = str(zone['name']).split('/')[0]
|
|
||||||
room['Room_Name'] = str(zone['name']).split('/')[1]
|
|
||||||
elif gateway_type == 'mlgw':
|
|
||||||
# MLGW has no zoning concept - devices are arranged in rooms only
|
|
||||||
room['Room_Name'] = str(zone['name'])
|
|
||||||
|
|
||||||
room['Products'] = []
|
|
||||||
|
|
||||||
for product in zone["products"]:
|
|
||||||
device = OrderedDict()
|
|
||||||
room['Products'].append(str(product["name"]))
|
|
||||||
# Device identification
|
|
||||||
device['Device'] = str(product["name"])
|
|
||||||
device['MLN'] = product["MLN"]
|
|
||||||
device['ML_ID'] = ''
|
|
||||||
try:
|
|
||||||
device['Serial_num'] = str(product["sn"])
|
|
||||||
except KeyError:
|
|
||||||
device['Serial_num'] = ''
|
|
||||||
|
|
||||||
# Physical location
|
|
||||||
if gateway_type == 'blgw':
|
|
||||||
# BLGW arranges rooms within zones
|
|
||||||
device['Zone'] = str(zone['name']).split('/')[0]
|
|
||||||
device['Room'] = str(zone['name']).split('/')[1]
|
|
||||||
elif gateway_type == 'mlgw':
|
|
||||||
# MLGW has no zoning concept - devices are arranged in rooms only
|
|
||||||
device['Room'] = str(zone['name'])
|
|
||||||
device['Room_Number'] = str(zone["number"])
|
|
||||||
# Source logging parameters for managing notifications
|
|
||||||
device['Sources'] = OrderedDict()
|
|
||||||
device['Current Source'] = 'None'
|
|
||||||
device['Current Source Type'] = 'None'
|
|
||||||
device['Now Playing'] = 'None'
|
|
||||||
device['Channel/Track'] = '0'
|
|
||||||
device['State'] = 'Standby'
|
|
||||||
device['last update'] = time.time()
|
|
||||||
|
|
||||||
for source in product["sources"]:
|
|
||||||
device['Sources'][str(source["name"])] = OrderedDict()
|
|
||||||
for selectCmd in source["selectCmds"]:
|
|
||||||
if gateway_type == 'blgw':
|
|
||||||
# get source information from the BLGW config file
|
|
||||||
if str(source['sourceId']) == '':
|
|
||||||
source_id = self._srcdictsanitize(CONST.beo4_commanddict, source['selectID']).upper()
|
|
||||||
else:
|
|
||||||
source_id = str(source['sourceId'].split(':')[0])
|
|
||||||
source_id = self._srcdictsanitize(CONST.blgw_srcdict, source_id).upper()
|
|
||||||
device['Sources'][str(source["name"])]['source'] = source_id
|
|
||||||
device['Sources'][str(source["name"])]['uniqueID'] = str(source['sourceId'])
|
|
||||||
else:
|
|
||||||
# MLGW config file is structured differently
|
|
||||||
source_id = self._srcdictsanitize(CONST.beo4_commanddict, source['selectID']).upper()
|
|
||||||
device['Sources'][str(source["name"])]['source'] = source_id
|
|
||||||
source_tuple = (str(source["name"]), source_id)
|
|
||||||
cmd_tuple = (source_id, (int(selectCmd["cmd"]), int(selectCmd["unit"])))
|
|
||||||
device['Sources'][str(source["name"])]['BR1_cmd'] = cmd_tuple
|
|
||||||
if 'channels' in source:
|
|
||||||
device['Sources'][str(source["name"])]['channels'] = []
|
|
||||||
for channel in source['channels']:
|
|
||||||
c = OrderedDict()
|
|
||||||
c_num = ''
|
|
||||||
if gateway_type == 'blgw':
|
|
||||||
num = channel['selectSEQ'][::2]
|
|
||||||
else:
|
|
||||||
num = channel['selectSEQ'][:-1]
|
|
||||||
for n in num:
|
|
||||||
c_num += str(n)
|
|
||||||
c['number'] = int(c_num)
|
|
||||||
c['name'] = channel['name']
|
|
||||||
c['icon'] = channel['icon']
|
|
||||||
device['Sources'][str(source["name"])]['channels'].append(c)
|
|
||||||
|
|
||||||
if source_tuple not in CONST.available_sources:
|
|
||||||
CONST.available_sources.append(source_tuple)
|
|
||||||
|
|
||||||
CONST.devices.append(device)
|
|
||||||
CONST.rooms.append(room)
|
|
||||||
|
|
||||||
self.log.info('Found ' + str(len(CONST.devices)) + ' AV Renderers!')
|
|
||||||
for i in range(len(CONST.devices)):
|
|
||||||
self.log.info('\tMLN ' + str(CONST.devices[i].get('MLN')) + ': ' + str(CONST.devices[i].get('Device')))
|
|
||||||
self.log.info('\tFound ' + str(len(CONST.available_sources)) + ' Available Sources [Name, Type]:')
|
|
||||||
for i in range(len(CONST.available_sources)):
|
|
||||||
self.log.info('\t\t' + str(list(CONST.available_sources[i])))
|
|
||||||
self.log.info('\tDone!\n')
|
|
||||||
|
|
||||||
self.log.debug(json.dumps(CONST.gateway, indent=4))
|
|
||||||
self.log.debug(json.dumps(CONST.rooms, indent=4))
|
|
||||||
self.log.debug(json.dumps(CONST.devices, indent=4))
|
|
||||||
|
|
||||||
def get_masterlink_id(self, mlgw, mlcli):
|
|
||||||
self.log.info("Finding MasterLink ID of products:")
|
|
||||||
if mlgw.is_connected and mlcli.is_connected and CONST.devices:
|
|
||||||
for device in CONST.devices:
|
|
||||||
self.log.info("Finding MasterLink ID of product " + device.get('Device'))
|
|
||||||
# Ping the device with a light timeout to elicit a ML telegram containing its ML_ID
|
|
||||||
mlgw.send_beo4_cmd(int(device.get('MLN')),
|
|
||||||
CONST.CMDS_DEST.get("AUDIO SOURCE"),
|
|
||||||
CONST.BEO4_CMDS.get("LIGHT TIMEOUT"))
|
|
||||||
|
|
||||||
if device.get('Serial_num') in [None, '']:
|
|
||||||
# If this is a MasterLink product it has no serial number...
|
|
||||||
# loop to until expected response received from ML Command Line Interface
|
|
||||||
test = True
|
|
||||||
while test:
|
|
||||||
try:
|
|
||||||
if mlcli.last_message['from_device'] == "MLGW" and \
|
|
||||||
mlcli.last_message['payload_type'] == "MLGW_REMOTE_BEO4" and \
|
|
||||||
mlcli.last_message['State_Update']['command'] == "Light Timeout":
|
|
||||||
|
|
||||||
device['ML_ID'] = mlcli.last_message.get('to_device')
|
|
||||||
device['Serial_num'] = 'NA'
|
|
||||||
self.log.info("\tMasterLink ID of product " +
|
|
||||||
device.get('Device') + " is " + device.get('ML_ID') + ".\n")
|
|
||||||
test = False
|
|
||||||
except KeyError:
|
|
||||||
asyncore.loop(count=1, timeout=0.2)
|
|
||||||
|
|
||||||
else:
|
|
||||||
# If this is a NetLink product then it has a serial number and no ML_ID
|
|
||||||
if device['Device'] == 'BeoMaster 7000':
|
|
||||||
device['ML_ID'] = "AUDIO MASTER"
|
|
||||||
self.log.info("\tNetworkLink ID of product " + device.get('Device') + " is " +
|
|
||||||
device.get('Serial_num') + ". MasterLink ID manually assigned to AUDIO MASTER.\n")
|
|
||||||
else:
|
|
||||||
device['ML_ID'] = 'NA'
|
|
||||||
self.log.info("\tNetworkLink ID of product " + device.get('Device') + " is " +
|
|
||||||
device.get('Serial_num') + ". No MasterLink ID assigned.\n")
|
|
||||||
|
|
||||||
self.log.debug(json.dumps(device, indent=4))
|
|
||||||
|
|
||||||
# ########################################################################################
|
|
||||||
# Utility Functions
|
|
||||||
@staticmethod
|
|
||||||
def _srcdictsanitize(d, s):
|
|
||||||
result = d.get(s)
|
|
||||||
if result is None:
|
|
||||||
result = s
|
|
||||||
return str(result)
|
|
||||||
Loading…
Reference in a new issue