m2_g2_fs/app/ispeaker-service/app/device.py

import time
from app import config, log, register, project_cf, speaker_cf, volume_cf, redisObj
import paho.mqtt.client as mqtt
import json
from app.app_config import *
from app.player import VLCController
import sys
from time import sleep
from threading import Lock
import os
import threading
import signal
from app.config import Config


"""
    desc:
        自定义超时异常
"""
class TimeoutError(Exception):
    def __init__(self, msg):
        super(TimeoutError, self).__init__()
        self.msg = msg


def time_out(interval, callback, msg):
    def decorator(func):
        def handler(signum, frame):
            raise TimeoutError(msg)

        def wrapper(*args, **kwargs):
            try:
                signal.signal(signal.SIGALRM, handler)
                signal.alarm(interval)       # interval秒后向进程发送SIGALRM信号
                result = func(*args, **kwargs)
                signal.alarm(0)              # 函数在规定时间执行完后关闭alarm闹钟
                return result
            except TimeoutError as e:
                callback(e)
        return wrapper
    return decorator

def callback_timeout(e):
    log.logger.error(e.msg)

"""
    desc:
        项目主要功能类
"""
class Device(object):
    def __init__(self, config = config):
        self.model = config.model
        self.soft_version = config.soft_version
        self.hard_version = config.hard_version
        self.hostname = config.hostname
        self.ipaddr = config.ipaddr
        self.hard_volume = config.hard_volume
        self.hard_volume_control = config.hard_volume_control
        self.exten = config.exten
        self.exten_password = config.exten_password
        self.server_ipaddr = config.server_ipaddr
        self.mac = config.mac
        self.connect_flags = 1
        self.player_current_volume = config.player_current_volume
        self.init_player_state = config.init_player_state
        self.current_uri = config.current_uri

        self.fifo_flag = 0
        self.pull_rtsp = 0
        self.broadcast_volume_lock = Lock()
        if not self._check_config():
            sys.exit(1)
        
        # self.player = Player("--aout=alsa", "--audio-resampler=soxr", "--network-caching=50", "--network-synchronisation", "--clock-synchro=-1", "--cr-average=500", "--clock-jitter=50", "--live-caching=100")
        self.cmd = [
            "/usr/sbin/vlc",
            "--control", "rc",
            "--rc-fake-tty",
            "--quiet",
            "--aout=alsa",
            "--gain-init=0.0",
            "--audio-resampler=soxr",
            "--network-caching=50",
            "--network-synchronisation",
            "--clock-synchro=-1",
            "--cr-average=500",
            "--clock-jitter=50",
            "--live-caching=100",
        ]
        self.player = VLCController(cmd=self.cmd, on_state_change=self.vlc_callback)
        # self.player = VLCController(on_state_change=self.vlc_callback)
        self.task_player_state = -1
        self.task_soft_volume = -1
        self.task_uri = ""
        self.player_publish_volume = config.player_current_volume
        self.publish_uri = ""
        self.task_flag = "normal"
        self.publish_uri_flag = 0
        self.old_player_state = 0
        self.fifo_uri = ""

        self.mqttClient = mqtt.Client(client_id = self.mac, clean_session = True)
        self.mqttClient.username_pw_set(username=self.exten, password=self.exten_password)
        self.mqttClient.on_connect = self.on_connect

        self.redis_sub_volume = redisObj.subscribe_volume()
        self.audioControlStatus = False
        self.upgradeFlag = False

        self.currentScreenName = ""
    
    def _update_data(self):
        new_config = Config()
        self.model = new_config.model
        self.soft_version = new_config.soft_version
        self.hard_version = new_config.hard_version
        self.hostname = new_config.hostname
        self.ipaddr = new_config.ipaddr
        self.hard_volume = new_config.hard_volume
        self.hard_volume_control = new_config.hard_volume_control
        self.exten = new_config.exten
        self.exten_password = new_config.exten_password
        self.server_ipaddr = new_config.server_ipaddr
        self.mac = new_config.mac
        self.player_current_volume = new_config.player_current_volume
        self.init_player_state = new_config.init_player_state
        self.current_uri = new_config.current_uri
        self.player_publish_volume = new_config.player_current_volume
    
    def vlc_callback(self, state):
        if state == 3:
            self.MediaPlayerPlaying_cb()
        elif state == -1 or state == 0  or state == 5:
            self.MediaPlayerStopped_cb()
        # else:
        #     log.logger.info("VLC state:%d not need handle" % (state))

    def start_relay_screen_action(self):
        speaker_cf.read(SPEAKER_CONFIG_FILE, encoding='utf-8')
        #广播音乐触发继电器
        relay_play_music = speaker_cf.get("relay_ctrl", "play_music")
        #告警触发继电器
        relay_alarm = speaker_cf.get("relay_ctrl", "alarm")

        url = self._handle_url(self.current_uri)
        log.logger.info("trigger url:%s" % (url))

        if url.startswith("rtp"):
            if relay_play_music != "Disabled":
                data = {}
                data["id"] = "gpio67"
                data["type"] = relay_play_music
                redisObj.lpush(json.dumps(data))
                log.logger.info("Trigger start play music relay")
            
            # if led_play_music != "Disabled":
            #     data = {}
            #     data["id"] = "gpio135"
            #     data["type"] = led_play_music
            #     redisObj.lpush(json.dumps(data))
            #     log.logger.info("Trigger play music flash led")
            data = {}
            data["action"] = "on"
            data["name"] = "PlayMusic"
            self.currentScreenName = "PlayMusic"
            redisObj.screen_publish(json.dumps(data))
            log.logger.info("Trigger start play music to screen")
        elif url.startswith("rtsp"):
            if "sourceId" in url:
                id = int(url.rsplit("-", 1)[-1])
                if id > 10:
                    if relay_play_music != "Disabled":
                        data = {}
                        data["id"] = "gpio67"
                        data["type"] = relay_play_music
                        redisObj.lpush(json.dumps(data))
                        log.logger.info("Trigger start play music relay")
                    
                    #触发场景联动
                    data = {}
                    data["action"] = "on"
                    data["name"] = "PlayMusic"
                    self.currentScreenName = "PlayMusic"
                    redisObj.screen_publish(json.dumps(data))
                    log.logger.info("Trigger start play music to screen")

                if id > 0 and id <= 10:
                    if relay_alarm != "Disabled":
                        data = {}
                        data["id"] = "gpio67"
                        data["type"] = relay_alarm
                        redisObj.lpush(json.dumps(data))
                        log.logger.info("Trigger start play alarm relay")

                    #触发场景联动
                    data = {}
                    data["action"] = "on"
                    data["name"] = "Alarm"
                    self.currentScreenName = "Alarm"
                    redisObj.screen_publish(json.dumps(data))
                    log.logger.info("Trigger start play alarm to screen")
            else:
                if relay_play_music != "Disabled":
                    data = {}
                    data["id"] = "gpio67"
                    data["type"] = relay_play_music
                    redisObj.lpush(json.dumps(data))
                    log.logger.info("Trigger start play music relay")
                
                data = {}
                data["action"] = "on"
                data["name"] = "PlayMusic"
                self.currentScreenName = "PlayMusic"
                redisObj.screen_publish(json.dumps(data))
                log.logger.info("Trigger start play music to screen")
        else:
            if "sourceId" in url:
                id = int(url.rsplit("-", 1)[1])
                if id > 10:
                    if relay_play_music != "Disabled":
                        data = {}
                        data["id"] = "gpio67"
                        data["type"] = relay_play_music
                        redisObj.lpush(json.dumps(data))
                        log.logger.info("Trigger start play music relay")
                    
                    data = {}
                    data["action"] = "on"
                    data["name"] = "PlayMusic"
                    self.currentScreenName = "PlayMusic"
                    redisObj.screen_publish(json.dumps(data))
                    log.logger.info("Trigger start play music to screen")

                if id > 0 and id <= 10:
                    if relay_alarm != "Disabled":
                        data = {}
                        data["id"] = "gpio67"
                        data["type"] = relay_alarm
                        redisObj.lpush(json.dumps(data))
                        log.logger.info("Trigger start alarm relay")
                    
                    data = {}
                    data["action"] = "on"
                    data["name"] = "Alarm"
                    self.currentScreenName = "Alarm"
                    redisObj.screen_publish(json.dumps(data))
                    log.logger.info("Trigger start alarm to screen")
            else:
                if relay_play_music != "Disabled":
                    data = {}
                    data["id"] = "gpio67"
                    data["type"] = relay_play_music
                    redisObj.lpush(json.dumps(data))
                    log.logger.info("Trigger start play music relay")
                
                data = {}
                data["action"] = "on"
                data["name"] = "PlayMusic"
                self.currentScreenName = "PlayMusic"
                redisObj.screen_publish(json.dumps(data))
                log.logger.info("Trigger start play music to screen")

    def stop_relay_screen_action(self):
        data = {}
        data["action"] = "off"
        data["name"] = self.currentScreenName
        redisObj.screen_publish(json.dumps(data))
        log.logger.info("Trigger stop to screen")
        speaker_cf.read(SPEAKER_CONFIG_FILE, encoding='utf-8')

        #广播音乐触发继电器
        relay_play_music = speaker_cf.get("relay_ctrl", "play_music")
        #告警触发继电器
        relay_alarm = speaker_cf.get("relay_ctrl", "alarm")

        if relay_play_music != "Disabled":
            data = {}
            data["id"] = "gpio67"
            data["type"] = "Off"
            redisObj.lpush(json.dumps(data))
            log.logger.info("Trigger stop play music relay")
        
        if relay_alarm != "Disabled":
            data = {}
            data["id"] = "gpio67"
            data["type"] = "Off"
            redisObj.lpush(json.dumps(data))
            log.logger.info("Trigger stop alerm relay")
        
    """
        desc:
            开始播放事件回调,控制声音渐入
    """
    def MediaPlayerPlaying_cb(self):
        self.player.set_volume(self.player_publish_volume)
        log.logger.info("MediaPlayerPlaying_cb")
        data = {}
        data["name"] = "BROADCAST"
        data["action"] = "on"
        redisObj.volumeEventPush(json.dumps(data))
        self.start_relay_screen_action()

    def MediaPlayerStopped_cb(self):
        log.logger.info("MediaPlayerStopped_cb")
        data = {}
        data["name"] = "BROADCAST"
        data["action"] = "off"
        redisObj.volumeEventPush(json.dumps(data))
        self.stop_relay_screen_action()

    """
        desc:
            系统信号处理函数，主要处理键盘输入，如ctrl+c等
        
        Parameters:
                param1 - 信号数字
                param2 - 处理方法
    """
    def singal_handler(self, singal_num, handler):
        self.mqttClient.loop_stop()
        self.mqttClient.disconnect()
        self.connect_flags = 0
        self.stop_pull_rtsp()
        if self.player:
            self.player.quit()
        log.logger.info("Mqtt disconnect now!")
        redisObj.logpush("MQTT,STATUS,Mqtt disconnect")
        sys.exit(0)
    
    """
        desc:
            连接mqtt服务器
    """
    def run(self):
        while self.connect_flags:
            try:
                log.logger.info("Mqtt to %s:%d, connectting..." % (self.server_ipaddr, SERVER_MQTT_PORT))
                self.mqttClient.connect(self.server_ipaddr, SERVER_MQTT_PORT, MQTT_CONNECT_TIMEOUT)
                self.connect_flags = 0
            except:
                log.logger.error("Mqtt connect to %s:%d error" % (self.server_ipaddr, SERVER_MQTT_PORT))
                log.logger.error("Mqtt reconnect...")
            sleep(RETRY_SEC)
        self.mqttClient.loop_start()

        self.mqttClient.on_message = register.recieve_run
        self.mqttClient.on_subcribe = self.on_subcribe
        self.mqttClient.on_disconnect = self.on_disconnect
        self.mqttClient.on_publish = self.on_publish
        self.mqttClient.on_socket_close = self.on_socket_close
        self.mqttClient.subscribe(TOPIC_COMMAND, MQTT_QOS)
        return True
    
    """
        desc:
            mqtt连接事件回调函数
    """
    def on_connect(self, client, userdata, flags, rc):
        log.logger.info(mqtt.connack_string(rc))
        self.old_player_state = 0
        if self.current_uri != "" and self.init_player_state == 3:
            if self.player.get_state() != 3:
                data = {}
                data.setdefault("url", self.current_uri)
                data.setdefault("type", "normal")
                try:
                    self.play(data)
                except:
                    log.logger.error("Player on_connect(play) is error")
                log.logger.info("On_connect finished!")
                sleep(1)
        else:
            log.logger.info("Not need reconnect!")
        self._update_data()
        init_data = self._get_init_data() 
        self.mqttClient.publish(TOPIC_EVENT, json.dumps(init_data).encode("utf-8"), MQTT_QOS)
        log.logger.info("Publish data: %s" % (init_data))
        return True

    """
        desc:
            组装mqtt连接到的服务器时需上报的初始化数据
    """
    def _get_init_data(self):
        init_data = {}
        data = {}
        data.setdefault(SOLF_VOLUME, self.player_current_volume)
        data.setdefault(HARD_VOLUME, self.hard_volume)
        data.setdefault("hard-volume-control", self.hard_volume_control)
        data.setdefault(EXTEN, self.exten)
        data.setdefault(DEVICE_MODEL, self.model)
        data.setdefault(PRIVATE_IP, self.ipaddr)
        data.setdefault("soft-version", self.soft_version)
        data.setdefault("hard-version", self.hard_version)
        status = self.player.get_state()
        if status == 3 or status == 2:
            data.setdefault(STATUS, PLAYER_PLAYING)
            init_send_uri = self._handle_url(self.current_uri)
            data.setdefault("sourceId", init_send_uri)
        elif status == 4:
            data.setdefault(STATUS, PLAYER_PAUSE)
        elif status == -1 or status == 0:
            data.setdefault(STATUS, PLAYER_IDLE)
        else:
            data.setdefault(STATUS, PLAYER_ERROR)

        if os.path.exists("/oem/.onvif_upgrade/mqtt_upgraded"):
            data.setdefault("upgrade-status", "UpgradeSuccess")
            os.remove("/oem/.onvif_upgrade/mqtt_upgraded")

        init_data.setdefault(ACTION_NAME, "init")
        init_data.setdefault(DATA, data)
        return init_data
    
    """
        desc:
            设置播放器音量
        
        Parameters:
                param1 - 音量
                param2 - 类型
    """
    def set_soft_volume(self, volume, type = "default"):
        self.player_current_volume = volume
        self.player_publish_volume = volume
        volume_cf.read(VOLUME_CONFIG_FILE, encoding='utf-8')
        volume_cf.set("volume", "broadcast_volume", str(volume))
        with open(VOLUME_CONFIG_FILE, "w") as f:
            volume_cf.write(f, space_around_delimiters=False)
        os.system("sync")

        webMsg = {}
        appVolume = {}
        webMsg["type"] = "broadcast"
        webMsg["appVolume"]=appVolume
        appVolume["broadcast_volume"]=volume
        redisObj.volumeItemPush(json.dumps(webMsg))

        if self.player.get_state() == 3 or self.player.get_state() == 2:
            self.player.set_volume(volume)
        
        return True
    
    def set_event_volume(self, volume):
        self.broadcast_volume_lock.acquire()
        if volume == -1:
            self.player_current_volume = 0
            self.player_publish_volume = 0
        else:
            self.player_current_volume = volume
            self.player_publish_volume = volume
        if volume == -1:
            self.player.set_volume(0)
        else:
            self.player.set_volume(volume)

        self.broadcast_volume_lock.release()

        return True

    """
        desc:
            task设置播放器音量
        
        Parameters:
                param1 - 音量
                param2 - 类型
    """
    def set_task_soft_volume(self, volume):
        self.task_soft_volume = volume
        self.player_publish_volume = volume

        if self.player.get_state() == 3 or self.player.get_state() == 2:
            self.player.set_volume(volume)
        
        return True

    """
        desc:
            设置系统相关数据
        
        Parameters:
                param1 - 相关数据
    """
    def set_service_settings(self, data):
        if "soft-volume" in data:
            self.player_current_volume = int(data.get("soft-volume"))
            self.player_publish_volume = int(data.get("soft-volume"))

        if "hard-volume" in data:
            self.set_hard_volume(int(data.get("hard-volume")))

        if "hard-volume-control" in data:
            self.set_hard_volume_control(data.get("hard-volume-control"))

        return True

    """
        desc:
            开始拉取本地摄像头的rtsp流并推送到服务器
        
        Parameters:
                param1 - 本地摄像头rtsp地址
                param2 - 服务器的rtsp地址
    """
    def start_pull_rtsp(self, src, dst):
        if not src:
            log.Logger.error("src not found")
            return False

        if not dst:
            log.Logger.error("dst not found")
            return False
        else:
            if not dst.startswith("rtsp"):
                dst = ''.join(["rtsp://", self.server_ipaddr, "/", dst])

        tmp_pull_rtsp = self.pull_rtsp
        self.stop_pull_rtsp()
        self.pull_rtsp = tmp_pull_rtsp

        sleep(1)
        os.system("%s %s %s > /dev/null 2>&1 &" % (RTSP_PUSHER, src, dst))
        log.logger.info("%s %s %s > /dev/null 2>&1 &" % (RTSP_PUSHER, src, dst))
        self.pull_rtsp = self.pull_rtsp + 1
        log.logger.info("self.pull_rtsp:%d" % (self.pull_rtsp))
        return True

    """
        desc:
            停止拉取本地摄像头的rtsp流
    """
    def stop_pull_rtsp(self):
        os.system("/usr/bin/killall rtsp_pull_push > /dev/null 2>&1 &")
        self.pull_rtsp = 0
        return True
    
    def publishActionStatus(self, id, action, status):
        data = {}
        event = {}
        event.setdefault("type", TOPIC_RESPONSE)
        event.setdefault("id", id)
        event.setdefault(DATA, data)
        data.setdefault("action", action)
        data.setdefault("status", status)
        # try:
        #     self._publish_msg(event)
        # except ERROR:
        #     log.logger.error("Publish event:\n%s" % (ERROR))
        self.mqttClient.publish(TOPIC_RESPONSE, json.dumps(event).encode('utf-8'), MQTT_QOS)

    def _reboot(self):
        os.system("/sbin/reboot")

    """
        desc:
            重启设备
    """
    def reboot(self, id):
        self.publishActionStatus(id, "reboot", "ok")
        sleep(0.5)
        self._reboot()
        return True

    def _reset(self):
        os.system("/bin/rm -rf /oem/.userdata && /bin/sync && /sbin/reboot")

    """
        desc:
            重置设备
    """
    def reset(self, id):
        self.publishActionStatus(id, "reset", "ok")
        sleep(0.5)
        self._reset()
        return True
    
    """
        desc:
            升级设备
    """
    def upgrade(self, id, data):
        if data.get("url"):
            reset= data.get("reset") or "no"
            os.system("/etc/scripts/upgrade.sh pre")
            cmd = "/etc/scripts/onvif_upgrade.sh upgrade " + data.get("url") + " " + reset + " >/dev/null 2>&1 &"
            os.system(cmd)
        else:
            # self.publishActionStatus(id, "upgrade", "missArgs")
            os.system("echo -n 'MissArgs' > /oem/.onvif_upgrade/status")
            return False
        
        # self.publishActionStatus(id, "upgrade", "Downloading")

        self.upgradeFlag = True
        return True
    
    """
        desc:
            获取升级状态
    """
    def getUpgradeStatus(self, id):
        statusPath="/oem/.onvif_upgrade/status"
        status=""
        try:
            with open(statusPath, 'r') as file:
                status = file.read()
        except IOError:
            log.logger.error("无法打开文件:%s" % statusPath)
            self.publishActionStatus(id, "upgrade-status", "ReadError")
            return True

        self.publishActionStatus(id, "upgrade-status", status)
        return True
    
    """
        desc:
            获取升级状态
    """
    def getUpgradeStatusFromFile(self):
        statusPath="/oem/.onvif_upgrade/status"
        status=""
        try:
            with open(statusPath, 'r') as file:
                status = file.read()
        except IOError:
            log.logger.error("无法打开文件:%s" % statusPath)
            return "ReadError"

        return status
    
    """
        desc:
            导入配置
    """
    def importConfig(self, id, data):
        if data.get("url"):
            self.publishActionStatus(id, "import-config", "ok")
            cmd = "/etc/scripts/auto_privisioning.sh dhcp " + data.get("url") + " mqtt" + " >/dev/null 2>&1 &"
            os.system(cmd)
        else:
            self.publishActionStatus(id, "upgrade", "missArgs")
            return False
        
        return True
    
    """
        desc:
            继电器控制
    """
    def relayControl(self, id, data):
        if not "switch" in data:
            self.publishActionStatus(id, "relay-control", "missArgs")
            return False
        
        duration = {}
        relayInfo = {}
        relayInfo["data"] = duration
        relayInfo["type"] = "relay"
        if data["switch"] == "on":
            relayInfo["action"] = "on"
            if "duration" in data:
                duration["duration"] = data["duration"]
            else:
                duration["duration"] = 0
        else:
            relayInfo["action"] = "off"
            duration["duration"] = 0
        
        redisObj.apipush(json.dumps(relayInfo))
        
        return True


    """
        desc:
            设置系统的音量
        
        Parameters:
                param1 - 系统音量值
    """
    def set_hard_volume(self, volume):
        if volume and volume < 0:
            real_hard_volume = 0
        elif volume and volume > 30:
            real_hard_volume = 0
        else:
            real_hard_volume = volume

        try:
            os.system("/etc/scripts/set_volume.sh set %d - > /dev/null 2>&1 &" % (real_hard_volume))
            volume_cf.read(VOLUME_CONFIG_FILE, encoding='utf-8')
            volume_cf.set("volume", "volume_out", str(real_hard_volume))
            with open(VOLUME_CONFIG_FILE, "w") as f:
                volume_cf.write(f, space_around_delimiters=False) 
            os.system("sync")
            self.hard_volume = real_hard_volume
        except ERROR:
            log.logger.error("Set hard volume error:\n%s" % (ERROR))
            return False
        
        return True
    
    """
        desc:
            设置强控，不允许通过硬件按键修改系统音量
        
        Parameters:
                param1 - 强控开关
    """
    def set_hard_volume_control(self, value):
        if value:
            try:
                volume_cf.read(VOLUME_CONFIG_FILE, encoding='utf-8')
                volume_cf.set("volume", "hard_volume_control", value)
                with open(VOLUME_CONFIG_FILE, "w") as f:
                    volume_cf.write(f, space_around_delimiters=False)
                os.system("sync")
                self.hard_volume_control = value
            except ERROR:
                log.logger.error("Set hard volume control value:\n%s" % (ERROR))
                return False
        else:
            log.logger.error("Set hard volume control value not exist")
            return False
        return True

    """
        desc:
            设置系统分机信息
        
        Parameters:
                param1 - 服务器地址
                param2 - 用户名(分机号)
                param3 - 分机密码
    """
    def set_exten(self, host, exten, password):
        if host and exten and password:
            try:
                speaker_cf.read(SPEAKER_CONFIG_FILE, encoding='utf-8')
                speaker_cf.set("account_info_1", "username", exten)
                speaker_cf.set("account_info_1", "passwd", password)
                speaker_cf.set("account_info_1", "server", host)
                with open(SPEAKER_CONFIG_FILE, "w") as f:
                        speaker_cf.write(f, space_around_delimiters=False)
                os.system("sync")
            except ERROR:
                log.logger.error("Set exten error:\n%s" % (ERROR))
                return False
        else:
            log.logger.error("Set exten value not full")
            return False
        
        return True
                
    """
        desc:
            vlc播放指定数据
        
        Parameters:
                param1 - 播放数据
    """
    def play(self, data):
        if "url" in data:
            url = data.get("url")
            self.publish_uri = self._handle_url(url)
            if self.publish_uri_flag == 1:
                self.publish_uri_flag = 0
            else:
                self.publish_uri_flag = 1
            # if url.startswith("x10-"):
            #    url_type=1
            # else:
            #     url_type=0
            if url.startswith("rtp"):
                pass
            elif url.startswith("rtsp"):
                pass
            else:
                if not url.startswith("http"):
                    url = ''.join(["http://", self.server_ipaddr, ":", SERVER_ICE_PORT, "/", url])
        else:
            log.logger.warning("Url not found!")
            return False

        if "soft-volume" in data:
            if "type" in data: 
                if data.get("type") == "task":
                    self.set_task_soft_volume(int(data.get("soft-volume")))
                else:
                    self.set_soft_volume(int(data.get("soft-volume")))
            else:
                self.set_soft_volume(int(data.get("soft-volume")))
        
        if "hard-volume" in data:
            self.set_hard_volume(int(data.get("hard-volume")))

        if "hard-volume-control" in data:
            self.set_hard_volume_control(data.get("hard-volume-control"))

        if "type" in data: 
            if data.get("type") == "task":
                self.task_player_state = 3
                self.task_uri = url
                self.task_flag = "task"
            else:
                self.init_player_state = 3
                self.current_uri = url
                self.task_flag = "normal"
        else:
            self.init_player_state = 3
            self.current_uri = url
            self.task_flag = "normal"

        try:
            if not self.player.running:
                self.player.start_vlc()
                self.player.play(url=url)
            else:
                self.player.play(url=url)
                self.player.set_volume(self.player_publish_volume)

            if "type" in data and data.get("type") == "task":
                if data.get("type") != "task":
                    if "soft-volume" in data:
                        self.task_soft_volume = int(data.get("soft-volume"))
            else:
                try:
                    project_cf.read(PROJECT_CONFIG, encoding='utf-8')
                    project_cf.set("general", "init_player_state", str(self.init_player_state))
                    project_cf.set("general", "current_uri", str(self.current_uri))
                    with open(PROJECT_CONFIG, "w") as f:
                        project_cf.write(f, space_around_delimiters=False)
                    os.system("sync")
                except ERROR:
                    log.logger.error("Save init player state or current uri error:\n%s" % (ERROR))
                    return False
        except:
            log.logger.error("Player play uri:%s is error" % (url))
            return False

        return True
    
    """
        desc:
            播放器暂停
        
        Parameters:
                param1 - 暂停模式
    """
    def pause(self, value):
        if self.audioControlStatus:
            return True
        self.init_player_state = 0
        self.task_player_state = -1
        try:
            self.player.stop()
        except:
            log.logger.error("Player pause(stop) is error")
        
        try:
            project_cf.read(PROJECT_CONFIG, encoding='utf-8')
            project_cf.set("general", "init_player_state", str(self.init_player_state))
            with open(PROJECT_CONFIG, "w") as f:
                project_cf.write(f, space_around_delimiters=False)
            os.system("sync")
        except ERROR:
            log.logger.error("Save init player state or current uri error:\n%s" % (ERROR))
            
        return True
    
    """
        desc:
            播放器恢复播放
        
        Parameters:
                param1 - 恢复数据
    """
    def resume(self, value):
        if self.audioControlStatus:
            return True
        self.init_player_state = 3
        self.task_player_state = -1
        self.task_soft_volume = -1
        self.task_uri = ""
        self.set_soft_volume(self.player_current_volume)
        try:
            project_cf.read(PROJECT_CONFIG, encoding='utf-8')
            project_cf.set("general", "init_player_state", str(self.init_player_state))
            with open(PROJECT_CONFIG, "w") as f:
                project_cf.write(f, space_around_delimiters=False)
            os.system("sync")
        except ERROR:
            log.logger.error("Save init player state or current uri error:\n%s" % (ERROR))
        
        if value == "play":
            data = {}
            if self.current_uri != "" and self.init_player_state == 3:
                data = {}
                data.setdefault("url", self.current_uri)
                try:
                    self.play(data)
                except:
                    log.logger.error("Player resume(play) is error")
            else:
                self.stop()
        else:
            try:
                self.player.start_vlc()
                self.player.play()
            except:
                log.logger.error("Player resume is error")
        
        return True
    
    def _publish_stop_event(self):
        data = {}
        event = {}
        event.setdefault(ACTION_NAME, TOPIC_UPDATE)
        event.setdefault(DATA, data)
        data.setdefault(STATUS, PLAYER_IDLE)
        try:
            self.mqttClient.publish(TOPIC_EVENT, json.dumps(event).encode('utf-8'), MQTT_QOS)
            log.logger.info("Publish event:%s" % (event))
        except ERROR:
            log.logger.error("Publish stop event error:\n%s" % (ERROR))
    
    """
        desc:
            播放器停止播放
    """
    def stop(self):
        self.init_player_state = -1
        self.current_uri = ""
        self.task_player_state = -1
        self.task_soft_volume = -1
        self.task_uri = ""
        try:
            self.player.quit()
            try:
                self.t = threading.Thread(target = self._publish_stop_event, args = ())
                self.t.setDaemon(True)
                self.t.start()
            except ERROR:
                log.logger.error(ERROR)

            try:
                project_cf.read(PROJECT_CONFIG, encoding='utf-8')
                project_cf.set("general", "init_player_state", str(self.init_player_state))
                project_cf.set("general", "current_uri", "")
                with open(PROJECT_CONFIG, "w") as f:
                    project_cf.write(f, space_around_delimiters=False)
                os.system("sync")
            except ERROR:
                log.logger.error("Save init player state or current uri error:\n%s" % (ERROR))
                return False
        except:
            log.logger.error("Player stop is error")
            return False
        return True
    
    def reconnect_loop(self):
        while True:
            try:
                log.logger.info("Attempting reconnect...")
                self.mqttClient.reconnect()
                log.logger.info("Reconnected successfully.")
                break
            except Exception as e:
                log.logger.error("Reconnect failed:", e)
                sleep(3)
    
    def on_subcribe(self, client, userdata, mid, granted_qos):
        pass

    def on_publish(self, client, userdata, mid):
        pass

    def on_disconnect(self, client, userdata, rc):
        log.logger.error("on_disconnect: %d" % (rc))
        redisObj.logpush("MQTT,STATUS,Mqtt disconnect")
        #mqtt重连时，是否恢复播放
        music_auto_resume = speaker_cf.get("system", "music_auto_resume")
        if music_auto_resume == "no":
            self.stop()
            log.logger.info("Not need resume play, stop play")
        #        if rc != 0:
        #            # 异步启动重连
        #            threading.Thread(target=self.reconnect_loop, daemon=True).start()
        #        else:
        #            log.logger.info("normal disconnect")

    """
        desc:
            检测系统的必要配置信息
    """
    def _check_config(self):
        if not self.model:
            log.logger.error("Model not found!")
            return False
        elif not self.mac:
            log.logger.error("Mac not found!")
            return False
        elif not self.server_ipaddr:
            log.logger.error("Server Ipaddr not found!")
            return False
        elif not self.exten:
            log.logger.error("Extension not found!")
            return False
        elif self.hard_volume < 0:
            log.logger.error("Hard Volume error!")
            return False
        else:
            return True
    
    @time_out(2, callback_timeout, "Read volume out timeout!")
    def _read_volume_out(self):
        volume_cf.read(VOLUME_CONFIG_FILE, encoding='utf-8')
        return int(volume_cf.get("volume", "volume_out"))

    def _get_volume_out(self):
        volume_out = self._read_volume_out()
        if volume_out:
            return volume_out
        else:
            return 0
    
    @time_out(2, callback_timeout, "Read volume control timeout!")
    def _read_hard_volume_control(self):
        volume_cf.read(VOLUME_CONFIG_FILE, encoding='utf-8')
        return volume_cf.get("volume", "hard_volume_control")
    
    def _get_hard_volume_control(self):
        hard_volume_control = self._read_hard_volume_control()

        if hard_volume_control:
            return hard_volume_control
        else:
            return "on"

    @time_out(5, callback_timeout, "Publish message timeout!")
    def _publish_msg(self, event):
        self.mqttClient.publish(TOPIC_EVENT, json.dumps(event).encode('utf-8'), MQTT_QOS)
        log.logger.info("Publish event:%s" % (event))

        
    """
        desc:
            监测系统状态变化,如音量，播放状态等
    """
    def _check_variable(self):
        old_soft_volume = self.player_publish_volume
        old_hard_volume = self.hard_volume
        old_exten = self.exten
        #old_player_state = self.plaer.get_state()
        old_hard_volume_control = self.hard_volume_control
        old_pull_rtsp = self.pull_rtsp
        old_publish_uri_flag = self.publish_uri_flag
        old_upgrade_status = ""
        while True:
            sleep(CHECK_INTERVAL)
            data = {}
            event = {}
            event.setdefault(ACTION_NAME, TOPIC_UPDATE)
            event.setdefault(DATA, data)

            if self.upgradeFlag:
                new_upgrade_status = self.getUpgradeStatusFromFile()
                #UpgradeFailed VerifyFailed downloadError MissArgs
                if new_upgrade_status == "UpgradeFailed" or new_upgrade_status == "VerifyFailed" or new_upgrade_status == "downloadError" or new_upgrade_status == "MissArgs":
                    self.upgradeFlag = False
                if new_upgrade_status != old_upgrade_status:
                    old_upgrade_status = new_upgrade_status
                    data.setdefault("upgrade-status", new_upgrade_status)
            try:
                new_soft_volume = self.player_publish_volume
                if old_soft_volume != new_soft_volume:
                    old_soft_volume = new_soft_volume
                    data.setdefault(SOLF_VOLUME, new_soft_volume)
            except ERROR:
                    log.logger.error("Checkout soft volume:%s" % (ERROR))
            
            try:
                new_hard_volume = self._get_volume_out()
                if old_hard_volume != new_hard_volume:
                    old_hard_volume = new_hard_volume
                    data.setdefault(HARD_VOLUME, new_hard_volume)
            except ERROR:
                    log.logger.error("Checkout hard volume:%s" % (ERROR))
            
            try:
                new_player_state = self.player.get_state()
                if self.old_player_state != new_player_state:
                    self.old_player_state = new_player_state

                    if new_player_state == -1 or new_player_state == 0:
                        player_status = PLAYER_IDLE
                    elif new_player_state == 4:
                        player_status = PLAYER_PAUSE
                    elif new_player_state == 3 or new_player_state == 2:
                        player_status = PLAYER_PLAYING
                        data.setdefault("sourceId", self.publish_uri)
                        data.setdefault(SOLF_VOLUME, self.player_publish_volume)
                    else:
                        player_status = PLAYER_ERROR
                    data.setdefault(STATUS, player_status)
            except ERROR:
                    log.logger.error("Checkout player state:%s" % (ERROR))
            
            try:
                new_hard_volume_control = self._get_hard_volume_control()
                if old_hard_volume_control != new_hard_volume_control:
                    old_hard_volume_control = new_hard_volume_control
                    data.setdefault(HARD_VOLUME_CONTROL, new_hard_volume_control)
            except ERROR:
                    log.logger.error("Checkout hard volume control:%s" % (ERROR))
            
            try:
                new_publish_uri_flag = self.publish_uri_flag
                if old_publish_uri_flag != new_publish_uri_flag:
                    old_publish_uri_flag = new_publish_uri_flag
                    data.setdefault("sourceId", self.publish_uri)
                    data.setdefault(SOLF_VOLUME, self.player_publish_volume)
            except ERROR:
                    log.logger.error("Checkout publish_uri:%s" % (ERROR))
            
            try:
                new_pull_rtsp = self.pull_rtsp
                if old_pull_rtsp != new_pull_rtsp:
                    old_pull_rtsp = new_pull_rtsp
                    if new_pull_rtsp != 0:
                        data.setdefault("pull-rtsp", 1)
                    else:
                        data.setdefault("pull-rtsp", 0)
            except ERROR:
                    log.logger.error("Checkout rtsp:%s" % (ERROR))

            if event.get(DATA) != {}:
                try:
                    self._publish_msg(event)
                except ERROR:
                    log.logger.error("Publish event:\n%s" % (ERROR))
                redisObj.logpush("MQTT,STATUS,%s" % json.dumps(data).replace(",", ";").replace('"', ''))


    def _handle_url(self, origin_url):
        if origin_url:
            if origin_url.startswith("rtp"):
                target_uri = origin_url
            elif origin_url.startswith("rtsp"):
                target_uri = origin_url
            else:
                if "/" in origin_url:
                    target_uri = origin_url.rsplit("/", 1)[1]
                else:
                    target_uri = origin_url
            
            return target_uri
        else:
            return ""

    """
        desc:
            监测fifo
    """
    def _fifo(self):
        if os.path.exists(FIFO_PATH):
            os.remove(FIFO_PATH)
        
        os.mkfifo(FIFO_PATH)
        
        while True:
            rf = os.open(FIFO_PATH, os.O_RDONLY)
            data = os.read(rf, 64)
            data = data.decode("utf-8")
            log.logger.info("received msg from fifo:%s" % (data))
            if data == "0":
                self.old_player_state = -1
                if self.player.get_state() == 3:
                    self.fifo_flag = 1
                    self.fifo_uri = self.current_uri
                    self.stop()
                    # os.system("/etc/scripts/pa_mute.sh 0 > /dev/null 2>&1 &")
            elif data == "1":
                if self.fifo_flag == 1:
                    self.fifo_flag = 0
                    play_data = {}
                    play_data.setdefault("url", self.fifo_uri)
                    try:
                        self.play(play_data)
                    except:
                        log.logger.error("Player play is error")
            else:
                log.logger.error("No support this action!")

            os.close(rf)
            
    """
        desc:
            启动监测线程
    """
    def check_start(self):
        try:
            self.t = threading.Thread(target = self._check_variable, args = ())
            self.t.setDaemon(True)
            self.t.start()

            return True
        except ERROR:
            log.logger.error(ERROR)

    """
        desc:
            DAEMON
    """
    def daemon(self):
        try:
            if not self.t.is_alive():
                log.logger.info("Data check is restarting...!")
                self.t = threading.Thread(target = self._check_variable, args = ())
                self.t.setDaemon(True)
                self.t.start()
                log.logger.info("Data check is restarted!")

            return True
        except ERROR:
            log.logger.error(ERROR)

    """
        desc:
            启动fifo监测
    """
    def check_fifo_start(self):
        try:
            t = threading.Thread(target = self._fifo, args = ())
            t.setDaemon(True)
            t.start()

            return True
        except ERROR:
            log.logger.error(ERROR)

    """
        desc:
            socket断线回调
    """
    def on_socket_close(self, client, userdata, sock):
        log.logger.error("Socket error(disconnect) occur!")
        redisObj.logpush("MQTT,STATUS,Mqtt disconnect")
    
    def handleRedisVolumeData(self, data):
        if data:
            if "broadcast_volume" in data:
                self.set_event_volume(data["broadcast_volume"]) 
                if data["broadcast_volume"] == -1:
                    log.logger.info("Set broadcast_volume:0")
                else:
                    log.logger.info("Set broadcast_volume:%d" % data["broadcast_volume"])
            else:
                pass

    def startParseRedisSubMsg(self):
        log.logger.info("Start parse redis sub...")
        while True:
            msg = self.redis_sub_volume.parse_response()
            # log.logger.info("msg:%s" % msg)
            try:
                data = json.loads(msg[2])
            except:
                log.logger.error("msg is not a json format:%s" % (msg[2]))
                continue

            t = threading.Thread(target=self.handleRedisVolumeData, args=[data])
            t.setDaemon(True)
            t.start()

    def startRedisSubMsg(self):
        t = threading.Thread(target=self.startParseRedisSubMsg, args=())
        t.setDaemon(True)
        t.start()