Quality of life improvements
Add helper functions to directly read/write bytes
This commit is contained in:
parent
98bcd88cb4
commit
5801539bb0
@ -16,6 +16,7 @@
|
|||||||
|
|
||||||
import io
|
import io
|
||||||
import struct
|
import struct
|
||||||
|
from typing import List
|
||||||
|
|
||||||
from .known_types import CslaKnownTypes
|
from .known_types import CslaKnownTypes
|
||||||
from .serialization_info import ChildData, FieldData, SerializationInfo
|
from .serialization_info import ChildData, FieldData, SerializationInfo
|
||||||
@ -23,11 +24,20 @@ from .serialization_info import ChildData, FieldData, SerializationInfo
|
|||||||
class CslaBinaryReader:
|
class CslaBinaryReader:
|
||||||
"""Reads binary-serialised CSLA data into SerializationInfo objects"""
|
"""Reads binary-serialised CSLA data into SerializationInfo objects"""
|
||||||
|
|
||||||
|
@classmethod
|
||||||
|
def read_from_bytes(cls, data: bytes) -> List[SerializationInfo]:
|
||||||
|
stream = io.BytesIO(data)
|
||||||
|
reader = cls(stream)
|
||||||
|
return reader.read()
|
||||||
|
|
||||||
|
# --------------
|
||||||
|
# Implementation
|
||||||
|
|
||||||
def __init__(self, stream: io.BufferedIOBase):
|
def __init__(self, stream: io.BufferedIOBase):
|
||||||
self.stream = stream
|
self.stream = stream
|
||||||
self.keywords_dictionary = {}
|
self.keywords_dictionary = {}
|
||||||
|
|
||||||
def read(self):
|
def read(self) -> List[SerializationInfo]:
|
||||||
# CslaBinaryReader.Read
|
# CslaBinaryReader.Read
|
||||||
total_count = self.read_int32()
|
total_count = self.read_int32()
|
||||||
|
|
||||||
|
@ -24,6 +24,16 @@ from .serialization_info import SerializationInfo
|
|||||||
class CslaBinaryWriter:
|
class CslaBinaryWriter:
|
||||||
"""Writes SerializationInfo objects into binary-serialised CSLA data"""
|
"""Writes SerializationInfo objects into binary-serialised CSLA data"""
|
||||||
|
|
||||||
|
@classmethod
|
||||||
|
def write_to_bytes(cls, serialisation_infos: List[SerializationInfo]) -> bytes:
|
||||||
|
stream = io.BytesIO()
|
||||||
|
writer = cls(stream)
|
||||||
|
writer.write(serialisation_infos)
|
||||||
|
return stream.getvalue()
|
||||||
|
|
||||||
|
# --------------
|
||||||
|
# Implementation
|
||||||
|
|
||||||
def __init__(self, stream: io.BufferedIOBase):
|
def __init__(self, stream: io.BufferedIOBase):
|
||||||
self.stream = stream
|
self.stream = stream
|
||||||
self.keywords_dictionary = []
|
self.keywords_dictionary = []
|
||||||
|
@ -20,23 +20,23 @@ from typing import Any, List
|
|||||||
@dataclass
|
@dataclass
|
||||||
class ChildData:
|
class ChildData:
|
||||||
# SerializationInfo.ChildData
|
# SerializationInfo.ChildData
|
||||||
name: str
|
name: str = ''
|
||||||
is_dirty: bool
|
is_dirty: bool = False
|
||||||
reference_id: int
|
reference_id: int = -1
|
||||||
|
|
||||||
@dataclass
|
@dataclass
|
||||||
class FieldData:
|
class FieldData:
|
||||||
# SerializationInfo.FieldData
|
# SerializationInfo.FieldData
|
||||||
name: str
|
name: str = ''
|
||||||
enum_type_name: str
|
enum_type_name: str = None
|
||||||
is_dirty: bool
|
is_dirty: bool = False
|
||||||
value: Any
|
value: Any = None
|
||||||
|
|
||||||
@dataclass
|
@dataclass
|
||||||
class SerializationInfo:
|
class SerializationInfo:
|
||||||
"""Format-agnostic representation of serialised CSLA object"""
|
"""Format-agnostic representation of serialised CSLA object"""
|
||||||
|
|
||||||
reference_id: int = 0
|
reference_id: int = -1
|
||||||
type_name: str = ''
|
type_name: str = ''
|
||||||
children: List[ChildData] = field(default_factory=list)
|
children: List[ChildData] = field(default_factory=list)
|
||||||
values: List[FieldData] = field(default_factory=list)
|
values: List[FieldData] = field(default_factory=list)
|
||||||
|
Loading…
x
Reference in New Issue
Block a user