Commit fb5776a3 authored by Lluis Gifre Renom's avatar Lluis Gifre Renom
Browse files

Forecaster component: 

- Updated TODO.txt
- Corrected input parameters in method get_kpi_id_samples of KpiManager
- Corrected method compute_forecast of Forecaster module
- Corrected multiple bugs in ForecasterServicerImpl
- Extended temporary test data files
parent 098a6cbe

src/forecaster/service/Forecaster.py

+6 −2
Original line number Diff line number Diff line
@@ -13,6 +13,7 @@ 
# limitations under the License.



import math, pandas

from datetime import datetime, timezone

from statistics import mean

from sklearn.ensemble import RandomForestRegressor

from common.proto.monitoring_pb2 import KpiId
@@ -20,7 +21,7 @@ from forecaster.Config import FORECAST_TO_HISTORY_RATIO 


def compute_forecast(samples : pandas.DataFrame, kpi_id : KpiId) -> float:

    kpi_uuid = kpi_id.kpi_id.uuid

    samples = samples[samples.kpi_id == kpi_uuid]

    samples = samples[samples.kpi_id == kpi_uuid].copy()



    num_samples = samples.shape[0]

    num_samples_test = math.ceil(num_samples / FORECAST_TO_HISTORY_RATIO)
@@ -28,11 +29,14 @@ def compute_forecast(samples : pandas.DataFrame, kpi_id : KpiId) -> float: 
    if num_samples_train <= 0: raise Exception('Wrong number of train samples: {:d}'.format(num_samples_train))

    if num_samples_test  <= 0: raise Exception('Wrong number of test  samples: {:d}'.format(num_samples_test ))



    samples['timestamp'] = pandas.to_datetime(samples['timestamp']) - datetime(1970, 1, 1, tzinfo=timezone.utc)

    samples['timestamp'] = samples['timestamp'].dt.total_seconds()



    train_set = samples[0:num_samples_train]

    test_set  = samples[num_samples_train:num_samples]



    rfr = RandomForestRegressor(n_estimators=600, random_state=42)

    rfr.fit(train_set.drop(['kpi_id', 'value'], axis=1), train_set['value'])

    forecast = rfr.predict(test_set.drop(['value'], axis=1))

    forecast = rfr.predict(test_set.drop(['kpi_id', 'value'], axis=1))

    avg_forecast = round(mean(forecast), 2)

    return avg_forecast

src/forecaster/service/ForecasterServiceServicerImpl.py

+17 −16
Original line number Diff line number Diff line
@@ -52,18 +52,19 @@ class ForecasterServiceServicerImpl(ForecasterServiceServicer): 
        history_window_seconds = FORECAST_TO_HISTORY_RATIO * forecast_window_seconds



        link_id = request.link_id

        link_uuid = link_id.link_uuid.uuid



        context_client = ContextClient()

        link = get_link(context_client, link_id.link_uuid.uuid)

        if link is None: raise NotFoundException('Link', link_id.link_uuid.uuid)

        link = get_link(context_client, link_uuid)

        if link is None: raise NotFoundException('Link', link_uuid)



        kpi_id_map = self._kpi_manager.get_kpi_ids_from_link_ids([link_id])

        kpi_to_link_ids = {

            link_id : kpi_id

            for (link_id, kpi_sample_type), kpi_id in kpi_id_map.items()

            if kpi_sample_type == KpiSampleType.KPISAMPLETYPE_LINK_USED_CAPACITY_GBPS

        link_uuid__to__kpi_id = {

            _link_uuid : _kpi_id

            for (_link_uuid, _kpi_sample_type), _kpi_id in kpi_id_map.items()

            if _kpi_sample_type == KpiSampleType.KPISAMPLETYPE_LINK_USED_CAPACITY_GBPS

        }

        kpi_id = kpi_to_link_ids[link_id]

        kpi_id = link_uuid__to__kpi_id[link_uuid]



        end_timestamp   = time_utc_now_to_float()

        start_timestamp = end_timestamp - history_window_seconds
@@ -71,7 +72,7 @@ class ForecasterServiceServicerImpl(ForecasterServiceServicer): 
        forecast_used_capacity_gbps = compute_forecast(df_historical_data, kpi_id)



        reply = ForecastLinkCapacityReply()

        reply.link_id.CopyFrom(link_id) # pylint: disable=no-member

        reply.link_id.link_uuid.uuid      = link_uuid

        reply.total_capacity_gbps         = link.attributes.total_capacity_gbps

        reply.current_used_capacity_gbps  = link.attributes.used_capacity_gbps

        reply.forecast_used_capacity_gbps = forecast_used_capacity_gbps
@@ -102,23 +103,23 @@ class ForecasterServiceServicerImpl(ForecasterServiceServicer): 
            link_capacities[link.link_id.link_uuid.uuid] = link.attributes



        kpi_id_map = self._kpi_manager.get_kpi_ids_from_link_ids(link_ids)

        kpi_to_link_ids = {

            link_id : kpi_id

            for (link_id, kpi_sample_type), kpi_id in kpi_id_map.items()

            if kpi_sample_type == KpiSampleType.KPISAMPLETYPE_LINK_USED_CAPACITY_GBPS

        link_uuid__to__kpi_id = {

            _link_id : _kpi_id

            for (_link_id, _kpi_sample_type), _kpi_id in kpi_id_map.items()

            if _kpi_sample_type == KpiSampleType.KPISAMPLETYPE_LINK_USED_CAPACITY_GBPS

        }



        kpi_ids = list(kpi_to_link_ids.keys())

        kpi_ids = list(link_uuid__to__kpi_id.values())

        end_timestamp   = time_utc_now_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)



        reply = ForecastTopologyCapacityReply()

        for link_id, kpi_id in kpi_to_link_ids.items():

            link_attributes = link_capacities[link_id]

        for link_uuid, kpi_id in link_uuid__to__kpi_id.items():

            link_attributes = link_capacities[link_uuid]

            forecast_used_capacity_gbps = compute_forecast(df_historical_data, kpi_id)

            link_capacity : ForecastLinkCapacityReply = reply.link_capacities.add() # pylint: disable=no-member

            link_capacity.link_id.CopyFrom(link_id)

            link_capacity.link_id.link_uuid.uuid      = link_uuid

            link_capacity.total_capacity_gbps         = link_attributes.total_capacity_gbps

            link_capacity.current_used_capacity_gbps  = link_attributes.used_capacity_gbps

            link_capacity.forecast_used_capacity_gbps = forecast_used_capacity_gbps

src/forecaster/service/KpiManager.py

+2 −2
Original line number Diff line number Diff line
@@ -35,10 +35,10 @@ class KpiManager: 
        return kpi_ids



    def get_kpi_id_samples(

        self, kpi_uuids : List[str], start_timestamp : float, end_timestamp : float

        self, kpi_ids : List[KpiId], start_timestamp : float, end_timestamp : float

    ) -> pandas.DataFrame:

        kpi_query = KpiQuery()

        for kpi_uuid in kpi_uuids: kpi_query.kpi_ids.add().kpi_id.uuid = kpi_uuid

        for kpi_id in kpi_ids: kpi_query.kpi_ids.add().kpi_id.uuid = kpi_id.kpi_id.uuid

        kpi_query.start_timestamp.timestamp = start_timestamp   # pylint: disable=no-member

        kpi_query.end_timestamp.timestamp   = end_timestamp     # pylint: disable=no-member

        raw_kpi_table = self._monitoring_client.QueryKpiData(kpi_query)

src/forecaster/tests/data/dataset-short.csv

+498 −0

File changed.

Preview size limit exceeded, changes collapsed.

src/forecaster/tests/data/descriptor.json

+1 −1

File changed.

Preview size limit exceeded, changes collapsed.