Sporadic frame support (#106)

Sporadic frames are now loaded and the contained unconditional
frames are resolved.

CHANGELOG.md

@@ -7,9 +7,14 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
 
 ## [Unreleased]
 
+### Added
+
+- Sporadic frames are parsed into Python objects
+
 ### Changed
 
 - Path to the grammar and LDF template are now resolved absolute to allow using them in `pyinstaller`
+- `LDF::get_frame` can also return sporadic frames
 
 ## [0.15.0] - 2022-08-20
 

ldfparser/frame.py

@@ -331,3 +331,10 @@ def __init__(self, frame_id: int, name: str, frames: List[LinUnconditionalFrame]
         super().__init__(frame_id, name)
         self.frames = frames
         self.collision_resolving_schedule_table = collision_resolving_schedule_table
+
+class LinSporadicFrame():
+    # pylint: disable=too-few-public-methods
+
+    def __init__(self, name: str, frames: List[LinUnconditionalFrame]) -> None:
+        self.name = name
+        self.frames = frames

ldfparser/ldf.py

@@ -4,7 +4,7 @@
 from typing import Union, Dict, List
 
 from .lin import LinVersion
-from .frame import LinFrame, LinUnconditionalFrame, LinEventTriggeredFrame
+from .frame import LinFrame, LinSporadicFrame, LinUnconditionalFrame, LinEventTriggeredFrame
 from .diagnostics import LinDiagnosticFrame, LinDiagnosticRequest, LinDiagnosticResponse
 from .signal import LinSignal
 from .encoding import LinSignalEncodingType
@@ -29,6 +29,7 @@ def __init__(self):
         self._diagnostic_signals: Dict[str, LinSignal] = {}
         self._unconditional_frames: Dict[str, LinUnconditionalFrame] = {}
         self._event_triggered_frames: Dict[str, LinEventTriggeredFrame] = {}
+        self._sporadic_frames: Dict[str, LinSporadicFrame] = {}
         self._diagnostic_frames: Dict[str, LinDiagnosticFrame] = {}
         self._signal_encoding_types: Dict[str, LinSignalEncodingType] = {}
         self._signal_representations: Dict[LinSignal, LinSignalEncodingType] = {}
@@ -86,11 +87,19 @@ def get_slaves(self) -> List[LinSlave]:
         """
         return self._slaves.values()
 
-    def get_frame(self, frame_id: Union[int, str]) -> LinFrame:
+    def get_frame(self, frame_id: Union[int, str]) -> Union[LinUnconditionalFrame, LinEventTriggeredFrame, LinSporadicFrame]:
         try:
             return self.get_unconditional_frame(frame_id)
         except LookupError:
+            pass
+        try:
             return self.get_event_triggered_frame(frame_id)
+        except LookupError:
+            pass
+        try:
+            return self.get_sporadic_frame(frame_id)
+        except LookupError as exc:
+            raise exc
 
     @staticmethod
     def _find_frame(frame_id: Union[int, str], collection: Dict[str, LinFrame]) -> LinFrame:
@@ -165,6 +174,30 @@ def get_event_triggered_frames(self) -> List[LinEventTriggeredFrame]:
         """
         return self._event_triggered_frames.values()
 
+    def get_sporadic_frame(self, frame_id: str) -> LinSporadicFrame:
+        """
+        Returns the sporadic frame with the given name
+
+        :param frame_id:
+        :type frame_id: str
+        :returns: Sporadic LIN frame
+        :rtype: LinSporadicFrame
+        :raises: LookupError if the given frame is not found
+        """
+        frame = self._sporadic_frames.get(frame_id)
+        if frame is None:
+            raise LookupError(f"No frame named '{frame_id}' found!")
+        return frame
+
+    def get_sporadic_frames(self) -> List[LinSporadicFrame]:
+        """
+        Returns all sporadic frames
+
+        :returns: List of sporadic LIN frames
+        :rtype: List[LinSporadicFrame]
+        """
+        return self._sporadic_frames.values()
+
     def get_diagnostic_frame(self, frame_id: Union[int, str]) -> LinDiagnosticFrame:
         """
         Returns the diagnostic frame with the given name or id

ldfparser/parser.py

@@ -6,7 +6,7 @@
 from .diagnostics import LIN_MASTER_REQUEST_FRAME_ID, LIN_SLAVE_RESPONSE_FRAME_ID, LinDiagnosticFrame, LinDiagnosticRequest, LinDiagnosticResponse
 from .schedule import AssignFrameIdEntry, AssignFrameIdRangeEntry, AssignNadEntry, ConditionalChangeNadEntry, DataDumpEntry, FreeFormatEntry, MasterRequestEntry, SaveConfigurationEntry, ScheduleTable, SlaveResponseEntry, UnassignFrameIdEntry, LinFrameEntry
 
-from .frame import LinEventTriggeredFrame, LinUnconditionalFrame
+from .frame import LinEventTriggeredFrame, LinSporadicFrame, LinUnconditionalFrame
 from .signal import LinSignal
 from .encoding import ASCIIValue, BCDValue, LinSignalEncodingType, LogicalValue, PhysicalValue, ValueConverter
 from .lin import LIN_VERSION_2_0, LIN_VERSION_2_1, LinVersion
@@ -64,6 +64,7 @@ def parse_ldf(path: str, capture_comments: bool = False, encoding: str = None) -
     _populate_ldf_signals(json, ldf)
     _populate_ldf_frames(json, ldf)
     _populate_ldf_event_triggered_frames(json, ldf)
+    _populate_ldf_sporadic_frames(json, ldf)
     _populate_diagnostic_signals(json, ldf)
     _populate_diagnostic_frames(json, ldf)
     _populate_ldf_nodes(json, ldf)
@@ -133,6 +134,15 @@ def _populate_ldf_event_triggered_frames(json: dict, ldf: LDF):
             frames.append(ldf.get_unconditional_frame(a))
         ldf._event_triggered_frames[frame['name']] = LinEventTriggeredFrame(frame['frame_id'], frame['name'], frames)
 
+def _populate_ldf_sporadic_frames(json: dict, ldf: LDF):
+    if "sporadic_frames" not in json:
+        return
+    for frame in json['sporadic_frames']:
+        frames = []
+        for a in frame['frames']:
+            frames.append(ldf.get_unconditional_frame(a))
+        ldf._sporadic_frames[frame['name']] = LinSporadicFrame(frame['name'], frames)
+
 def _populate_ldf_nodes(json: dict, ldf: LDF):
     nodes = _require_key(json, 'nodes', 'Missing Nodes section.')
     if nodes.get('master'):

ldfparser/templates/ldf.jinja2

@@ -85,6 +85,17 @@ Event_triggered_frames {
 }
 {%- endif %}
 
+{%- if ldf.get_sporadic_frames() | length > 0 %}
+Sporadic_frames {
+    {%- for frame in ldf.get_sporadic_frames() %}
+    {{frame.name}}: 
+    {%- for unconditional in frame.frames -%}
+        {{unconditional.name}}{%- if not loop.last %}, {% endif -%}
+    {%- endfor -%};
+    {%- endfor %}
+}
+{%- endif %}
+
 Schedule_tables {
     {%- for table in ldf.get_schedule_tables() %}
     {{table.name}} {

schemas/ldf.json

@@ -147,6 +147,20 @@
                 "frames"
             ]
         },
+        "sporadic_frame": {
+            "type": "object",
+            "properties": {
+                "name": {
+                    "type": "string"
+                },
+                "frames": {
+                    "type": "array",
+                    "items": {
+                        "type": "string"
+                    }
+                }
+            }
+        },
         "diagnostic_signal": {
             "type": "object",
             "properties": {
@@ -623,6 +637,12 @@
                 "$ref": "#/definitions/event_triggered_frame"
             }
         },
+        "sporadic_frames": {
+            "type": "array",
+            "items": {
+                "$ref": "#/definitions/sporadic_frame"
+            }
+        },
         "node_attributes": {
             "type": "array",
             "items": {

tests/ldf/ldf_with_sporadic_frames.ldf

@@ -0,0 +1,64 @@
+LIN_description_file;
+LIN_protocol_version = "2.2";
+LIN_language_version = "2.2";
+LIN_speed = 19.2 kbps;
+
+Nodes {
+  Master: MASTER, 10 ms, 0 ms ;
+  Slaves: SLAVE ;
+}
+
+Signals {
+  REQ_POST_RUN_RPM: 16, 0, MASTER, SLAVE ;
+  REQ_POST_RUN_DURATION: 12, 0, MASTER, SLAVE ;
+  CYC_READ_STATUS_LIN_RESPONSE: 1, 0, SLAVE, MASTER;
+}
+
+
+Frames {
+  REQ_POST_RUN: 30, MASTER, 4 {
+    REQ_POST_RUN_RPM, 0 ;
+    REQ_POST_RUN_DURATION, 16 ;
+  }
+}
+
+Sporadic_frames {
+  SF_REQ_POST_RUN: REQ_POST_RUN ;
+}
+
+Node_attributes {
+  SLAVE{
+    LIN_protocol = "2.2" ;
+    configured_NAD = 0xD ;
+    initial_NAD = 0xD ;
+    product_id = 0x2, 0x0, 255 ;
+    response_error = CYC_READ_STATUS_LIN_RESPONSE ;
+    P2_min = 50 ms ;
+    ST_min = 0 ms ;
+    N_As_timeout = 1000 ms ;
+    N_Cr_timeout = 1000 ms ;
+    configurable_frames {
+      REQ_POST_RUN ;
+    }
+  }
+}
+
+Schedule_tables {
+ POST_RUN {
+    SF_REQ_POST_RUN delay 10 ms ;
+  }
+}
+
+
+Signal_encoding_types {
+  POST_RUN_DURATION_Encoding {
+    physical_value, 0, 4092, 1, 0, "s" ;
+    logical_value, 4093, "not avaible" ;
+    logical_value, 4094, "error" ;
+    logical_value, 4095, "signal invalid" ;
+  }
+}
+
+Signal_representation {
+  POST_RUN_DURATION_Encoding: REQ_POST_RUN_DURATION ;
+}

tests/test_parser.py

@@ -184,3 +184,17 @@ def test_load_valid_diagnostics():
 
     with pytest.raises(LookupError):
         ldf.get_diagnostic_signal('MasterReqB9')
+
+@pytest.mark.unit
+def test_load_sporadic_frames():
+    path = os.path.join(os.path.dirname(__file__), "ldf", "ldf_with_sporadic_frames.ldf")
+    ldf = parse_ldf(path)
+
+    assert len(ldf.get_sporadic_frames()) >= 0
+
+    sporadic_frame = ldf.get_frame('SF_REQ_POST_RUN')
+    assert sporadic_frame.name == 'SF_REQ_POST_RUN'
+    assert ldf.get_unconditional_frame('REQ_POST_RUN') in sporadic_frame.frames
+
+    with pytest.raises(LookupError):
+        ldf.get_frame('SF_123')