VERSION 0.10.0

Author: jordimariasi2cat

examples/cam_sender_and_receiver.py

@@ -0,0 +1,89 @@
+import os
+import sys
+THIS_DIR = os.path.dirname(os.path.abspath(__file__))
+PROJECT_ROOT = os.path.abspath(os.path.join(THIS_DIR, os.pardir))
+SRC_DIR = os.path.join(PROJECT_ROOT, "src")
+if os.path.isdir(SRC_DIR) and SRC_DIR not in sys.path:
+    sys.path.insert(0, SRC_DIR)
+
+# Configure logging
+import logging
+logging.basicConfig(level=logging.INFO)
+
+# Link Layer Imports
+from flexstack.linklayer.raw_link_layer import RawLinkLayer
+
+# GeoNetworking imports
+from flexstack.geonet.router import Router as GNRouter
+from flexstack.geonet.mib import MIB
+from flexstack.geonet.gn_address import GNAddress, M, ST, MID
+
+# BTP Router imports
+from flexstack.btp.router import Router as BTPRouter
+
+# Location Service imports
+from flexstack.utils.static_location_service import ThreadStaticLocationService
+
+# CA Basic Service imports
+from flexstack.facilities.ca_basic_service.ca_basic_service import (
+    CooperativeAwarenessBasicService,
+)
+from flexstack.facilities.ca_basic_service.cam_transmission_management import (
+    VehicleData,
+)
+
+POSITION_COORDINATES = [41.386931, 2.112104]
+MAC_ADDRESS = b"\xaa\xbb\xcc\x11\x21\x11"
+STATION_ID = int(MAC_ADDRESS[-1])
+
+
+def main() -> None:    
+    # Instantiate a Location Service
+    location_service = ThreadStaticLocationService(
+        period=1000, latitude=POSITION_COORDINATES[0], longitude=POSITION_COORDINATES[1]
+    )
+
+    # Instantiate a GN router
+    mib = MIB(
+        itsGnLocalGnAddr=GNAddress(
+            m=M.GN_MULTICAST,
+            st=ST.CYCLIST,
+            mid=MID(MAC_ADDRESS),
+            ),
+    )
+    gn_router = GNRouter(mib=mib, sign_service=None)
+    location_service.add_callback(gn_router.refresh_ego_position_vector)
+
+    # Instantiate a Link Layer
+    link_layer = RawLinkLayer(
+        "lo", MAC_ADDRESS, receive_callback=gn_router.gn_data_indicate
+    )
+
+    gn_router.link_layer = link_layer
+
+    # Instantiate a BTP router
+    btp_router = BTPRouter(gn_router)
+    gn_router.register_indication_callback(btp_router.btp_data_indication)
+
+    # Instantiate a CA Basic Service
+    vehicle_data = VehicleData()
+    vehicle_data.station_id = STATION_ID  # Station Id of the ITS PDU Header
+    vehicle_data.station_type = 5  # Station Type as specified in ETSI TS 102 894-2 V2.3.1 (2024-08)
+    vehicle_data.drive_direction = "forward"
+    vehicle_data.vehicle_length = {
+        "vehicleLengthValue": 1023,  # as specified in ETSI TS 102 894-2 V2.3.1 (2024-08)
+        "vehicleLengthConfidenceIndication": "unavailable",
+    }
+    vehicle_data.vehicle_width = 62
+
+    ca_basic_service = CooperativeAwarenessBasicService(
+        btp_router=btp_router,
+        vehicle_data=vehicle_data,
+    )
+    location_service.add_callback(ca_basic_service.cam_transmission_management.location_service_callback)
+
+    print("Press Ctrl+C to stop the program.")
+    location_service.location_service_thread.join()
+
+if __name__ == "__main__":
+    main()

pyproject.toml

@@ -4,7 +4,7 @@ build-backend = "setuptools.build_meta"
 
 [project]
 name = "v2xflexstack"
-version = "0.9.22"
+version = "0.10.0"
 authors = [
   { name = "Jordi Marias-i-Parella", email = "[email protected]" },
   { name = "Daniel Ulied Guevara", email = "[email protected]" },

src/flexstack/btp/btp_header.py

@@ -1,6 +1,8 @@
 from .service_access_point import BTPDataRequest
+from dataclasses import dataclass
 
 
+@dataclass(frozen=True)
 class BTPAHeader:
     """
     BTP-A Header class.
@@ -15,21 +17,20 @@ class BTPAHeader:
         (16 bit integer) Source Port field of BTP-A Header
     """
 
-    def __init__(self) -> None:
-        self.destination_port = 0
-        self.source_port = 0
+    destination_port: int = 0
+    source_port: int = 0
 
-    def initialize_with_request(self, request: BTPDataRequest) -> None:
+    @classmethod
+    def initialize_with_request(cls, request: BTPDataRequest) -> "BTPAHeader":
         """
-        Initializes the BTP-A Header with a GNDataRequest.
+        Initialize a BTP-A Header from a BTPDataRequest.
 
         Parameters
         ----------
-        request : GNDataRequest
-            GNDataRequest to use.
+        request : BTPDataRequest
+            Request to use for initialization.
         """
-        self.destination_port = request.destination_port
-        self.source_port = request.source_port
+        return cls(destination_port=request.destination_port, source_port=request.source_port)
 
     def encode_to_int(self) -> int:
         """
@@ -53,19 +54,22 @@ def encode(self) -> bytes:
         """
         return self.encode_to_int().to_bytes(4, byteorder='big')
 
-    def decode(self, data: bytes) -> None:
+    @classmethod
+    def decode(cls, data: bytes) -> "BTPAHeader":
         """
-        Decodes the BTP-A Header from bytes.
+        Decodes the BTP-A Header from bytes and returns a new instance.
 
         Parameters
         ----------
         data : bytes
             Bytes to decode.
         """
-        self.destination_port = int.from_bytes(data[0:2], byteorder='big')
-        self.source_port = int.from_bytes(data[2:4], byteorder='big')
+        destination_port = int.from_bytes(data[0:2], byteorder='big')
+        source_port = int.from_bytes(data[2:4], byteorder='big')
+        return cls(destination_port=destination_port, source_port=source_port)
 
 
+@dataclass(frozen=True)
 class BTPBHeader:
     """
     BTP-B Header class.
@@ -80,21 +84,20 @@ class BTPBHeader:
         (16 bit integer) Destination Port Info field of BTP-B Header
     """
 
-    def __init__(self) -> None:
-        self.destination_port = 0
-        self.destination_port_info = 0
+    destination_port: int = 0
+    destination_port_info: int = 0
 
-    def initialize_with_request(self, request: BTPDataRequest) -> None:
+    @classmethod
+    def initialize_with_request(cls, request: BTPDataRequest) -> "BTPBHeader":
         """
-        Initializes the BTP-B Header with a GNDataRequest.
+        Initialize a BTP-B Header from a BTPDataRequest.
 
         Parameters
         ----------
-        request : GNDataRequest
-            GNDataRequest to use.
+        request : BTPDataRequest
+            Request to use for initialization.
         """
-        self.destination_port = request.destination_port
-        self.destination_port_info = request.destinaion_port_info
+        return cls(destination_port=request.destination_port, destination_port_info=getattr(request, 'destination_port_info', 0))
 
     def encode_to_int(self) -> int:
         """
@@ -118,14 +121,16 @@ def encode(self) -> bytes:
         """
         return self.encode_to_int().to_bytes(4, byteorder='big')
 
-    def decode(self, data: bytes) -> None:
+    @classmethod
+    def decode(cls, data: bytes) -> "BTPBHeader":
         """
-        Decodes the BTP-B Header from bytes.
+        Decodes the BTP-B Header from bytes and returns a new instance.
 
         Parameters
         ----------
         data : bytes
             Bytes to decode.
         """
-        self.destination_port = int.from_bytes(data[0:2], byteorder='big')
-        self.destination_port_info = int.from_bytes(data[2:4], byteorder='big')
+        destination_port = int.from_bytes(data[0:2], byteorder='big')
+        destination_port_info = int.from_bytes(data[2:4], byteorder='big')
+        return cls(destination_port=destination_port, destination_port_info=destination_port_info)

src/flexstack/btp/router.py

@@ -1,8 +1,9 @@
 from __future__ import annotations
 from collections.abc import Callable
+from types import MappingProxyType
 import logging
 
-from .btp_header import BTPAHeader, BTPBHeader
+from .btp_header import BTPBHeader
 from .service_access_point import BTPDataIndication, BTPDataRequest
 from ..geonet.common_header import CommonNH
 from ..geonet.service_access_point import GNDataIndication, GNDataRequest
@@ -17,20 +18,29 @@ class Router:
 
     Attributes
     ----------
-    indication_callbacks : Dict[int, Callable[[BTPDataIndication], None]]
-        Dictionary of indication callbacks. The key is the port and the value is the callback.
+    indication_callbacks : MappingProxyType[int, Callable[[BTPDataIndication], None]] | None
+        MappingProxyType of indication callbacks. The key is the port and the value is the callback.
     gn_router : GNRouter
         Geonetworking Router.
     """
 
     def __init__(self, gn_router: GNRouter) -> None:
         self.logging = logging.getLogger("btp")
-
-        self.indication_callbacks: dict[int, Callable[[BTPDataIndication], None]] = {}
+        self.pre_indication_callbacks: dict[int,
+                                            Callable[[BTPDataIndication], None]] = {}
+        self.indication_callbacks: MappingProxyType[int, Callable[[
+            BTPDataIndication], None]] | None = None
         self.gn_router = gn_router
 
         self.logging.info("BTP Router Initialized!")
 
+    def freeze_callbacks(self):
+        """
+        Freezes the indication callbacks. After calling this method, no more callbacks can be registered.
+        """
+        self.indication_callbacks = MappingProxyType(
+            dict(self.pre_indication_callbacks))
+
     def register_indication_callback_btp(
         self, port: int, callback: Callable[[BTPDataIndication], None]
     ) -> None:
@@ -44,7 +54,9 @@ def register_indication_callback_btp(
         callback : Callable[[BTPDataIndication], None]
             Callback to register.
         """
-        self.indication_callbacks[port] = callback
+        if self.indication_callbacks is not None:
+            raise RuntimeError("Cannot register callbacks after freezing")
+        self.pre_indication_callbacks[port] = callback
         self.logging.info("Indication callback registered")
 
     def btp_data_request(self, request: BTPDataRequest) -> None:
@@ -57,19 +69,22 @@ def btp_data_request(self, request: BTPDataRequest) -> None:
             BTPDataRequest to handle.
         """
         if request.btp_type == CommonNH.BTP_B:
-            header = BTPBHeader()
-            header.destination_port_info = request.destinaion_port_info
-            header.destination_port = request.destination_port
-            gn_data_request = GNDataRequest()
-            gn_data_request.upper_protocol_entity = request.btp_type
-            gn_data_request.packet_transport_type = request.gn_packet_transport_type
-            gn_data_request.area = request.gn_area
-            gn_data_request.communication_profile = request.communication_profile
-            gn_data_request.traffic_class = request.traffic_class
-            gn_data_request.data = header.encode() + request.data
-            gn_data_request.length = len(gn_data_request.data)
-
-            self.logging.debug("Sending BTP Data Request: %s", gn_data_request.data)
+            header = BTPBHeader(
+                destination_port=request.destination_port,
+                destination_port_info=request.destination_port_info,
+            )
+            data = header.encode() + request.data
+            gn_data_request = GNDataRequest(
+                upper_protocol_entity=request.btp_type,
+                packet_transport_type=request.gn_packet_transport_type,
+                area=request.gn_area,
+                communication_profile=request.communication_profile,
+                traffic_class=request.traffic_class,
+                data=data,
+                length=len(data)
+            )
+            self.logging.debug(
+                "Sending BTP Data Request: %s", gn_data_request.data)
             self.gn_router.gn_data_request(gn_data_request)
         elif request.btp_type == CommonNH.BTP_A:
             raise NotImplementedError("BTPADataRequest not implemented")
@@ -85,16 +100,20 @@ def btp_b_data_indication(self, gn_data_indication: GNDataIndication) -> None:
         gn_data_indication : GNDataIndication
             GNDataIndication to handle.
         """
-        indication = BTPDataIndication()
-        indication.initialize_with_gn_data_indication(gn_data_indication)
-        header = BTPBHeader()
-        header.decode(gn_data_indication.data)
-        indication.destination_port = header.destination_port
-        indication.destinaion_port_info = header.destination_port_info
-        for port, callback in self.indication_callbacks.items():
-            if port == indication.destination_port:
-                self.logging.debug("Sending BTP B Data Indication: %s", indication.data)
+        indication = BTPDataIndication.initialize_with_gn_data_indication(
+            gn_data_indication)
+        header = BTPBHeader.decode(gn_data_indication.data)
+        indication = indication.set_destination_port_and_info(
+            destination_port=header.destination_port,
+            destination_port_info=header.destination_port_info
+        )
+        if self.indication_callbacks:
+            callback = self.indication_callbacks.get(
+                indication.destination_port)
+            if callback:
                 callback(indication)
+        else:
+            raise RuntimeError("Indication callbacks not frozen")
 
     def btp_a_data_indication(self, gn_data_indication: GNDataIndication) -> None:
         """
@@ -105,8 +124,6 @@ def btp_a_data_indication(self, gn_data_indication: GNDataIndication) -> None:
         gn_data_indication : GNDataIndication
             GNDataIndication to handle.
         """
-        header = BTPAHeader()
-        header.decode(gn_data_indication.data)
         raise NotImplementedError("BTPADataIndication not implemented")
 
     def btp_data_indication(self, gn_data_indication: GNDataIndication) -> None: