#!/usr/bin/env python3 # -*- coding: utf-8 -*- """ twitch-irl-docker Copyright (C) 2022 gpkvt This program is free software: you can redistribute it and/or modify it under the terms of the GNU Affero General Public License as published by the Free Software Foundation, either version 3 of the License, or (at your option) any later version. This program is distributed in the hope that it will be useful, but WITHOUT ANY WARRANTY; without even the implied warranty of MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the GNU Affero General Public License for more details. You should have received a copy of the GNU Affero General Public License along with this program. If not, see . """ import yaml import time import logging import memcache import requests as webrequests from pprint import pprint from lib import config conf = config.get_config() def login(liveu_username, liveu_password, liveu_url): login_header = { 'Origin': 'https://solo.liveu.tv', 'Authorization': 'Basic '+ liveu_password, 'Content-Type': 'application/json;charset=UTF-8', 'Accept': 'application/json, text/plain, */*', 'Referer': 'https://solo.liveu.tv/login', 'User-Agent': 'Mozilla/5.0 (Windows NT 10.0; WOW64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/75.0.3770.102 Safari/537.36', 'DNT': '1', 'x-user-name': liveu_username } login_payload = '{"return_to":"https://solo.liveu.tv/#/dashboard/units"}' if conf['DEBUG']: return "DEBUG" try: r = webrequests.post(liveu_url, data=login_payload.encode(), headers=login_header) rjson = r.json() if r.status_code == 200 and 'access_token' in rjson['data']['response']: access_token = rjson['data']['response']['access_token'] else: access_token = 'invalid' logging.fatal("Could not get access token") except Exception as e: logging.warning("Could not connect to service or invalid JSON response") logging.warning(e) access_token = False return access_token def get_data(access_token, liveu_unit, liveu_api, endpoint): api_header = { 'Accept': 'application/json, text/plain, */*', 'Referer': 'https://solo.liveu.tv/dashboard/units/' + str(liveu_unit), 'Origin': 'https://solo.liveu.tv', 'User-Agent': 'Mozilla/5.0 (Windows NT 10.0; WOW64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/75.0.3770.102 Safari/537.36', 'Authorization': 'Bearer ' + str(access_token), 'DNT': '1' } liveu_api = liveu_api + '/' + str(liveu_unit) + '/' + str(endpoint) if conf['DEBUG']: if endpoint == "status": ryaml = '{"deviceId":"Boss100_XXXXXXXXXXXXXXXXXXXXXXXXXXXXXX","upTime":2305.560333,"liveuVersion":null,"state":"streaming","battery":{"connected":true,"percentage":64,"runTimeToEmpty":100,"discharging":false,"charging":true},"lu1100":null,"modems":[{"connected":true,"name":"Ethernet","downlinkKbps":0,"uplinkKbps":632,"enabled":true,"port":"eth0","technology":"","upSignalQuality":4,"downSignalQuality":1,"activeSim":"","isCurrentlyRoaming":false,"kbps":632,"signalQuality":2},{"connected":false,"name":"No network","downlinkKbps":0,"uplinkKbps":0,"enabled":false,"port":"wlan0","technology":"","upSignalQuality":3,"downSignalQuality":3,"activeSim":"","isCurrentlyRoaming":false,"kbps":0,"signalQuality":3},{"connected":false,"name":"No Device","downlinkKbps":0,"uplinkKbps":0,"enabled":true,"port":"0","technology":"","upSignalQuality":3,"downSignalQuality":3,"activeSim":"","isCurrentlyRoaming":false,"kbps":0,"signalQuality":3},{"connected":false,"name":"No Device","downlinkKbps":0,"uplinkKbps":0,"enabled":true,"port":"1","technology":"","upSignalQuality":3,"downSignalQuality":3,"activeSim":"","isCurrentlyRoaming":false,"kbps":0,"signalQuality":3},{"connected":false,"name":"No Device","downlinkKbps":0,"uplinkKbps":0,"enabled":true,"port":"2","technology":"","upSignalQuality":3,"downSignalQuality":3,"activeSim":"","isCurrentlyRoaming":false,"kbps":0,"signalQuality":3},{"connected":false,"name":"No Device","downlinkKbps":0,"uplinkKbps":0,"enabled":true,"port":"3","technology":"","upSignalQuality":3,"downSignalQuality":3,"activeSim":"","isCurrentlyRoaming":false,"kbps":0,"signalQuality":3}],"videoMode":"1080p60","applicationType":"video","video":{"state":"streaming","channelUniqueId":null,"modemKeys":["eth0","wlan0","0","1","2","3"],"speedTest":null,"videoStandard":"1080p60"},"databridge":{"state":"idle","channelUniqueId":null,"modemKeys":["eth0","wlan0","0","1","2","3"],"speedTest":null},"filetransfer":null,"bitrate":632,"IsCameraConnected":false,"fetch logs":{"state":"idle","channelUniqueId":null,"modemKeys":null,"speedTest":null}}' ryaml = yaml.load(ryaml, Loader=yaml.FullLoader) else: try: r = webrequests.get(liveu_api, headers=api_header) ryaml = yaml.load(r.text, Loader=yaml.FullLoader) except Exception as e: logging.warning("Could get or parse information from service") logging.info(e) return ryaml def main(): logging.basicConfig(format='%(asctime)s %(module)s %(levelname)s: %(message)s') logging.getLogger().setLevel(conf['LOG_LEVEL']) mc = memcache.Client(['localhost:11211'], debug=0) counter = conf['LIVEU_TOKEN_TTL'] mc.set('LIVEU', conf['LIVEU']) # wait for other services to populate memcache data logging.info('Waiting for other services...') time.sleep(10) while conf['LIVEU']: logging.info('-------------------------------------------------------------------------------') if not mc.get('OBS_STREAMING_STATUS'): logging.info('OBS is not streaming, skipping...') time.sleep(300) continue if int(counter) >= int(conf['LIVEU_TOKEN_TTL']): logging.info("Getting LiveU access token") access_token = login(conf['LIVEU_USERNAME'], conf['LIVEU_PASSWORD'], conf['LIVEU_URL']) counter = 0 if access_token: logging.debug('Getting LiveU infos') data = get_data(access_token, conf['LIVEU_UNIT'], conf['LIVEU_API'], 'status') if data: if 'battery' in data: logging.info('LiveU Battery status') for key, value in data['battery'].items(): mc.set('LIVEU_BATTERY_'+str(key.upper()), float(value)) logging.info(str(key) +': ' + str(value)) else: logging.warning('Could not get liveU battery status') else: logging.error('Could not get data from liveU webservice') else: logging.error('Error obtaining access_token for liveU') logging.info('-------------------------------------------------------------------------------') time.sleep(30) if __name__ == "__main__": main()