VERSION 0.9.19

Author: jordimariasi2cat

pyproject.toml

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

src/flexstack/applications/road_hazard_signalling_service/emergency_vehicle_approaching_service.py

@@ -1,9 +1,9 @@
-import time
 from .service_access_point import DENRequest, PriorityLevel
 from ...facilities.decentralized_environmental_notification_service.den_service import (
     DecentralizedEnvironmentalNotificationService,
 )
 from ...btp.router import Router as BTPRouter
+from ...utils.time_service import TimeService
 
 
 class EmergencyVehicleApproachingService:
@@ -59,7 +59,7 @@ def __init__(
         self.denm_interval = 100
         self.priority_level = PriorityLevel.WARNING
         # Get DENM data to simulate the hazard detection
-        self.detection_time = int((time.time() - 1072911600 - 5) * 1000)
+        self.detection_time = int((TimeService.time() - 1072915200 + 5) * 1000)
         self.event_position = {
             "latitude": 900000001,
             "longitude": 1800000001,
@@ -101,4 +101,5 @@ def trigger_denm_sending(self, tpv: dict) -> None:
                 )
 
         request.with_emergency_vehicle_approaching(self)
-        self.den_service.denm_transmission_management.request_denm_sending(request)
+        self.den_service.denm_transmission_management.request_denm_sending(
+            request)

src/flexstack/facilities/decentralized_environmental_notification_service/denm_transmission_management.py

@@ -1,6 +1,7 @@
 import time
 import threading
 import logging
+from ...utils.time_service import TimeService
 from ...geonet.service_access_point import Area, GeoBroadcastHST, HeaderType
 from .denm_coder import DENMCoder
 from ...btp.router import Router as BTPRouter
@@ -88,7 +89,7 @@ def fullfill_with_denrequest(self, request: DENRequest) -> None:
         # Add Management Container
         self.denm["denm"]["management"]["detectionTime"] = request.detection_time
         self.denm["denm"]["management"]["referenceTime"] = int(
-            (time.time() - 1072911600 - 5) * 1000
+            (TimeService.time() - 1072915200 + 5) * 1000
         )
         self.denm["denm"]["management"]["TransmissionInterval"] = request.denm_interval
 
@@ -146,7 +147,7 @@ def fullfill_with_collision_risk_warning(self, request: DENRequest) -> None:
         # Add Management Container
         self.denm["denm"]["management"]["detectionTime"] = request.detection_time
         self.denm["denm"]["management"]["referenceTime"] = int(
-            (time.time() - 1072911600-5) % 65536
+            (TimeService.time() - 1072915200 + 5) % 65536
         )
         self.denm["denm"]["management"]["TransmissionInterval"] = request.denm_interval
 
@@ -216,7 +217,8 @@ def request_denm_sending(self, denm_request: DENRequest) -> None:
         """
         Request to send a DENM and starts a thread.
         """
-        t = threading.Thread(target=self.trigger_denm_messages, args=[denm_request])
+        t = threading.Thread(
+            target=self.trigger_denm_messages, args=[denm_request])
         t.start()
 
     def send_collision_risk_warning_denm(self, denm_request: DENRequest) -> None:

tests/flexstack/facilities/decentralized_environmental_notification_service/test_denm_transmission_management.py

@@ -10,6 +10,7 @@
 
 class TestDecentralizedEnvironmentalNotificationMessage(unittest.TestCase):
     """Test class for the DecentralizedEnvironmentalNotificationMessage."""
+
     def test__init__(self):
         """Test DecentralizedEnvironmentalNotificationMessage initialization"""
         decentralized_environmental_notification_message = \
@@ -22,7 +23,7 @@ def test__init__(self):
         # print(encoded_white)
         self.assertEqual(encoded_white, expected_denm)
 
-    @patch('time.time')
+    @patch('flexstack.utils.time_service.TimeService.time')
     def test_fullfill_with_denrequest(self, time_mock):
         """Test DENMCoder decoding"""
         # Given
@@ -55,7 +56,8 @@ def test_fullfill_with_denrequest(self, time_mock):
         # When
         decentralized_environmental_notification_message = \
             DecentralizedEnvironmentalNotificationMessage()
-        decentralized_environmental_notification_message.fullfill_with_denrequest(denm_request)
+        decentralized_environmental_notification_message.fullfill_with_denrequest(
+            denm_request)
 
         # Then
         self.assertEqual(decentralized_environmental_notification_message.denm['denm']
@@ -88,7 +90,7 @@ def test_fullfill_with_denrequest(self, time_mock):
                          ['management']['stationType'], denm_request.rhs_vehicle_type)
         time_mock.assert_called_once()
 
-    @patch('time.time')
+    @patch('flexstack.utils.time_service.TimeService.time')
     def test_fullfill_with_collision_risk_warning(self, time_mock):
         """Test fullfill_with_collision_risk_warning function"""
         # Given
@@ -115,7 +117,8 @@ def test_fullfill_with_collision_risk_warning(self, time_mock):
         # When
         decentralized_environmental_notification_message = \
             DecentralizedEnvironmentalNotificationMessage()
-        decentralized_environmental_notification_message.fullfill_with_collision_risk_warning(denm_request)
+        decentralized_environmental_notification_message.fullfill_with_collision_risk_warning(
+            denm_request)
 
         # Then
         self.assertEqual(decentralized_environmental_notification_message.denm['denm']
@@ -156,6 +159,7 @@ def test_fullfill_with_vehicle_data(self):
 
 class TestDENMTransmissionManagement(unittest.TestCase):
     """Test class for the DENM Transmission Management."""
+
     def setUp(self):
         btp_router = MagicMock()
         btp_router.BTPDataRequest = MagicMock()
@@ -195,12 +199,14 @@ def test_send_collision_risk_warning_denm(self, mock_fullfill_with_vehicle_data,
         self.denm_transmission_management.transmit_denm = MagicMock()
 
         # When
-        self.denm_transmission_management.send_collision_risk_warning_denm(denm_request)
+        self.denm_transmission_management.send_collision_risk_warning_denm(
+            denm_request)
 
         # Then
         mock_fullfill_with_vehicle_data.assert_called_once_with(
             self.denm_transmission_management.vehicle_data)
-        mock_fullfill_with_collision_risk_warning.assert_called_once_with(denm_request)
+        mock_fullfill_with_collision_risk_warning.assert_called_once_with(
+            denm_request)
         self.denm_transmission_management.transmit_denm.assert_called_once()
 
     @patch.object(DecentralizedEnvironmentalNotificationMessage,
@@ -245,4 +251,5 @@ def test_transmit_denm(self):
 
         # Then
         self.denm_transmission_management.btp_router.btp_data_request.assert_called()
-        self.denm_transmission_management.denm_coder.encode.assert_called_once_with(new_denm.denm)
+        self.denm_transmission_management.denm_coder.encode.assert_called_once_with(
+            new_denm.denm)