crrc/internal/app/stc/broadcast/stc-broadcast.go

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) {

	//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])
	TrainInfo := int(data[2])

	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%c%c", carr, pos)

	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
}