MCPcopy Create free account
hub / github.com/FunAudioLLM/Fun-Audio-Chat / decode_message

Function decode_message

web_demo/server/protocal.py:40–110  ·  view source on GitHub ↗
(data: bytes)

Source from the content-addressed store, hash-verified

38
39
40def decode_message(data: bytes) -> dict:
41 if len(data) == 0:
42 raise ValueError("Empty message data")
43
44 msg_type = data[0]
45 payload = data[1:]
46
47 if msg_type == MessageType.HANDSHAKE:
48 version = payload[0] if len(payload) > 0 else 0
49 model = payload[1:] if len(payload) > 1 else 0
50 return {
51 'type': 'handshake',
52 'version': version,
53 'model': model
54 }
55
56 elif msg_type == MessageType.AUDIO:
57 return {
58 'type': 'audio',
59 'data': payload
60 }
61
62 elif msg_type == MessageType.TEXT:
63 return {
64 'type': 'text',
65 'data': payload.decode('utf-8')
66 }
67
68 elif msg_type == MessageType.COLOREDTEXT:
69 color = payload[0] if len(payload) > 0 else 0
70 text = payload[1:].decode('utf-8') if len(payload) > 1 else ""
71 return {
72 'type': 'coloredtext',
73 'color': color,
74 'data': text
75 }
76
77 elif msg_type == MessageType.CONTROL:
78 action = payload[0] if len(payload) > 0 else 0
79 action_names = {
80 ControlMessage.START: 'start',
81 ControlMessage.END_TURN: 'endTurn',
82 ControlMessage.PAUSE: 'pause',
83 ControlMessage.RESTART: 'restart'
84 }
85 return {
86 'type': 'control',
87 'action': action_names.get(action, 'unknown')
88 }
89
90 elif msg_type == MessageType.METADATA:
91 import json
92 metadata = json.loads(payload.decode('utf-8'))
93 return {
94 'type': 'metadata',
95 'data': metadata
96 }
97

Callers 1

recv_loopMethod · 0.90

Calls 1

decodeMethod · 0.80

Tested by

no test coverage detected