alexander/bproto
1
0
Fork 0
Code Issues Pull Requests Packages Projects Releases Wiki Activity

Init Commit: Moved bproto to seperate repo

Browse Source
This commit is contained in:
AlexanderHD27
2025-04-14 14:43:03 +02:00
commit 45bfc724fc
125 changed files with 10822 additions and 0 deletions
Download Patch File Download Diff File Expand all files Collapse all files

91
src/protocol_components/__init__.py Normal file
View File

@@ -0,0 +1,91 @@
from abc import ABC, abstractmethod
from collections import OrderedDict
from nameHandling.base import ComponentName
class AbstractProtocolComponent(ABC):
@abstractmethod
def get_identifier(self) -> int | ComponentName:
"""Get the identifier of the protocol component.
This identifier should be unique and is ment for sorting and indexing.
Returns:
int | str: Identifier of the protocol component
"""
pass
@abstractmethod
def get_name(self) -> ComponentName:
"""Get the display name of the protocol component.
This is ment to display error messages, render into code, etc.
Should be human readable and unique.
Returns:
str: Display name of the protocol component
"""
pass
@abstractmethod
def get_type_name(self) -> str:
"""Get the type name of the protocol component.
This is ment to display error messages.
Should be human readable and describe the type of the component.
Returns:
str: Type name of the protocol component
"""
pass
@abstractmethod
def get_size_bytes(self) -> int:
"""Get the size of the protocol component in bytes.
Should be the same as ceil(get_bitsize() / 8).
Returns:
int: Size of the protocol component in bytes
"""
pass
@abstractmethod
def get_size_bits(self) -> int:
"""Get the size of the protocol component in bits.
Should be the same as get_byte_size() * 8.
Returns:
int: Size of the protocol component in bits
"""
pass
def sort_dict_by_value(d: dict[str, int]) -> dict[str, int]:
"""Sort a dictionary by its values.
Values should be something sortable.
Args:
d (dict[str, int]): dictionary to be sorted
Returns:
dict[str,int]: Sorted dictionary
"""
return OrderedDict(sorted(d.items(), key=lambda x: x[1]))
def find_smallset_next_value(s: set[int], inital=0) -> int:
"""Find the next smallest value that is not in the set.
Used to fill in missing values in an enumeration or bitfield.
Args:
s (set[int]): Set of values
inital (int, optional): Start value to search from. Defaults to 0. For optimization purposes.
Returns:
int: Next smallest value not in the set
"""
i = inital
while i in s:
i += 1
return i

167
src/protocol_components/bitfields.py Normal file
View File

@@ -0,0 +1,167 @@
from . import AbstractProtocolComponent, find_smallset_next_value, sort_dict_by_value
from errors import BprotoDuplicateNameError, BprotoDuplicateBitfieldPositionError
from nameHandling.base import ComponentName, NameStyleBproto
import math
from copy import deepcopy
class Bitfield(AbstractProtocolComponent):
"""Represents a bitfield in a protocol.
This should not be instantiated directly, but rather through the FactoryBitfield class.
Inherit from AbstractProtocolComponent.
"""
def __init__(self, name: ComponentName, inline: bool = False):
"""Initialize a Bitfield object; should not be used directly.
This methode should only be called by the FactoryBitfield class.
Args:
name (ComponentName): Name of the bitfield, should be unique
length (int): Length of the bitfield in bits.
inline: If True mark this bitfield as a inline definition
"""
self.name: ComponentName = name
self.length: int = 0
self.bits: dict[ComponentName, int] = {}
self.inline: bool = inline
# Inherited from AbstractProtocolComponent
def get_identifier(self):
return self.name
def get_name(self):
return self.name
def get_type_name(self):
return "bitfield"
def get_size_bits(self):
return self.length
def get_size_bytes(self):
return max(math.ceil(self.get_size_bits() / 8), 1)
# Methodes specific to this class
def apply_naming(self, naming: str):
"""Apply a naming scheme to the bitfield.
TODO: Is this method needed? REFACTOR!!!
"""
self.name = f"{naming}_{self.name}"
def __repr__(self):
field_names = ", ".join([f"{name}:{pos}" for name, pos in self.bits.items()])
return f"Bitfield({self.name}, {field_names})"
def __deepcopy__(self, memo={}):
if id(self) in memo:
return memo[id(self)]
new_bitfield = Bitfield(
deepcopy(self.name, memo),
deepcopy(self.inline, memo)
)
new_bitfield.length = deepcopy(self.length, memo)
new_bitfield.bits = {
deepcopy(k, memo): deepcopy(v, memo)
for k, v in self.bits.items()
}
memo[id(self)] = new_bitfield
memo[id(new_bitfield)] = new_bitfield
return new_bitfield
class FactoryBitfield:
"""Factory class to create Bitfield objects.
Used to assemble a Bitfield object from its parts.
This class should be used to create Bitfield objects.
Its a one-time use class, so create a new instance for each Bitfield object.
This Factory is ment to be used during traversal of the abstract syntax tree.
Simular to the FactoryEnumeration class.
"""
def __init__(self):
self.bits_list: list[tuple[str, int | None]] = []
def add_bit(self, name: str, position: int | None):
"""Add a bit to the bitfield.
Used to build the Bitfield object.
Duplication checks are not done here, but in the assemble
Args:
name (str): Name of the bit
position (int | None): Position of the bit in the bitfield. If None, the position will be filled in later.
"""
self.bits_list.append((NameStyleBproto.fromStr(name), position))
@staticmethod
def calculate_bitsize(bits: dict[int, str]):
"""Given a dictionary of bit names and their positions, calculate the bitsize of the bitfield.
Its just a wrapper around the max methode to be honest.
Args:
bits (dict[int, str]): Dictionary of bit names and their positions
Returns:
int: Number of bits needed to represent the bit
"""
return max(bits.values(), default=0) + 1
def assemble(self, name: str, inline: bool = False):
"""Finalize the Bitfield object and return it.
This method should be called after adding all bits to the bitfield.
This will:
1. Check for duplicate names and positions
2. Fill in missing positions
3. Calculate the bitsize of the bitfield
4. Sort the dictionary by value
(Very simular to the FactoryEnumeration.assemble method)
Args:
name (str): Name of the bitfield, should be unique
inline (bool): If true marks the resulting bitfield as inline
Raises:
BprotoDuplicateBitfieldPositionError: Thrown when a bitfield position is duplicated
BprotoDuplicateNameError: Thrown when a bitfield name is duplicated
Returns:
_type_: The finalized Bitfield object
"""
resulting_bitfield = Bitfield(NameStyleBproto.fromStr(name), inline=inline)
bits: dict[ComponentName, int | None] = {}
bit_names: set[ComponentName] = set()
bit_positions = set()
# Check for duplicates in name or value
for name, pos in self.bits_list:
if pos is not None and pos in bit_positions:
raise BprotoDuplicateBitfieldPositionError(pos, resulting_bitfield)
if name in bit_names:
raise BprotoDuplicateNameError(name, resulting_bitfield)
if pos is not None:
bit_positions.add(pos)
bits[name] = pos
bit_names.add(name)
# Fill in missing values
for name, pos in self.bits_list:
if pos is None:
pos = find_smallset_next_value(bit_positions)
bit_positions.add(pos)
bits[name] = pos
# Sort the dictionary by value
resulting_bitfield.length = FactoryBitfield.calculate_bitsize(bits)
resulting_bitfield.bits = sort_dict_by_value(bits)
return resulting_bitfield

22
src/protocol_components/crc.py Normal file
View File

@@ -0,0 +1,22 @@
CRC_SIZE = 2
def crc16_calc_inital(data: bytearray) -> int:
"""CRC-16/CITT-FALSE
from https://stackoverflow.com/questions/35205702/calculating-crc16-in-python
Args:
data (bytearray): protocol hash
Returns:
bytes: initals for crc16_calc_with_initial
"""
crc = 0xFFFF
for i in range(0, len(data)):
crc ^= data[i] << 8
for j in range(0, 8):
if (crc & 0x8000) > 0:
crc = (crc << 1) ^ 0x1021
else:
crc = crc << 1
return crc & 0xFFFF

153
src/protocol_components/dtypes.py Normal file
View File

@@ -0,0 +1,153 @@
import enum
from . import AbstractProtocolComponent
from errors import BprotoFrontendError
class BprotoFieldBaseType(enum.Enum):
BOOL = "bool"
CHAR = "char"
STRING = "string"
FLOAT32 = "float32"
FLOAT64 = "float64"
INT8 = "int8"
INT16 = "int16"
INT32 = "int32"
INT64 = "int64"
UINT8 = "uint8"
UINT16 = "uint16"
UINT32 = "uint32"
UINT64 = "uint64"
ENUM = "enum"
BITFIELD = "bitfield"
class ArrayPolicy(enum.Enum):
"""Enum respresenting the options for policy, regarding if an datatype is an array or not.
Options: DISALLOW, ALLOW, FORCE
"""
DISALLOW = 0
ALLOW = 1
FORCE = 2
DTYPE_SIZE_MAP = {
BprotoFieldBaseType.UINT8: 1,
BprotoFieldBaseType.UINT16: 2,
BprotoFieldBaseType.UINT32: 4,
BprotoFieldBaseType.UINT64: 8,
BprotoFieldBaseType.INT8: 1,
BprotoFieldBaseType.INT16: 2,
BprotoFieldBaseType.INT32: 4,
BprotoFieldBaseType.INT64: 8,
BprotoFieldBaseType.FLOAT32: 4,
BprotoFieldBaseType.FLOAT64: 8,
BprotoFieldBaseType.BOOL: 1,
BprotoFieldBaseType.CHAR: 1,
BprotoFieldBaseType.STRING: 1,
}
DTYPE_ARRAY_POLICY_MAP = {
BprotoFieldBaseType.UINT8: ArrayPolicy.ALLOW,
BprotoFieldBaseType.UINT16: ArrayPolicy.ALLOW,
BprotoFieldBaseType.UINT32: ArrayPolicy.ALLOW,
BprotoFieldBaseType.UINT64: ArrayPolicy.ALLOW,
BprotoFieldBaseType.INT8: ArrayPolicy.ALLOW,
BprotoFieldBaseType.INT16: ArrayPolicy.ALLOW,
BprotoFieldBaseType.INT32: ArrayPolicy.ALLOW,
BprotoFieldBaseType.INT64: ArrayPolicy.ALLOW,
BprotoFieldBaseType.FLOAT32: ArrayPolicy.ALLOW,
BprotoFieldBaseType.FLOAT64: ArrayPolicy.ALLOW,
BprotoFieldBaseType.BOOL: ArrayPolicy.ALLOW,
BprotoFieldBaseType.CHAR: ArrayPolicy.ALLOW,
BprotoFieldBaseType.STRING: ArrayPolicy.FORCE,
BprotoFieldBaseType.ENUM: ArrayPolicy.DISALLOW,
BprotoFieldBaseType.BITFIELD: ArrayPolicy.DISALLOW,
}
def validate_datatype(dtype: str, array_size: int, field: AbstractProtocolComponent) -> tuple[BprotoFieldBaseType, int]:
"""Given a string and array size, validate the datatype and array size.
Valid datatypes are defined in BprotoFieldBaseType enum.
Array size is validated based on the datatype and the array policy.
Args:
dtype (str): string representation of the datatype
array_size (int): size of the array
field (AbstractProtocolComponent): Datatypes parent field, for error reporting
Raises:
BprotoUnknownDataTypeError: Throws when the datatype is unknown
ValueError: Thrown when something went wrong, this should not happen, like an unimplemented array policy. This is not user error, but a developer error.
BprotoArrayPolicyViolationError: When array policy is violated: E.g. an Datatype is declared as an array, but the policy is disallowing it.
Returns:
tuple[BprotoFieldBaseType,int]: (datatype as enum, array size)
"""
if dtype not in BprotoFieldBaseType:
raise BprotoUnknownDataTypeError(dtype)
dtype: BprotoFieldBaseType = BprotoFieldBaseType(dtype)
array_policy = DTYPE_ARRAY_POLICY_MAP.get(dtype, None)
if array_policy is None:
raise ValueError("Internal Error: Array policy for this type was not found, fix this!")
match(array_policy):
case ArrayPolicy.ALLOW:
pass
case ArrayPolicy.DISALLOW:
if array_size > 1:
raise BprotoArrayPolicyViolationError(array_policy, dtype, field)
case ArrayPolicy.FORCE:
# This is fine, an array size of 1 is allowed
pass
case _:
raise ValueError("Internal Error: Array policy not implemented!")
return dtype, array_size
# Expections
class BprotoUnknownDataTypeError(BprotoFrontendError):
"""Error raise when building a field and the datatype is unknown.
This error should be use during the frontend compilation state.
Inherited from BprotoFrontendError.
"""
def __init__(self, data_type: str):
self.data_type = data_type
super().__init__(f"Unknown data type {data_type}!")
class BprotoArrayPolicyViolationError(BprotoFrontendError):
"""Error raise when building a field and the array policy is violated.
This error should be use during the frontend compilation state.
Inherited from BprotoFrontendError.
"""
def __init__(self, policy: ArrayPolicy, dtype: BprotoFieldBaseType, field: AbstractProtocolComponent):
self.policy = policy
self.field = field
match(policy):
case ArrayPolicy.ALLOW:
raise ValueError("Internal Error: Array policy ALLOW should not be violated!")
case ArrayPolicy.FORCE:
super().__init__(f"{field.get_name()} of type {dtype} must be an array!")
case ArrayPolicy.DISALLOW:
super().__init__(f"{field.get_name()} of type {dtype} must not be an array!")
case _:
raise ValueError("Internal Error: Array policy not implemented!")

232
src/protocol_components/enumeration.py Normal file
View File

@@ -0,0 +1,232 @@
from . import AbstractProtocolComponent, find_smallset_next_value, sort_dict_by_value
from .dtypes import DTYPE_SIZE_MAP, BprotoFieldBaseType
from errors import BprotoDuplicateNameError, BprotoDuplicateEnumValueError, BprotoEnumBitsizeTooLargeError
from nameHandling.base import ComponentName, NameStyleBproto
import math
from copy import deepcopy
class Enumeration(AbstractProtocolComponent):
"""Represents an enumeration in a protocol.
This should not be instantiated directly, but rather through the FactoryEnumeration class.
Inherit from AbstractProtocolComponent.
"""
def __init__(self, name: ComponentName, bitsize: int = 0, inline: bool = False):
"""Initialize an Enumeration object; should not be used directly.
This methode should only be called by the FactoryEnumeration class.
Args:
name (ComponentName): Name of the enumeration, should be unique
bitsize (int, optional): How many bits are need to represent the largest value? Defaults to 0.
inline (bool): If true marks this enum as inline defintion for later renaming
"""
self.name: ComponentName = name
self.values: dict[ComponentName, int] = {}
self.bitsize: int = bitsize
self.inline: bool = inline
def __deepcopy__(self, memo={}):
if id(self) in memo:
return memo[id(self)]
new_enum = Enumeration(
self.name,
self.bitsize,
self.inline
)
new_enum.name = deepcopy(self.name, memo)
new_enum.bitsize = deepcopy(self.bitsize, memo)
new_enum.inline = deepcopy(self.inline, memo)
new_enum.values = deepcopy(self.values, memo)
memo[id(self)] = new_enum
return new_enum
# Inherited from AbstractProtocolComponent
def get_identifier(self):
return self.name
def get_name(self):
return self.name
def get_type_name(self):
return "enum"
def get_size_bits(self):
"""Returns the bits used to represent the enumeration.
Returns:
int: Number of bits used to represent the enumeration
"""
return self.bitsize
def get_size_bytes(self):
"""Return the number of bytes needed to represent the enumeration.
Simular to get_bitsize, but returns the number of bytes.
Formula: ceil(bitsize / 8)
Returns:
int: Number of bytes needed to represent the enumeration
"""
return math.ceil(self.bitsize / 8)
# Methodes specific to this class
def apply_naming(self, naming: str):
"""TODO: Is this method needed? REFACTOR!!!
Args:
naming (str): _description_
"""
self.name = f"{naming}_{self.name}"
def map_auxiliary_datatypes(self) -> BprotoFieldBaseType:
"""Return the bproto datatype that best represents the enumeration.
There is a theoretical limit of 64 bits for an enumeration.
TODO: Do we need to handle this case? Probably...
Returns:
BprotoFieldBaseType: Datatype to represent the enumeration
"""
bytes_size = self.get_size_bytes()
if bytes_size <= 1:
return BprotoFieldBaseType.UINT8
elif bytes_size <= 2:
return BprotoFieldBaseType.UINT16
elif bytes_size <= 4:
return BprotoFieldBaseType.UINT32
elif bytes_size <= 8:
return BprotoFieldBaseType.UINT64
else:
raise BprotoEnumBitsizeTooLargeError(self)
def __repr__(self):
value_names = ", ".join([
f"{name}={val}"
for name, val in self.values.items()
])
return f"Enum({self.name}, {value_names})"
class FactoryEnumeration:
"""Factory class for creating Enumeration objects.
Used to assemble Enumerations from key-value pairs.
This is a one-time use factory. After calling assemble, the object should not be used anymore.
This Factory is ment to be used during traversale of the abstract syntax tree.
Simular to the FactoryBitfield class.
"""
def __init__(self):
self.key_value_pairs: list[tuple[NameStyleBproto, int]] = []
def add_value(self, name: str, value: int | None):
"""Add a key-value pair to the enumeration.
Used to build up the enumeration.
Duplication checks are not done here, but in the assemble
Args:
name (str): Name of the value
value (int | None): Value of the enumeration. If None, the value will be filled in later.
"""
self.key_value_pairs.append((NameStyleBproto.fromStr(name, "enum"), value))
@staticmethod
def calculate_bitsize(values: list[int]) -> int:
"""Given a list of unique values, calculate the bitsize needed to represent the largest value/ amount of values.
For empty lists, it will return 1.
Args:
values (list[int]): List of UNIQUE values, usually a list values of an enumeration
Returns:
int: Number of bits needed to represent the values
"""
if len(values) == 0:
return 1
max_value = max(values)
if max_value <= 8589934592:
return math.ceil(math.log2(max_value + 1))
else:
# This is a bit of a hack, but it works for now
# The problem is that the log2 function is not accurate enough for large numbers
# log2(2**64) = log2(2**64 - 1) = log2(2**64 + 1) = 64
# And this makes in this case a difference of 1 bit
if max_value <= 18446744073709551615:
return 64
return 128
def assemble(self, name: ComponentName, inline: bool = False) -> Enumeration:
"""Finalize the enumeration and create an Enumeration object.
It uses the key-value pairs added with add_values earlier.
This will:
1. Check for duplicates in name or value
2. Fill in missing values
3. Sort the dictionary by value
4. Calculate the bitsize of the enumeration and ensure it is within the protocol specs
After calling this method, the FactoryEnumeration object should not be used anymore.
Args:
name (ComponentName): Name of the enumeration, should be unique
inline (bool): if True marks this enum as aline
Raises:
BprotoDuplicateNameError: Error when a name of value is duplicated
BprotoDuplicateEnumValueError: Error when a value is duplicated
BprotoEnumBitsizeTooLargeError: Error when the bitsize of the enumeration is too large for the protocol specs
Returns:
Enumeration: Finished enumeration object to be used in the backend processing step
"""
enum_dict = {}
used_values = set()
resulting_enum = Enumeration(NameStyleBproto.fromStr(name), 0, inline=inline)
# Check for duplicates in name or value
for key, value in self.key_value_pairs:
if key in enum_dict:
raise BprotoDuplicateNameError(key, resulting_enum)
if value in used_values and value is not None:
raise BprotoDuplicateEnumValueError(value, resulting_enum)
if value is not None:
used_values.add(value)
enum_dict.update({key: value})
# Fill in missing values
search_start = 0
for key, value in enum_dict.items():
if value is None:
value = find_smallset_next_value(used_values, search_start)
search_start = value + 1
used_values.add(value)
enum_dict[key] = value
# Setting a few last properties
resulting_enum.values = sort_dict_by_value(enum_dict)
resulting_enum.bitsize = FactoryEnumeration.calculate_bitsize(list(used_values))
# Getting the best datatype to represent the enumeration
# This will raise an error if the enumeration is too large
dtype = resulting_enum.map_auxiliary_datatypes()
# Mapping datatype to its bitsize
resulting_enum.bitsize = DTYPE_SIZE_MAP[dtype] * 8
return resulting_enum

337
src/protocol_components/field.py Normal file
View File

@@ -0,0 +1,337 @@
from .dtypes import BprotoFieldBaseType, DTYPE_SIZE_MAP, validate_datatype
from .bitfields import Bitfield
from .enumeration import Enumeration
from . import AbstractProtocolComponent
from nameHandling.base import ComponentName, NameStyleBproto
from copy import deepcopy
class Field(AbstractProtocolComponent):
"""Represents a field in a protocol message.
This object should not be instantiated directly, but rather through the FactoryField class.
This is done during the frontend parsing of the protocol definition.
Extra classes exist for bitfields and enumerations / their references.
Inherit from AbstractProtocolComponent.
"""
def __init__(self,
name: ComponentName,
pos: int,
dtype: BprotoFieldBaseType,
array_size: int
):
"""Initialize a Field object; should not be used directly. Use the FactoryField class.
Args:
name (ComponentName): Name of the field, should be unique
pos (int): Position of the field in the message
dtype (BprotoFieldBaseType): Datatype of the field, this should already be validated
array_size (int): Size of the array, 1 for scalar fields
"""
self.name: ComponentName = name
self.pos = pos
self.type: BprotoFieldBaseType = dtype
self.array_size: int = array_size
self.ref = None
def resolve_reference(self,
enum_dict: dict[ComponentName, Enumeration],
bitfield_dict: dict[ComponentName, Bitfield]
) -> AbstractProtocolComponent:
"""_summary_
Args:
enum_dict (dict[ComponentName, Enumeration]): Dictoray
bitfield_dict (dict[ComponentName, Bitfield]): _description_
Returns:
AbstractProtocolComponent: Return the resolved reference object
"""
return self
# Inherited from AbstractProtocolComponent
def get_identifier(self) -> ComponentName:
return self.name
def get_name(self) -> ComponentName:
return self.name
def get_type_name(self) -> str:
return "field"
def get_size_bytes(self) -> int:
return DTYPE_SIZE_MAP[self.type] * self.array_size
def get_size_bits(self) -> int:
return DTYPE_SIZE_MAP[self.type] * self.array_size * 8
def get_base_size_bytes(self) -> int:
return DTYPE_SIZE_MAP[self.type]
# Dunder methods
def __repr__(self):
return f"{self.__class__.__name__}[{self.pos}]({self.name}, {self.type}, {self.array_size})" + (f" ref:{self.ref}" if self.ref is not None else "")
def __deepcopy__(self, memo):
if id(self) in memo:
return memo[id(self)]
new_field = Field(
deepcopy(self.name, memo),
deepcopy(self.pos, memo),
deepcopy(self.type, memo),
deepcopy(self.array_size, memo)
)
new_field.ref = deepcopy(self.ref, memo)
memo[id(self)] = new_field
return new_field
# Bitfield Fields
class FieldBitfield(Field):
"""Field object specifically for inline bitfields.
This object should not be instantiated directly, but rather through the FactoryField class.
Similar to FieldEnum, but for bitfields.
Inherit from Field and behaves like a Field object.
"""
def __init__(self, name: ComponentName, pos: int, bitfield: Bitfield):
"""Should not be used directly. Use the FactoryField class.
Args:
name (ComponentName): Name of the field, should be unique
pos (int): Position of the field in the message
bitfield (Bitfield): Bitfield object that this field represents
"""
super().__init__(name, pos, BprotoFieldBaseType.BITFIELD, 1)
self.bitfield: Bitfield = bitfield
def __deepcopy__(self, memo):
if id(self) in memo:
return memo[id(self)]
new_field = FieldBitfield(
deepcopy(self.name, memo),
deepcopy(self.pos, memo),
deepcopy(self.bitfield, memo)
)
new_field.type = deepcopy(self.type, memo)
new_field.array_size = deepcopy(self.array_size, memo)
memo[id(self)] = new_field
return new_field
def get_size_bytes(self) -> int:
return self.bitfield.get_size_bytes()
def get_size_bits(self):
return self.bitfield.get_bitsize()
def get_base_size_bytes(self) -> int:
return self.bitfield.get_size_bytes()
class FieldBitfieldRef(Field):
"""Field object specifically for bitfield references.
This object should not be instantiated directly, but rather through the FactoryField class.
The reference is the name of the bitfield that this field represents.
This refrence is not validated.
An object of this class should be eventually resolve to a FieldBitfield object.
Similar to FieldRefEnum, but for bitfield references.
Inherit from Field and behaves like a Field object.
"""
def __init__(self, name: ComponentName, pos: int, bitfield: ComponentName):
"""Should not be used directly. Use the FactoryField class.
Args:
name (ComponentName): Name of the field, should be unique
pos (int): Position of the field in the message
bitfield (ComponentName): Name of the bitfield that this field references
"""
super().__init__(name, pos, BprotoFieldBaseType.BITFIELD, 1)
self.ref: ComponentName = bitfield
def __deepcopy__(self, memo):
return super().__deepcopy__(memo)
def get_size_bytes(self):
return 0
def get_size_bits(self):
return 0
def get_base_size_bytes(self) -> int:
return 0
# Enum Fields
class FieldEnum(Field):
"""Field object specifically for inline enumerations.
This object should not be instantiated directly, but rather through the FactoryField class.
Similar to FieldBitfield, but for enumerations.
Inherit from Field and behaves like a Field object.
"""
def __init__(self, name: ComponentName, pos: int, enum: Enumeration):
"""Should not be used directly. Use the FactoryField class.
Args:
name (ComponentName): Name of the field, should be unique
pos (int): Position of the field in the message
enum (Enumeration): Enumeration object that this field represents
"""
super().__init__(name, pos, BprotoFieldBaseType.ENUM, 1)
self.enum: Enumeration = enum
def __deepcopy__(self, memo={}):
if id(self) in memo:
return memo[id(self)]
new_field = FieldEnum(
deepcopy(self.name, memo),
deepcopy(self.pos, memo),
deepcopy(self.enum, memo)
)
new_field.type = deepcopy(self.type, memo)
new_field.array_size = deepcopy(self.array_size, memo)
memo[id(self)] = new_field
return new_field
def get_size_bytes(self) -> int:
return self.enum.get_size_bytes()
def get_size_bits(self) -> int:
return self.enum.get_bitsize()
def get_base_size_bytes(self) -> int:
return self.enum.get_size_bytes()
class FieldEnumRef(Field):
"""Field object specifically for enumeration references.
This object should not be instantiated directly, but rather through the FactoryField class.
The reference is the name of the enumeration that this field represents.
This refrence is not validated.
An object of this class should be eventually resolve to a BprotoFieldEnum object.
Similar to FieldRefBitfield, but for enumerations.
Inherit from Field and behaves like a Field object.
"""
def __init__(self, name: ComponentName, pos: int, enum: ComponentName):
"""Should not be used directly. Use the FactoryField class.
Args:
name (ComponentName): Name of the field, should be unique
pos (int): Position of the field in the message
enum (ComponentName): Name of the enumeration that this field references
"""
super().__init__(name, pos, BprotoFieldBaseType.ENUM, 1)
self.ref: ComponentName = enum
def __deepcopy__(self, memo):
return super().__deepcopy__(memo)
def get_size_bytes(self):
return 0
def get_size_bits(self):
return 0
def get_base_size_bytes(self) -> int:
return 0
# Factory class
class FactoryField():
"""Factory class for building Field objects representing a bproto field.
This is ment to be used during the frontend compilation stage, during the traversal of the abstract syntax tree.
Its a one-time use class, so create a new instance for each Field object.
This class is compleatly statless (yet)
"""
def __init__(self):
"""Initialize the FactoryField object. Nothing to do here yet.
"""
pass
def assemble(self,
name: str,
pos: int,
dtype: str,
array_size: int,
ref: None | str | Bitfield | Enumeration) -> Field:
"""Finalize the field object and return it. Builds Enum and Bitfield Field objects if needed.
If the datatype is a bitfield or enumeration, the reference object should be provided.
The reference object should be a Bitfield or Enumeration object, or the name of the object.
Acorrding to the reference (and datatype), the correct specialized Field object is created.
Otherwise, the reference should be None.
If the datatype is a standard type, a Field object is created.
This methode will:
1. Validate the datatype and array size (array policy)
2. Create the correct Field object based on the datatype and reference
Args:
name (str): Name of the field, should be unique
pos (int): Position of the field in the message
dtype (str): Datatype of the field in string form
array_size (int): Size of the array, 1 for scalar fields
ref (None | str | Bitfield | Enumeration): For bitfields and enumerations, the reference object / reference name
Raises:
ValueError: Thrown when a reference type is provided that is neither a string nor a Bitfield or Enumeration object, or when an unimplemented array policy is encountered.
BprotoUnknownDataTypeError: Throws when the datatype is unknown
BprotoArrayPolicyViolationError: When array policy is violated: E.g. an Datatype is declared as an array, but the policy is disallowing it.
Returns:
Field: The finalized Field object, can also be a specialized Field object like FieldBitfield or BprotoFieldEnum
"""
converted_name = NameStyleBproto.fromStr(name)
converted_ref = NameStyleBproto.fromStr(ref) if isinstance(ref, str) else ref
resulting_field = Field(
converted_name,
pos,
dtype,
array_size
)
dtype, array_size = validate_datatype(dtype, array_size, resulting_field)
if dtype == BprotoFieldBaseType.BITFIELD:
if isinstance(ref, Bitfield):
resulting_field = FieldBitfield(converted_name, pos, ref)
elif isinstance(ref, str):
resulting_field = FieldBitfieldRef(converted_name, pos, converted_ref)
else:
raise ValueError("Internal Error: Invalid reference type for bitfield")
elif dtype == BprotoFieldBaseType.ENUM:
if isinstance(ref, Enumeration):
resulting_field = FieldEnum(converted_name, pos, ref)
elif isinstance(ref, str):
resulting_field = FieldEnumRef(converted_name, pos, converted_ref)
else:
raise ValueError("Internal Error: Invalid reference type for enum")
else:
resulting_field = Field(converted_name, pos, dtype, array_size)
return resulting_field

144
src/protocol_components/message.py Normal file
View File

@@ -0,0 +1,144 @@
from .field import Field
from . import AbstractProtocolComponent
from nameHandling.base import ComponentName, NameStyleBproto
from errors import BprotoDuplicateNameError, BprotoMessageIDAlreadyUsed
from copy import deepcopy
from collections import OrderedDict
class Message(AbstractProtocolComponent):
"""Representation of a bproto message.
Should not be instantiated directly, but rather through the FactoryMessage class.
Contains a dict of fields, where the key is the field name and the value is the Field object.
Inherit from AbstractProtocolComponent
"""
def __init__(self, name: ComponentName, index_number: int):
"""Should not be used directly. Use the FactoryMessage class.
Args:
name (str): Name of the message, should be unique
index_number (int): Index number of message, used for ordering, should be unique
"""
# field: {name: (type, array_size)}
self.name: ComponentName = name
self.message_index_number = index_number
self.fields: dict[ComponentName, Field] = {}
def __deepcopy__(self, memo={}):
if id(self) in memo:
return memo[id(self)]
new_message = Message(
deepcopy(self.name, memo),
deepcopy(self.message_index_number, memo)
)
new_message.fields = {
deepcopy(k, memo): deepcopy(v, memo)
for k, v in self.fields.items()
}
memo[id(self)] = new_message
return new_message
# Inherited from AbstractProtocolComponent
def get_identifier(self):
return self.message_index_number
def get_name(self):
return self.name
def get_type_name(self):
return "message"
def get_size_bytes(self) -> int:
return sum([field.get_size_bytes() for field in self.fields.values()])
def get_size_bits(self) -> int:
return sum([field.get_size_bits() for field in self.fields.values()])
# Methodes specific to this class
def apply_naming(self, naming: str):
"""Refactor this, should not be used.
Args:
naming (str): _description_
"""
self.name = f"{naming}_{self.name}"
class FactoryMessage():
"""Factory class for build Message objects representing a bproto message.
This is ment to be used during the frontend compilation stage, during the traversal of the abstract syntax tree.
Fields are added to the message using the add_field method.
The message is finalized using the assemble method.
After that the Factory should not be used anymore.
Its a one-time use class, so create a new instance for each Message object.
This class should be used to create Message objects.
"""
def __init__(self):
self.fields: list[Field] = []
def add_field(self, field: Field):
"""Adds a finished field to the message.
This does not check for duplications, this is done in the assemble method.
Args:
field (BprotoField): The field to add to the message.
"""
self.fields.append(field)
@staticmethod
def sort_fields_dict(fields: dict[ComponentName, Field]) -> dict[ComponentName, Field]:
"""Static methode for sorting a dictionary of bproto fields by their position.
Args:
fields (dict[ComponentName, BprotoField]): The fields to sort.
Returns:
dict[ComponentName, BprotoField]: The sorted fields.
"""
return OrderedDict(sorted(fields.items(), key=lambda x: x[1].pos))
def assemble(self, name: ComponentName, index_number: int) -> Message:
"""Finalize the message and create the Message object, returning it.
After this method is called, the Factory should not be used anymore.
Args:
name (ComponentName): Name of the message, should be unique. (No uniqueness checks are done here)
index_number (int): Index number of message, used for ordering, should be unique. (No uniqueness checks are done here)
Raises:
BprotoDuplicateNameError: Raise if a field name is used more than once.
Returns:
BprotoMessage: The finished message object.
"""
resulting_message = Message(NameStyleBproto.fromStr(name), index_number)
field_names: set[ComponentName] = set()
field_positions: set[int] = set()
fields_dict: dict[ComponentName, Field] = {}
for i in self.fields:
if i.name in field_names:
raise BprotoDuplicateNameError(i.name, resulting_message)
if i.pos in field_positions:
raise BprotoMessageIDAlreadyUsed(i.pos, resulting_message)
field_positions.add(i.pos)
field_names.add(i.name)
fields_dict[i.name] = i
fields_dict = FactoryMessage.sort_fields_dict(fields_dict)
resulting_message.fields = fields_dict
return resulting_message

57
src/protocol_components/protocol.py Normal file
View File

@@ -0,0 +1,57 @@
from protocol_components.message import Message
from protocol_components.bitfields import Bitfield
from protocol_components.enumeration import Enumeration
from nameHandling.base import ComponentName
from copy import deepcopy
class ProtocolDefinitions():
"""Incapuslates all the components of a protocol definition
"""
def __init__(self, name: ComponentName, version: int,
enums: dict[ComponentName, Enumeration],
bitfields: dict[ComponentName, Bitfield],
messages: dict[ComponentName, Message]
):
"""Initializes the ProtocolDefinitions class; Should not be called directly
Use FactoryProtocolDefition.assemble instead
Args:
name (ComponentName): Name of the protocol
version (int): Version of the protocol
enums (dict[ComponentName, Enumeration]): Enums of the protocol
bitfields (dict[ComponentName, Bitfield]): Bitfields of the protocol
messages (dict[ComponentName, Message]): Messages of the protocol
"""
self.name: ComponentName = name
self.version = version
self.enums = enums
self.bitfields = bitfields
self.messages = messages
self.protocol_hash_inital: None | int = None
def __deepcopy__(self, memo={}):
if id(self) in memo:
return memo[id(self)]
res = ProtocolDefinitions(
deepcopy(self.name, memo),
deepcopy(self.version, memo),
enums={
deepcopy(k, memo): deepcopy(v, memo)
for k, v in self.enums.items()
},
bitfields={
deepcopy(k, memo): deepcopy(v, memo)
for k, v in self.bitfields.items()
},
messages={
deepcopy(k, memo): deepcopy(v, memo)
for k, v in self.messages.items()
}
)
res.protocol_hash_inital = self.protocol_hash_inital
memo[id(self)] = res
return res

107
src/protocol_components/protocolFactory.py Normal file
View File

@@ -0,0 +1,107 @@
from protocol_components.message import Message
from protocol_components.bitfields import Bitfield
from protocol_components.enumeration import Enumeration
from protocol_components.protocol import ProtocolDefinitions
from protocol_components import AbstractProtocolComponent
from nameHandling.resolver import ProtocolReferenceResolver
from errors import BprotoAlreadyDefinedError
from nameHandling.base import ComponentName, NameStyleBproto
from collections import OrderedDict
class FactoryProtocolDefition:
"""Factory class for creating ProtocolDefinitions
This a one-time use class, once the ProtocolDefinitions object is created, the factory should not be used anymore
"""
def __init__(self):
"""Initializes the FactoryProtocolDefition class; Nothing special here
"""
self.bitfields: dict[ComponentName, Bitfield] = {}
self.enums: dict[ComponentName, Enumeration] = {}
self.messages: dict[ComponentName, Message] = {}
def add_message(self, message: Message):
"""Add a message to the factory, name must be unique
Args:
message (Message): message to add
Raises:
BprotoAlreadyDefinedError: Thrown if message name was already defined
"""
if message.name in self.messages:
raise BprotoAlreadyDefinedError(message)
self.messages.update({message.name: message})
def add_bitfield(self, bitfield: Bitfield):
"""Add a bitfield to the factory, name must be unique
Args:
bitfield (Bitfield): Bitfield to add with unique name
Raises:
BprotoAlreadyDefinedError: Thrown if bitfield name was already defined
"""
if bitfield.name in self.bitfields:
raise BprotoAlreadyDefinedError(bitfield)
self.bitfields.update({bitfield.name: bitfield})
def add_enum(self, enum: Enumeration):
"""Add an enumeration to the factory, name must be unique
Args:
enum (Enumeration): Enumeration to add with unique name
Raises:
BprotoAlreadyDefinedError: Thrown if enum name was already defined
"""
if enum.name in self.enums:
raise BprotoAlreadyDefinedError(enum)
self.enums.update({enum.name: enum})
@staticmethod
def sort_components(components: dict[str, AbstractProtocolComponent]) -> dict[str, AbstractProtocolComponent]:
"""Sorts the components of a protocol definitiown by their identifier (can be integer or string)
Args:
components (dict[str, AbstractProtocolComponent]): Components, dictionary of messages
Returns:
dict[str,AbstractProtocolComponent]: Sorted dictionary of components, is a OrderedDict
"""
return OrderedDict(sorted(components.items(), key=lambda x: x[1].get_identifier()))
def assemble(self, name: str, version: int) -> ProtocolDefinitions:
"""Finalizes the ProtocolDefinitions object and returns it
After calling this method, the factory should not be used anymore
This:
1. Resolves refrence fields
2. Sorts the messages
3. Creates the ProtocolDefinitions object
Args:
name (str): Name of the protocol, gets converted to ComponentName
version (int): Version of the protocol
Returns:
ProtocolDefinitions: final ProtocolDefinitions object
"""
self.messages = self.sort_components(self.messages)
resulting_protocol_defition = ProtocolDefinitions(
NameStyleBproto.fromStr(name, "protocol_name"),
version,
self.enums,
self.bitfields,
FactoryProtocolDefition.sort_components(self.messages)
)
resolver = ProtocolReferenceResolver(resulting_protocol_defition)
resolver.resolve_refrence_protocol()
resolver.back_resolve_inlines()
return resulting_protocol_defition