Add custom exceptions for set and parse errors

Ref. #510

rflx/error.py

@@ -65,6 +65,7 @@ class Subsystem(Enum):
     MODEL = auto()
     CLI = auto()
     GRAPH = auto()
+    PYRFLX = auto()
     ID = auto()
 
     def __str__(self) -> str:

rflx/pyrflx/__init__.py

@@ -1,12 +1,12 @@
 from .bitstring import Bitstring  # noqa: F401
+from .error import PyRFLXError  # noqa: F401
 from .package import Package  # noqa: F401
 from .pyrflx import PyRFLX  # noqa: F401
 from .typevalue import (  # noqa: F401
     ArrayValue,
     EnumValue,
     IntegerValue,
     MessageValue,
-    NotInitializedError,
     OpaqueValue,
     TypeValue,
 )

rflx/pyrflx/bitstring.py

@@ -1,10 +1,12 @@
 from typing import Sequence, Union
 
+from rflx.pyrflx.error import PyRFLXError
+
 
 class Bitstring:
     def __init__(self, bits: str = ""):
         if not self.valid_bitstring(bits):
-            raise ValueError("Bitstring does not consist of only 0 and 1")
+            raise PyRFLXError("Bitstring does not consist of only 0 and 1")
         self._bits = bits
 
     def __add__(self, other: "Bitstring") -> "Bitstring":

rflx/pyrflx/error.py

@@ -0,0 +1,7 @@
+from rflx.error import RecordFluxError, Severity, Subsystem
+
+
+class PyRFLXError(RecordFluxError):
+    def __init__(self, message: str) -> None:
+        super().__init__()
+        self.append(message, Subsystem.PYRFLX, Severity.ERROR)

rflx/pyrflx/typevalue.py

@@ -1,5 +1,4 @@
 # pylint: disable=too-many-lines
-
 from abc import abstractmethod
 from dataclasses import dataclass
 from typing import Any, Callable, Dict, List, Mapping, Optional, Sequence, Tuple, Union
@@ -40,10 +39,7 @@
     Type,
 )
 from rflx.pyrflx.bitstring import Bitstring
-
-
-class NotInitializedError(Exception):
-    pass
+from rflx.pyrflx.error import PyRFLXError, Severity, Subsystem
 
 
 class TypeValue(Base):
@@ -79,7 +75,7 @@ def initialized(self) -> bool:
 
     def _raise_initialized(self) -> None:
         if not self.initialized:
-            raise NotInitializedError("value not initialized")
+            raise PyRFLXError(f"value {self.identifier} not initialized")
 
     def clear(self) -> None:
         self._value = None
@@ -129,7 +125,7 @@ def construct(
             return ArrayValue(vtype)
         if isinstance(vtype, Message):
             return MessageValue(vtype, refinements)
-        raise ValueError("cannot construct unknown type: " + type(vtype).__name__)
+        raise PyRFLXError("cannot construct unknown type: " + type(vtype).__name__)
 
 
 class ScalarValue(TypeValue):
@@ -174,7 +170,7 @@ def assign(self, value: int, check: bool = True) -> None:
             .simplified()
             != TRUE
         ):
-            raise ValueError(f"value {value} not in type range {self._first} .. {self._last}")
+            raise PyRFLXError(f"value {value} not in type range {self._first} .. {self._last}")
         self._value = value
 
     def parse(self, value: Union[Bitstring, bytes], check: bool = True) -> None:
@@ -226,7 +222,7 @@ def assign(self, value: str, check: bool = True) -> None:
             else self._type.package * value
         )
         if Variable(prefixed_value) not in self.literals:
-            raise KeyError(f"{value} is not a valid enum value")
+            raise PyRFLXError(f"{value} is not a valid enum value")
         r = (
             (
                 And(*self._type.constraints("__VALUE__", check, not self.__imported))
@@ -258,7 +254,7 @@ def parse(self, value: Union[Bitstring, bytes], check: bool = True) -> None:
             if self._type.always_valid:
                 self._value = "UNKNOWN", value_as_number
             else:
-                raise KeyError(f"Number {value_as_number.value} is not a valid enum value")
+                raise PyRFLXError(f"Number {value_as_number.value} is not a valid enum value")
         else:
             for k, v in self.literals.items():
                 if v == value_as_number:
@@ -320,7 +316,7 @@ def _check_size_of_assigned_value(
             and isinstance(self._expected_size, Number)
             and size_of_value != self._expected_size.value
         ):
-            raise ValueError(
+            raise PyRFLXError(
                 f"invalid data size: input size is {len(value) * 8} "
                 f"while expected input size is {self._expected_size.value}"
             )
@@ -350,11 +346,13 @@ def parse(self, value: Union[Bitstring, bytes], check: bool = True) -> None:
             nested_msg = self._refinement_message.clone()
             try:
                 nested_msg.parse(value, check)
-            except (IndexError, ValueError, KeyError) as e:
-                raise ValueError(
-                    f"Error while parsing nested message "
-                    f"{self._refinement_message.identifier}: {e}"
-                ) from e
+            except PyRFLXError as e:
+                e.append(
+                    f"Error while parsing nested message " f"{self._refinement_message.identifier}",
+                    Subsystem.PYRFLX,
+                    Severity.ERROR,
+                )
+                raise e
             assert nested_msg.valid_message
             self._nested_message = nested_msg
             self._value = nested_msg.bytestring
@@ -414,22 +412,22 @@ def assign(self, value: List[TypeValue], check: bool = True) -> None:
                 if isinstance(v, MessageValue):
                     assert isinstance(self._element_type, Message)
                     if not v.equal_type(self._element_type):
-                        raise ValueError(
+                        raise PyRFLXError(
                             f'cannot assign "{v.name}" to an array of "{self._element_type.name}"'
                         )
                     if not v.valid_message:
-                        raise ValueError(
+                        raise PyRFLXError(
                             f'cannot assign message "{v.name}" to array of messages: '
                             f"all messages must be valid"
                         )
                 else:
-                    raise ValueError(
+                    raise PyRFLXError(
                         f"cannot assign {type(v).__name__} to an array of "
                         f"{type(self._element_type).__name__}"
                     )
             else:
                 if isinstance(v, MessageValue) or not v.equal_type(self._element_type):
-                    raise ValueError(
+                    raise PyRFLXError(
                         f"cannot assign {type(v).__name__} to an array of "
                         f"{type(self._element_type).__name__}"
                     )
@@ -447,11 +445,14 @@ def parse(self, value: Union[Bitstring, bytes], check: bool = True) -> None:
                 assert isinstance(nested_message, MessageValue)
                 try:
                     nested_message.parse(value, check)
-                except (IndexError, ValueError, KeyError) as e:
-                    raise ValueError(
+                except PyRFLXError as e:
+                    e.append(
                         f"cannot parse nested messages in array of type "
-                        f"{self._element_type.full_name}: {e}"
-                    ) from e
+                        f"{self._element_type.full_name}",
+                        Subsystem.PYRFLX,
+                        Severity.ERROR,
+                    )
+                    raise e
                 assert nested_message.valid_message
                 self._value.append(nested_message)
                 value = value[len(nested_message.bitstring) :]
@@ -472,7 +473,7 @@ def parse(self, value: Union[Bitstring, bytes], check: bool = True) -> None:
 
             self._value = new_value
         else:
-            raise NotImplementedError(f"Arrays of {self._element_type} currently not supported")
+            raise PyRFLXError(f"Arrays of {self._element_type} currently not supported")
 
     @property
     def size(self) -> Expr:
@@ -738,7 +739,7 @@ def set_field_with_size(field_name: str, field_size: int) -> Tuple[int, int]:
                         current_field_first_in_bitstr,
                     ) = set_field_with_size(current_field_name, current_field_size)
                 except IndexError:
-                    raise IndexError(
+                    raise PyRFLXError(
                         f"Bitstring representing the message is too short - "
                         f"stopped while parsing field: {current_field_name}"
                     ) from None
@@ -783,7 +784,7 @@ def check_outgoing_condition_satisfied() -> None:
                 for o in self._type.outgoing(Field(field_name))
             ):
                 self._fields[field_name].typeval.clear()
-                raise ValueError(
+                raise PyRFLXError(
                     f"none of the field conditions "
                     f"{[str(o.condition) for o in self._type.outgoing(Field(field_name))]}"
                     f" for field {field_name} have been met by the assigned value: {value!s}"
@@ -819,14 +820,19 @@ def check_outgoing_condition_satisfied() -> None:
                 elif isinstance(value, field.typeval.accepted_type):
                     field.typeval.assign(value)
                 else:
-                    raise TypeError(
+                    raise PyRFLXError(
                         f"cannot assign different types: {field.typeval.accepted_type.__name__}"
                         f" != {type(value).__name__}"
                     )
-            except (ValueError, KeyError, TypeError) as e:
-                raise ValueError(f"Error while setting value for field {field_name}: {e}") from e
+            except PyRFLXError as e:
+                e.append(
+                    f"cannot set value for field {field_name}",
+                    Subsystem.PYRFLX,
+                    Severity.ERROR,
+                )
+                raise e
         else:
-            raise KeyError(f"cannot access field {field_name}")
+            raise PyRFLXError(f"cannot access field {field_name}")
 
         self.__update_simplified_mapping()
         check_outgoing_condition_satisfied()
@@ -880,14 +886,14 @@ def _preset_fields(self, fld: str) -> None:
     def set_checksum_function(self, checksums: Dict[str, Callable]) -> None:
         for checksum_field_name, checksum_function in checksums.items():
             if checksum_field_name not in self.fields:
-                raise KeyError(
+                raise PyRFLXError(
                     f"cannot set checksum function: field {checksum_field_name} is not defined"
                 )
             for field_name, checksum in self._checksums.items():
                 if field_name == checksum_field_name:
                     checksum.function = checksum_function
                 else:
-                    raise KeyError(
+                    raise PyRFLXError(
                         f"cannot set checksum function: field {checksum_field_name} "
                         f"has not been defined as a checksum field"
                     )
@@ -970,7 +976,7 @@ def _calculate_checksum(
         self, checksum: "MessageValue.Checksum"
     ) -> Union[bytes, int, str, Sequence[TypeValue], Bitstring]:
         if not checksum.function:
-            raise AttributeError(
+            raise PyRFLXError(
                 f"cannot calculate checksum for {checksum.field_name}: "
                 f"no callable checksum function provided"
             )
@@ -1000,7 +1006,7 @@ def _calculate_checksum(
 
     def get(self, field_name: str) -> Union["MessageValue", Sequence[TypeValue], int, str, bytes]:
         if field_name not in self.valid_fields:
-            raise ValueError(f"field {field_name} not valid")
+            raise PyRFLXError(f"field {field_name} not valid")
         field = self._fields[field_name]
         if isinstance(field.typeval, OpaqueValue) and field.typeval.nested_message is not None:
             return field.typeval.nested_message
@@ -1039,7 +1045,7 @@ def _unchecked_bytestring(self) -> bytes:
     @property
     def bytestring(self) -> bytes:
         if not self._skip_verification and not self.valid_message:
-            raise RuntimeError("cannot create bytestring of invalid message")
+            raise PyRFLXError(f"cannot create bytestring of invalid message: {self.identifier}")
         return self._unchecked_bytestring()
 
     @property