Implement basic CslaBinaryWriter

csla_binary/__init__.py

@@ -15,3 +15,5 @@
 #   along with this program.  If not, see <https://www.gnu.org/licenses/>.
 
 from .binary_reader import CslaBinaryReader
+from .binary_writer import CslaBinaryWriter
+from .serialization_info import ChildData, FieldData, SerializationInfo

csla_binary/binary_reader.py

@@ -17,12 +17,13 @@
 import io
 import struct
 
+from .known_types import CslaKnownTypes
 from .serialization_info import ChildData, FieldData, SerializationInfo
 
 class CslaBinaryReader:
 	"""Reads binary-serialised CSLA data into SerializationInfo objects"""
 	
-	def __init__(self, stream: io.BytesIO):
+	def __init__(self, stream: io.BufferedIOBase):
 		self.stream = stream
 		self.keywords_dictionary = {}
 	
@@ -109,79 +110,79 @@ class CslaBinaryReader:
 		# CslaBinaryReader.ReadObject
 		known_type = self.stream.read(1)[0]
 		
-		if known_type == 1:  # CslaKnownTypes.Boolean
+		if known_type == CslaKnownTypes.Boolean.value:
 			raise NotImplementedError()
 		
-		if known_type == 2:  # CslaKnownTypes.Char
+		if known_type == CslaKnownTypes.Char.value:
 			raise NotImplementedError()
 		
-		if known_type == 3:  # CslaKnownTypes.SByte
+		if known_type == CslaKnownTypes.SByte.value:
 			raise NotImplementedError()
 		
-		if known_type == 4:  # CslaKnownTypes.Byte
+		if known_type == CslaKnownTypes.Byte.value:
 			raise NotImplementedError()
 		
-		if known_type == 5:  # CslaKnownTypes.Int16
+		if known_type == CslaKnownTypes.Int16.value:
 			raise NotImplementedError()
 		
-		if known_type == 6:  # CslaKnownTypes.UInt16
+		if known_type == CslaKnownTypes.UInt16.value:
 			raise NotImplementedError()
 		
-		if known_type == 7:  # CslaKnownTypes.Int32
+		if known_type == CslaKnownTypes.Int32.value:
 			raise NotImplementedError()
 		
-		if known_type == 8:  # CslaKnownTypes.UInt32
+		if known_type == CslaKnownTypes.UInt32.value:
 			raise NotImplementedError()
 		
-		if known_type == 9:  # CslaKnownTypes.Int64
+		if known_type == CslaKnownTypes.Int64.value:
 			raise NotImplementedError()
 		
-		if known_type == 10:  # CslaKnownTypes.UInt64
+		if known_type == CslaKnownTypes.UInt16.value:
 			raise NotImplementedError()
 		
-		if known_type == 11:  # CslaKnownTypes.Single
+		if known_type == CslaKnownTypes.Single.value:
 			raise NotImplementedError()
 		
-		if known_type == 12:  # CslaKnownTypes.Double
+		if known_type == CslaKnownTypes.Double.value:
 			raise NotImplementedError()
 		
-		if known_type == 13:  # CslaKnownTypes.Decimal
+		if known_type == CslaKnownTypes.Decimal.value:
 			raise NotImplementedError()
 		
-		if known_type == 14:  # CslaKnownTypes.DateTime
+		if known_type == CslaKnownTypes.DateTime.value:
 			raise NotImplementedError()
 		
-		if known_type == 15:  # CslaKnownTypes.String
+		if known_type == CslaKnownTypes.String.value:
 			return self.read_string()
 		
-		if known_type == 16:  # CslaKnownTypes.TimeSpan
+		if known_type == CslaKnownTypes.TimeSpan.value:
 			raise NotImplementedError()
 		
-		if known_type == 17:  # CslaKnownTypes.DateTimeOffset
+		if known_type == CslaKnownTypes.DateTimeOffset.value:
 			raise NotImplementedError()
 		
-		if known_type == 18:  # CslaKnownTypes.Guid
+		if known_type == CslaKnownTypes.Guid.value:
 			raise NotImplementedError()
 		
-		if known_type == 19:  # CslaKnownTypes.ByteArray
+		if known_type == CslaKnownTypes.ByteArray.value:
 			raise NotImplementedError()
 		
-		if known_type == 20:  # CslaKnownTypes.CharArray
+		if known_type == CslaKnownTypes.CharArray.value:
 			raise NotImplementedError()
 		
-		if known_type == 21:  # CslaKnownTypes.ListOfInt
+		if known_type == CslaKnownTypes.ListOfInt.value:
 			raise NotImplementedError()
 		
-		if known_type == 22:  # CslaKnownTypes.Null
+		if known_type == CslaKnownTypes.Null.value:
 			raise NotImplementedError()
 		
-		if known_type == 23:  # CslaKnownTypes.StringWithDictionaryKey
+		if known_type == CslaKnownTypes.StringWithDictionaryKey.value:
 			system_string = self.read_string()
 			dictionary_key = self.read_int32()
 			self.keywords_dictionary[dictionary_key] = system_string
 			return system_string
 		
-		if known_type == 24:  # CslaKnownTypes.StringDictionaryKey
+		if known_type == CslaKnownTypes.StringDictionaryKey.value:
 			raise NotImplementedError()
 		
 		raise ValueError('Unexpected object tag {}'.format(known_type))

csla_binary/binary_writer.py

@@ -0,0 +1,111 @@
+#   pycsla-binary: Python implementation of CSLA .NET binary serialisation
+#   Copyright (C) 2025  Lee Yingtong Li (RunasSudo)
+#
+#   This program is free software: you can redistribute it and/or modify
+#   it under the terms of the GNU Affero General Public License as published by
+#   the Free Software Foundation, either version 3 of the License, or
+#   (at your option) any later version.
+#
+#   This program is distributed in the hope that it will be useful,
+#   but WITHOUT ANY WARRANTY; without even the implied warranty of
+#   MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
+#   GNU Affero General Public License for more details.
+#
+#   You should have received a copy of the GNU Affero General Public License
+#   along with this program.  If not, see <https://www.gnu.org/licenses/>.
+
+import io
+import struct
+from typing import List
+
+from .known_types import CslaKnownTypes
+from .serialization_info import SerializationInfo
+
+class CslaBinaryWriter:
+	"""Writes SerializationInfo objects into binary-serialised CSLA data"""
+	
+	def __init__(self, stream: io.BufferedIOBase):
+		self.stream = stream
+		self.keywords_dictionary = []
+	
+	def write(self, serialisation_infos: List[SerializationInfo]):
+		# Reverse of CslaBinaryReader.Read
+		self.write_int32(len(serialisation_infos))
+		
+		for info in serialisation_infos:
+			# Write ReferenceID
+			self.write_int32(info.reference_id)
+			
+			# Write TypeName
+			self.write_object_system_string(info.type_name)
+			
+			# Write children
+			self.write_int32(len(info.children))
+			for child in info.children:
+				self.write_object_system_string(child.name)
+				self.write_object_bool(child.is_dirty)
+				self.write_object_int32(child.reference_id)
+			
+			# Write field values
+			self.write_int32(len(info.values))
+			for value in info.values:
+				self.write_object_system_string(value.name)
+				self.write_object_system_string(value.enum_type_name or '')
+				self.write_object_bool(value.is_dirty)
+				self.write_object(value.value)
+	
+	def write_7bit_encoded_int(self, value):
+		# BinaryWriter.Write7BitEncodedInt
+		# "The integer of the value parameter is written out seven bits at a time, starting with the seven least-significant bits. The high bit of a byte indicates whether there are more bytes to be written after this one."
+		while True:
+			value_7lsb = value & 0b01111111
+			value >>= 7
+			
+			if value == 0:
+				# Final byte
+				self.stream.write(bytes([value_7lsb]))
+				return
+			else:
+				# Further bytes remaining
+				self.stream.write(bytes([value_7lsb | 0b10000000]))
+	
+	def write_int32(self, value):
+		# BinaryWriter.WriteInt32
+		self.stream.write(struct.pack('<i', value))
+	
+	def write_string(self, value):
+		# BinaryWriter.WriteString - "The string is prefixed with the length, encoded as an integer seven bits at a time."
+		encoded_string = value.encode('utf-8')
+		
+		self.write_7bit_encoded_int(len(encoded_string))
+		self.stream.write(encoded_string)
+	
+	def write_object(self, value):
+		# CslaBinaryWriter.Write(...)
+		raise NotImplementedError('Writing objects of dynamic type is not yet implemented')
+	
+	def write_object_bool(self, value):
+		# CslaBinaryWriter.Write(bool)
+		self.stream.write(bytes([CslaKnownTypes.Boolean.value, 1 if value else 0]))
+	
+	def write_object_int32(self, value):
+		# CslaBinaryWriter.Write(int)
+		self.stream.write(bytes([CslaKnownTypes.Int32.value]))
+		self.write_int32(value)
+	
+	def write_object_system_string(self, value):
+		# CslaBinaryWriter.WriteSystemString
+		if value in self.keywords_dictionary:
+			# Dictionary key already exists
+			dictionary_key = self.keywords_dictionary.index(value)
+			
+			self.stream.write(bytes([CslaKnownTypes.StringDictionaryKey.value]))
+			self.write_int32(dictionary_key)
+		else:
+			# New dictionary key
+			dictionary_key = len(self.keywords_dictionary)
+			self.keywords_dictionary.append(value)
+			
+			self.stream.write(bytes([CslaKnownTypes.StringWithDictionaryKey.value]))
+			self.write_string(value)
+			self.write_int32(dictionary_key)

csla_binary/known_types.py

@@ -0,0 +1,43 @@
+#   pycsla-binary: Python implementation of CSLA .NET binary serialisation
+#   Copyright (C) 2025  Lee Yingtong Li (RunasSudo)
+#
+#   This program is free software: you can redistribute it and/or modify
+#   it under the terms of the GNU Affero General Public License as published by
+#   the Free Software Foundation, either version 3 of the License, or
+#   (at your option) any later version.
+#
+#   This program is distributed in the hope that it will be useful,
+#   but WITHOUT ANY WARRANTY; without even the implied warranty of
+#   MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
+#   GNU Affero General Public License for more details.
+#
+#   You should have received a copy of the GNU Affero General Public License
+#   along with this program.  If not, see <https://www.gnu.org/licenses/>.
+
+from enum import Enum
+
+class CslaKnownTypes(Enum):
+	Boolean = 1
+	Char = 2
+	SByte = 3
+	Byte = 4
+	Int16 = 5
+	UInt16 = 6
+	Int32 = 7
+	UInt32 = 8
+	Int64 = 9
+	UInt64 = 10
+	Single = 11
+	Double = 12
+	Decimal = 13
+	DateTime = 14
+	String = 15
+	TimeSpan = 16
+	DateTimeOffset = 17
+	Guid = 18
+	ByteArray = 19
+	CharArray = 20
+	ListOfInt = 21
+	Null = 22
+	StringWithDictionaryKey = 23
+	StringDictionaryKey = 24