(
self, text: Union[bytes, bytearray, str], node_id: str, sid=None
)
| 1433 | return json_data |
| 1434 | |
| 1435 | def send_progress_text( |
| 1436 | self, text: Union[bytes, bytearray, str], node_id: str, sid=None |
| 1437 | ): |
| 1438 | if isinstance(text, str): |
| 1439 | text = text.encode("utf-8") |
| 1440 | node_id_bytes = str(node_id).encode("utf-8") |
| 1441 | |
| 1442 | # Pack the node_id length as a 4-byte unsigned integer, followed by the node_id bytes |
| 1443 | message = struct.pack(">I", len(node_id_bytes)) + node_id_bytes + text |
| 1444 | |
| 1445 | self.send_sync(BinaryEventTypes.TEXT, message, sid) |
no test coverage detected