Forecaster component:

- Migrated to Common/Tools/TimeStamp tools
- Minor code cleanup

src/forecaster/service/ForecasterServiceServicerImpl.py

@@ -25,11 +25,11 @@ from common.proto.forecaster_pb2_grpc import ForecasterServiceServicer
 from common.proto.kpi_sample_types_pb2 import KpiSampleType
 from common.tools.context_queries.Link import get_link
 from common.tools.context_queries.Topology import get_topology_details
+from common.tools.timestamp.Converters import timestamp_utcnow_to_float
 from context.client.ContextClient import ContextClient
 from forecaster.Config import FORECAST_TO_HISTORY_RATIO
 from forecaster.service.Forecaster import compute_forecast
 from forecaster.service.KpiManager import KpiManager
-from forecaster.service.Tools import time_utc_now_to_float
 
 LOGGER = logging.getLogger(__name__)
 
@@ -66,7 +66,7 @@ class ForecasterServiceServicerImpl(ForecasterServiceServicer):
         }
         kpi_id = link_uuid__to__kpi_id[link_uuid]
 
-        end_timestamp   = time_utc_now_to_float()
+        end_timestamp   = timestamp_utcnow_to_float()
         start_timestamp = end_timestamp - history_window_seconds
         df_historical_data = self._kpi_manager.get_kpi_id_samples([kpi_id], start_timestamp, end_timestamp)
         forecast_used_capacity_gbps = compute_forecast(df_historical_data, kpi_id)
@@ -110,7 +110,7 @@ class ForecasterServiceServicerImpl(ForecasterServiceServicer):
         }
 
         kpi_ids = list(link_uuid__to__kpi_id.values())
-        end_timestamp   = time_utc_now_to_float()
+        end_timestamp   = timestamp_utcnow_to_float()
         start_timestamp = end_timestamp - history_window_seconds
         df_historical_data = self._kpi_manager.get_kpi_id_samples(kpi_ids, start_timestamp, end_timestamp)
 

src/forecaster/service/Tools.py

-# Copyright 2022-2023 ETSI TeraFlowSDN - TFS OSG (https://tfs.etsi.org/)
-#
-# Licensed under the Apache License, Version 2.0 (the "License");
-# you may not use this file except in compliance with the License.
-# You may obtain a copy of the License at
-#
-#      http://www.apache.org/licenses/LICENSE-2.0
-#
-# Unless required by applicable law or agreed to in writing, software
-# distributed under the License is distributed on an "AS IS" BASIS,
-# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
-# See the License for the specific language governing permissions and
-# limitations under the License.
-
-import calendar
-from datetime import datetime, timezone
-
-def time_datetime_to_int(dt_time : datetime) -> int:
-    return int(calendar.timegm(dt_time.timetuple()))
-
-def time_datetime_to_float(dt_time : datetime) -> float:
-    return time_datetime_to_int(dt_time) + (dt_time.microsecond / 1.e6)
-
-def time_utc_now_to_datetime() -> datetime:
-    return datetime.now(tz=timezone.utc)
-
-def time_utc_now_to_float() -> float:
-    return time_datetime_to_float(time_utc_now_to_datetime())

src/forecaster/tests/Tools.py

@@ -12,31 +12,19 @@
 # See the License for the specific language governing permissions and
 # limitations under the License.
 
-import calendar, logging, math, pandas
-from datetime import datetime, timezone
+import logging, math, pandas
 from typing import Dict
 from common.tools.object_factory.Context import json_context
 from common.tools.object_factory.Device import (
     json_device_emulated_connect_rules, json_device_emulated_packet_router_disabled, json_device_id
 )
-from common.tools.object_factory.EndPoint import json_endpoint, json_endpoint_descriptor, json_endpoint_id
+from common.tools.object_factory.EndPoint import json_endpoint, json_endpoint_id
 from common.tools.object_factory.Link import json_link
 from common.tools.object_factory.Topology import json_topology
+from common.tools.timestamp.Converters import timestamp_datetime_to_int, timestamp_utcnow_to_float
 
 LOGGER = logging.getLogger(__name__)
 
-def time_datetime_to_int(dt_time : datetime) -> int:
-    return int(calendar.timegm(dt_time.timetuple()))
-
-def time_datetime_to_float(dt_time : datetime) -> float:
-    return time_datetime_to_int(dt_time) + (dt_time.microsecond / 1.e6)
-
-def time_utc_now_to_datetime() -> datetime:
-    return datetime.now(tz=timezone.utc)
-
-def time_utc_now_to_float() -> float:
-    return time_datetime_to_float(time_utc_now_to_datetime())
-
 def read_csv(csv_file : str) -> pandas.DataFrame:
     LOGGER.info('Using Data File "{:s}"...'.format(csv_file))
 
@@ -57,8 +45,8 @@ def read_csv(csv_file : str) -> pandas.DataFrame:
 
     LOGGER.info('Updating timestamps...')
     df['timestamp'] = pandas.to_datetime(df['timestamp'])
-    max_timestamp = time_datetime_to_int(df['timestamp'].max())
-    now_timestamp = time_datetime_to_int(datetime.now(tz=timezone.utc))
+    max_timestamp = timestamp_datetime_to_int(df['timestamp'].max())
+    now_timestamp = timestamp_utcnow_to_float()
     df['timestamp'] = df['timestamp'] + pandas.offsets.Second(now_timestamp - max_timestamp)
     LOGGER.info('  DONE')
 

src/forecaster/tests/test_unitary.py

@@ -12,7 +12,7 @@
 # See the License for the specific language governing permissions and
 # limitations under the License.
 
-import json, logging, pandas, pytest
+import logging, pandas, pytest #, json
 from typing import Dict, Tuple
 from common.Constants import DEFAULT_CONTEXT_NAME, DEFAULT_TOPOLOGY_NAME
 from common.proto.context_pb2 import ContextId, TopologyId