package broadcast import ( "bytes" "context" "encoding/binary" "fmt" "io" "log" "net" "pbx-api-gin/internal/app/ami/action" "pbx-api-gin/internal/app/stc/active" msgdata "pbx-api-gin/internal/app/stc/data" "pbx-api-gin/internal/app/stc/priority" alstatus "pbx-api-gin/internal/app/stc/sendstatus" "pbx-api-gin/pkg/lfshook" "pbx-api-gin/pkg/utils" "strconv" "sync" "time" ) var tagLog = 0 func HandleStcCmd(ctx context.Context, conn net.Conn) { for { select { case <-ctx.Done(): return default: var buf bytes.Buffer tmp := make([]byte, 1024) if conn != nil { n, err := conn.Read(tmp) if err != nil { if err != io.EOF { conn.Close() } return } buf.Write(tmp[:n]) } /*for { packet, err := msgdata.ExtractPacket(&buf) if err != nil { lfshook.NewLogger().Logger.Infof("ExtractPacket:%+v", err) break } go processPacket(packet) }*/ //lfshook.NewLogger().Logger.Infof("buf:%x==============================", buf.Bytes()) for { packet, err := msgdata.ExtractPacket(&buf) if err != nil { log.Printf("parse error: %v, resetting buffer", err) buf.Reset() // 解析失败,清空避免污染 break } if packet == nil { break // 当前无完整包,等待下次 ReadFrom } //处理 packet... go processPacket(packet) } } } } // get train info and pacu info from heartbeat func processHeartbeat(info []byte) { //dataCount := info[0] //Get train info value := binary.BigEndian.Uint16(info[2:4]) //info[2] + info[3] active.TrainNum = "TS" + strconv.Itoa(int(value)) //lfshook.NewLogger().Logger.Infof("TrainNum: %s", active.TrainNum) //Get ICP volume icpBit8 := info[9] & 0xF icpBit1 := (info[9] >> 4) & 0xF active.DeviceEndpoint.ICPInfo[0].Volume = fmt.Sprintf("%d", icpBit1) active.DeviceEndpoint.ICPInfo[0].ID = "1" active.DeviceEndpoint.ICPInfo[1].Volume = fmt.Sprintf("%d", icpBit8) active.DeviceEndpoint.ICPInfo[1].ID = "8" //Get Pacu info eidsStat := info[1] //PACU mute pacuStat := info[4] //PACU status pacuVolume := binary.BigEndian.Uint32(info[5:9]) //info[5]-info[8] //lfshook.NewLogger().Logger.Infof("=eidsStat:%x=======pacuStat:%x=====", eidsStat, pacuStat) for i := 0; i < 8; i++ { eidsBit := (eidsStat >> i) & 0x01 // 右移 i 位,再与 1 取最低位 //fmt.Printf("eidsBit bit %d = %d\n", i, eidsBit) // bit 0 是 LSB(最低位,最右) //get 1 bit every time pacuStatBit := (pacuStat >> i) & 0x01 //fmt.Printf("pacuStatBit bit %d = %d\n", i, pacuStatBit) //get 4 bit every time var pacuVolBit int8 pacuVolBit = int8((pacuVolume >> i) & 0xF) //fmt.Printf("pacuVolBit bit %d = %x\n", i, pacuVolBit) active.DeviceEndpoint.PacuInfo[i].ID = strconv.Itoa(i + 1) if eidsBit == 1 { active.DeviceEndpoint.PacuInfo[i].Mute = true } else { active.DeviceEndpoint.PacuInfo[i].Mute = false } if pacuStatBit == 0 { active.DeviceEndpoint.PacuInfo[i].Status = "normal" } else { active.DeviceEndpoint.PacuInfo[i].Status = "abnormal" } active.DeviceEndpoint.PacuInfo[i].Volume = strconv.Itoa(int(pacuVolBit)) //lfshook.NewLogger().Logger.Infof("PACU INFO===ID:%s===Mute:%+v===Stat:%s===Vol:%s", active.DeviceEndpoint.PacuInfo[i].ID, active.DeviceEndpoint.PacuInfo[i].Mute, active.DeviceEndpoint.PacuInfo[i].Status, active.DeviceEndpoint.PacuInfo[i].Volume) } } // 处理单个数据包(原 switch 逻辑迁移过来) func processPacket(packet []byte) { if len(packet) < 6 { lfshook.NewLogger().Logger.Infof("Get data wrong length from STC !") return } //for recv data log debug if packet[5] != 0x03 && packet[5] != 0x0c && packet[5] != 0x01 { lfshook.NewLogger().Logger.Infof("Get data from STC:%x", packet) } //check if the cmd type is avtive if packet[5] == 0x03 { // ACTIVE Active([3]byte{packet[8], packet[9], packet[10]}) return } //check if Master role if !active.Master { if tagLog == 0 { lfshook.NewLogger().Logger.Infof("=========Not Master Role Ignore data !=============") tagLog = 1 } return } tagLog = 0 switch packet[5] { case 0x01: //heartbeat dataLen := binary.BigEndian.Uint16(packet[6:8]) if dataLen == 10 { processHeartbeat(packet[8:]) } else { //lfshook.NewLogger().Logger.Infof("=========Heartbeat data err !====len < 10=========") } case 0x02: // STN if active.ActivedCab != "" { if priority.CheckPriority("STN") { action.InterruptRunningTask("STN") //STN interrupt other time.Sleep(time.Millisecond * 100) //wait endpoimt release StationAnn(packet) } else { alstatus.PaStatus("", "STN", "refuse") } } case 0x05: // SPC if active.ActivedCab != "" { if priority.CheckPriority("SPC") { action.InterruptRunningTask("SPC") //SPC interrupt other time.Sleep(time.Millisecond * 100) //wait endpoimt release SpecialAnn(packet) } else { alstatus.PaStatus("", "SPC", "refuse") } } case 0x06: // EMG if active.ActivedCab != "" { if priority.CheckPriority("EMG") { action.InterruptRunningTask("EMG") //EMG interrupt other time.Sleep(time.Millisecond * 100) //wait endpoimt release EmgMsg(packet) } else { alstatus.PaStatus("", "EMG", "refuse") } } case 0x07: // STOP AnnStop([4]byte{packet[8], packet[9], packet[10], packet[11]}) case 0x08: // DCS if active.ActivedCab != "" { if priority.CheckPriority("DCS") { action.InterruptRunningTask("DCS") //DCS interrupt other time.Sleep(time.Millisecond * 100) //wait endpoimt release DcsAnn(packet) } else { alstatus.PaStatus("", "DCS", "refuse") } } case 0x09: // SELF CHECK if active.ActivedCab != "" { if priority.CheckPriority("CHK") { action.InterruptRunningTask("CHK") //CHK interrupt other time.Sleep(time.Millisecond * 100) //wait endpoimt release SelfCheck(packet) } else { alstatus.PaStatus("", "CHK", "refuse") } } case 0x0a: // Tone-test if active.ActivedCab != "" { if priority.CheckPriority("VOL") { action.InterruptRunningTask("VOL") //VOL interrupt other time.Sleep(time.Millisecond * 100) //wait endpoimt release ToneTest(packet) } else { alstatus.PaStatus("", "VOL", "refuse") } } case 0x0e: //TMS answer PAD if priority.CheckPriority("PAD-TMS") { //Before Answer PAD if packet[8] == 0x01 { action.InterruptRunningTask("PAD-TMS") //PAD-ICP interrupt other time.Sleep(time.Millisecond * 100) //wait endpoimt release } AlarmHandleTMS(packet) if active.QueueTimer != nil { if active.QueueTimer.Stop() { lfshook.NewLogger().Logger.Infof("=========Release PAD timer true !============") } else { lfshook.NewLogger().Logger.Infof("=========Release PAD timer false ! ============") } } } else { alstatus.PaStatus("", "PAD-TMS", "refuse") } case 0x0b: // reset all PAD AlarmHoldResetAll(packet[8]) // reset all pad case 0x0c: // Set PAD time oute PadTimeOutSetting(packet[8:]) // timeout setting case 0x0d: // ICP answer PAD if priority.CheckPriority("PAD-ICP") { //Before Answer PAD if packet[8] == 0x01 { action.InterruptRunningTask("PAD-ICP") //PAD-ICP interrupt other time.Sleep(time.Millisecond * 100) //wait endpoimt release } if active.QueueTimer != nil { if active.QueueTimer.Stop() { lfshook.NewLogger().Logger.Infof("=========Release PAD timer true !============") } else { lfshook.NewLogger().Logger.Infof("=========Release PAD timer false ! ============") } } AlarmHandleICP(packet) } else { alstatus.PaStatus("", "PAD-ICP", "refuse") } case 0xf1: //Set remote master //default: //fmt.Printf("Unknown command: %x\n", packet[5]) } } func PadTimeOutSetting(data []byte) { Seconds := data[7] if Seconds != 0 { active.PADTimeout = int(Seconds) //lfshook.NewLogger().Logger.Infof("=========Set PAD Timeout seconds to %d ! ============", active.PADTimeout) } } // STN , 自动报站广播 func StationAnn(data []byte) (err error) { specialVoice := int(data[8]) delay := data[9] cycleCount := data[10] datalen := int(data[11]) filename := msgdata.SubstrByRune(string(data[12:]), 0, datalen-4) //set spc voice tag priority.SpecialVoice = specialVoice action.PlaybackPacu(strconv.Quote(filename), int(cycleCount), int(delay), "STN") return nil } // 激活信号 func Active(data [3]byte) { //lfshook.NewLogger().Logger.Infof("===cab%s======Active Type %+v", active.CabNum, data) //check asterisk available /* if active.Master { // master true if !utils.CheckAsterisk() { //check asterisk not available and set master false if active.MasterTag > 2 { active.Master = false active.MasterTag = 0 } else { active.MasterTag++ } //utils.ExecCmd("/etc/init.d/asterisk", "stop", "PBX") } else { active.MasterTag = 0 } } else { // master false, check and start asterisk //Master == false if !utils.CheckAsterisk() { if active.CabNum == "8" { //utils.ExecCmd("/etc/init.d/asterisk", "start", "PBX") utils.ExecCmd("/usr/sbin/asterisk") } } //Master == false cab == 1 //if active.CabNum == "1" { // if utils.CheckAsterisk() { // utils.ExecCmd("/etc/init.d/asterisk", "stop", "PBX") // } //} } */ //var info model.Sysinfo Signal := int(data[0]) master := int(data[1]) TrainInfo := int(data[2]) lfshook.NewLogger().Logger.Infof("===========cab=%s Master=%d===========", active.CabNum, master) //Set Master if master == 1 && active.CabNum == "1" { active.Master = true lfshook.NewLogger().Logger.Infof("=====cab1 Master") if !utils.CheckAsterisk() { lfshook.NewLogger().Infof("Check asterisk , if not running , run cmd /usr/sbin/asterisk !") utils.ExecCmdAsync("/usr/sbin/asterisk") time.Sleep(3 * time.Second) } } else if master == 8 && active.CabNum == "8" { active.Master = true lfshook.NewLogger().Logger.Infof("=====cab8 Master") if !utils.CheckAsterisk() { lfshook.NewLogger().Infof("Check asterisk , if not running , run cmd /usr/sbin/asterisk !") utils.ExecCmdAsync("/usr/sbin/asterisk") time.Sleep(3 * time.Second) } } else if master == 8 && active.CabNum == "1" { lfshook.NewLogger().Logger.Infof("=====cab1 slave check stop asterisk") if utils.CheckAsterisk() { utils.ExecCmd("/etc/init.d/asterisk", "stop", "PBX") } } if TrainInfo != 0 { //Get train devide info DevideInfo := TrainInfo & 0x10 if DevideInfo == 1 { active.TrainDevide = 1 active.Master = true //列车断开,设置两边都Master if !utils.CheckAsterisk() { utils.ExecCmd("/usr/sbin/asterisk") } } else { active.TrainDevide = 0 } //Radio fault RadioFault1 := TrainInfo & 0x03 RadioFault8 := TrainInfo & 0x0c if RadioFault1 == 0 && RadioFault8 == 1 { active.RadioFault = 1 } else { active.RadioFault = 0 } } switch Signal { case 0: if active.ActivedCab != "" { active.ActivedCab = "" action.InActiveHangup() } case 1: //active signal from 8 to 1 if active.ActivedCab == "8" || active.ActivedCab == "" { active.ActivedCab = "1" action.InActiveHangup() } case 8: //active signal from 1 to 8 if active.ActivedCab == "1" || active.ActivedCab == "" { active.ActivedCab = "8" action.InActiveHangup() } } } // SPC ,特殊服务消息广播 func SpecialAnn(data []byte) { delay := data[8] cycleCount := data[9] datalen := int(data[10]) filename := msgdata.SubstrByRune(string(data[11:]), 0, datalen-4) if int(cycleCount) == 255 { action.PlaybackPacu(strconv.Quote(filename), 9999999, int(delay), "SPC") } else { action.PlaybackPacu(strconv.Quote(filename), int(cycleCount), int(delay), "SPC") } } // EMG ,紧急服务消息广播 func EmgMsg(data []byte) { delay := data[8] cycleCount := data[9] datalen := int(data[10]) filename := msgdata.SubstrByRune(string(data[11:]), 0, datalen-4) if int(cycleCount) == 255 { action.PlaybackPacu(strconv.Quote(filename), 9999999, int(delay), "EMG") } else { action.PlaybackPacu(strconv.Quote(filename), int(cycleCount), int(delay), "EMG") } } // 停止指定类型广播 func AnnStop(data [4]byte) { //lfshook.NewLogger().Logger.Infof("=========AnnStop Type %x", data[0]) switch data[0] { case 0x03: action.HangupTask("DCS") //STOP DCS time.Sleep(time.Millisecond * 100) //wait endpoimt release case 0x04: action.HangupTask("EMG") //STOP EMG time.Sleep(time.Millisecond * 100) //wait endpoimt release case 0x07: action.HangupTask("SPC") //STOP SPC time.Sleep(time.Millisecond * 100) //wait endpoimt release case 0x08: action.HangupTask("STN") //STOP STN time.Sleep(time.Millisecond * 100) //wait endpoimt release case 0x09: action.HangupTask("CHK") //STOP CHK time.Sleep(time.Millisecond * 100) //wait endpoimt release case 0x0a: action.HangupTask("VOL") //STOP VOL time.Sleep(time.Millisecond * 100) //wait endpoimt release default: action.InterruptRunningTask("") time.Sleep(time.Millisecond * 100) //wait endpoimt release } } // DCS 语音 func DcsAnn(data []byte) { delay := data[8] cycleCount := data[9] datalen := int(data[10]) filename := msgdata.SubstrByRune(string(data[11:]), 0, datalen-4) if int(cycleCount) == 255 { action.PlaybackPacu(strconv.Quote(filename), 9999999, int(delay), "DCS") } else { action.PlaybackPacu(strconv.Quote(filename), int(cycleCount), int(delay), "DCS") } } // tone-test广播 func ToneTest(data []byte) { check := data[8] delay := data[9] cycleCount := data[10] datalen := int(data[11]) filename := msgdata.SubstrByRune(string(data[12:]), 0, datalen-4) switch check { case 0x01: //start action.PlaybackPacu(strconv.Quote(filename), int(cycleCount), int(delay), "VOL") case 0x02: //stop action.HangupAllExcept("") } } // 自检广播 func SelfCheck(data []byte) { check := data[8] delay := data[9] cycleCount := data[10] //cycleCount := 0x32 datalen := int(data[11]) filename := msgdata.SubstrByRune(string(data[12:]), 0, datalen-4) switch check { case 0x01: //start action.PlaybackPacu(strconv.Quote(filename), int(cycleCount), int(delay), "CHK") case 0x02: //stop action.HangupAllExcept("") } } // 全局变量:记录正在抑制的 exten var ( suppressedExts = sync.Map{} // map[string]struct{},值存在即表示被抑制 //suppressionMu sync.Mutex // 保护初始化和清理操作(可选) ) // suppressKey 生成用于抑制的 key(可以根据需求扩展) func suppressKey(exten string, handler byte) string { return fmt.Sprintf("%s_h%x", exten, handler) } // ICP操作乘客报警(根据激活信息判断转到1车还是8车================) func AlarmHandleICP(data []byte) { handler := data[8] carr := data[12] pos := data[13] exten := fmt.Sprintf("24%c%c", carr, pos) key := suppressKey(exten, handler) //Drop other handler in 2 sec //PACUs---call---->ICP1 //PAD---->Chanspy(WEq)-->ICP1;PAD--->Call---->ICP2 if handler == 0x01 { if _, loaded := suppressedExts.LoadOrStore(key, struct{}{}); loaded { lfshook.NewLogger().Logger.Infof("Suppressed duplicate ICP Alarm (handler=0x01) for exten: %s within 4 seconds", exten) return } time.AfterFunc(4*time.Second, func() { suppressedExts.Delete(key) lfshook.NewLogger().Logger.Debugf("Suppression released for key: %s", key) }) } switch handler { case 0x01: //answer(ICP+Alarm+PACU) //NotifyPaiu(exten, "answer") priority.ICPAnswer = 1 if priority.PADStart == 0 { alstatus.PaStatus("", "PAD", "start") priority.PADStart = 1 } lfshook.NewLogger().Logger.Infof("================ ICP Answer PAD ================:%s ", exten) if active.ActivedCab == "1" { action.Dial("0402", "0511", "pad-rule-pacus", "ani1", exten, "1") // PACUs dial ICP1 //goto ami event ConfbridgeJoin, ICP answer PAD } else if active.ActivedCab == "8" { action.Dial("0402", "0511", "pad-rule-pacus", "ani8", exten, "8") // PACUs dial ICP8 //goto ami event ConfbridgeJoin, ICP answer PAD } else if active.ActivedCab == "" { // No cab occupied action.Dial("0402", "0511", "pad-rule-pacus", "ani1", exten, "1") // PACUs dial ICP1 } case 0x02: //hold 重新放回队列里面 active.NotifyPaiu(exten, "hold") err := action.RedirectInQueue(exten, "0300", "queues-icp-redirect", "1") if err != nil { lfshook.NewLogger().Info(err) } action.InterruptRunningTask("") time.Sleep(time.Millisecond * 100) //wait endpoimt release //action.HangupICP() case 0x03: //hangup //NotifyPaiu(exten, "hangup") lfshook.NewLogger().Logger.Infof("============ Hangup PAD-ICP =============== ") action.Hangup(exten) //Pad //action.HangupICP() action.HangupTask("PAD-ICP") } } // TMS操作乘客报警(根据激活信息判断转到1车还是8车================) func AlarmHandleTMS(data []byte) { handler := data[8] //extlen := data[9] carr := data[12] pos := data[13] exten := fmt.Sprintf("24%c%c", carr, pos) PacuNum := fmt.Sprintf("21%c1", carr) key := suppressKey(exten, handler) //Drop other handler in 2 sec // 只对 handler == 0x01 做 2 秒去重 if handler == 0x01 { if _, loaded := suppressedExts.LoadOrStore(key, struct{}{}); loaded { lfshook.NewLogger().Logger.Infof("Suppressed duplicate ICP Alarm (handler=0x01) for exten: %s within 4 seconds", exten) return // 已存在,说明在2秒窗口期内,直接丢弃 } // 设置4秒后删除该 key,允许下次通过 time.AfterFunc(4*time.Second, func() { suppressedExts.Delete(key) lfshook.NewLogger().Logger.Debugf("Suppression released for key: %s", key) }) } switch handler { case 0x01: //answer(ICP+Alarm+PACU) //PACU---call---->ICP1 //PAD---->Chanspy(WEq)-->ICP1;PAD--->Call---->ICP2 if priority.PADStart == 0 { alstatus.PaStatus("", "PAD", "start") priority.PADStart = 1 } priority.ICPAnswer = 1 lfshook.NewLogger().Logger.Infof("==============TMS Answer PAD Exten:%s PACU:%s==========", exten, PacuNum) if action.ExtenStatus(PacuNum) == "Idle" { if active.ActivedCab == "1" { action.Dial("0403", PacuNum, "default", PacuNum, exten, "1") // PACU dial ICP1 //goto ami event BridgeEnter, ICP8 whisper ICP1 } else if active.ActivedCab == "8" { action.Dial("0403", PacuNum, "default", PacuNum, exten, "8") // PACU dial ICP8 //goto ami event BridgeEnter, ICP1 whisper ICP8 } else if active.ActivedCab == "" { // No cab occupied action.Dial("0403", PacuNum, "default", PacuNum, exten, "1") // PACU dial ICP1 } } else { action.RedirectInQueue(exten, "0405", "default", exten) // PAD dial ICPs } case 0x02: //hold 重新放回队列里面 active.NotifyPaiu(exten, "hold") err := action.RedirectInQueue(exten, "0300", "queues-icp-redirect", "1") if err != nil { lfshook.NewLogger().Info(err) } action.HangupAllLocalChan() case 0x03: //hangup //NotifyPaiu(exten, "hangup") action.Hangup(exten) //Pad action.HangupTask("PAD-TMS") //action.HangupTask("PAD-ICP") } } // 挂断所有报警器 func AlarmHoldResetAll(handler byte) { //hangup all actived PAD action.HangupAllPAD() //hangup running task action.InterruptRunningTask("AlarmHoldResetAll") //Reset PAD ALL time.Sleep(time.Millisecond * 100) //wait endpoimt release }