VoxelServer/ServerCore/JSONStreamParser.py

60 lines
1.5 KiB
Python

#!/usr/bin/env python3
from collections import deque
import json
class JSONStreamParser(object):
def __init__(self):
# initialize empty queue object
self.q = deque()
self.streambuffer = ""
self.bracket_depth = 0
def __newObjectFromBuffer(self):
assert len(self.streambuffer) > 0
self.q.append(self.streambuffer)
self.streambuffer = ""
def appendUTF8Data(self, data: str):
counter = 0
for c in data:
counter += 1
if c == "{":
self.bracket_depth += 1
elif c == "}":
self.bracket_depth -= 1
assert self.bracket_depth >= 0
if self.bracket_depth == 0:
self.streambuffer += data[:counter]
# build new json object from buffer
self.__newObjectFromBuffer()
# recursive call on remaining data
self.appendUTF8Data(data[counter:])
return
self.streambuffer += data
def appendRawData(self, rawData: bytes):
# decode raw bytes to utf-8
data = rawData.decode("utf-8")
self.appendUTF8Data(data)
def isQueueEmpty(self):
return len(self.q) == 0
def popAsJSON(self):
if self.isQueueEmpty():
return None
return self.q.popleft()
def popAsDictionary(self):
if self.isQueueEmpty():
return None
return json.loads(self.q.popleft())