Service component - L3NM IETF Slice:

# Copyright 2022-2025 ETSI SDG TeraFlowSDN (TFS) (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 json

from deepdiff import DeepDiff

from typing import Dict, Optional

from common.proto.context_pb2 import Service

RUNNING_RESOURCE_KEY   = 'running_ietf_slice'

CANDIDATE_RESOURCE_KEY = 'candidate_ietf_slice'

class DataStoreDelta:

    def __init__(self, service : Service):

        self._service = service

        self._service_config = service.service_config

        self._candidate_data = self._get_datastore_data(CANDIDATE_RESOURCE_KEY)

        self._running_data   = self._get_datastore_data(RUNNING_RESOURCE_KEY  )

    def _get_datastore_data(self, resource_key : str) -> Optional[Dict]:

        for cr in self._service_config.config_rules:

            if cr.WhichOneof('config_rule') != 'custom': continue

            if cr.custom.resource_key != resource_key: continue

            resource_value = json.loads(cr.custom.resource_value)

            return resource_value.get('network-slice-services', dict()).get('slice-service')

        return None

    @property

    def candidate_data(self): return self._candidate_data

    @property

    def running_data(self): return self._running_data

    def get_diff(self) -> Dict:

        return DeepDiff(self._running_data, self._candidate_data)

# Copyright 2022-2025 ETSI SDG TeraFlowSDN (TFS) (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.

from dataclasses import dataclass

from typing import Dict, TypedDict

from common.proto.context_pb2 import Device

def get_device_endpoint_name(device_obj : Device, endpoint_uuid : str) -> str:

    '''

    Given a device object and an endpoint UUID, return the device endpoint name.

    Raises an exception if not found.

    '''

    for d_ep in device_obj.device_endpoints:

        if d_ep.endpoint_id.endpoint_uuid.uuid == endpoint_uuid:

            return d_ep.name

    device_uuid = str(device_obj.device_id.device_uuid.uuid)

    device_name = str(device_obj.name)

    MSG = 'Device({:s},{:s})/Endpoint({:s}) not found'

    raise Exception(MSG.format(device_uuid, device_name, str(endpoint_uuid)))

@dataclass

class Ipv4Info:

    src_prefix : str = ''

    dst_prefix : str = ''

    src_port   : str = ''

    dst_port   : str = ''

    vlan       : str = ''

def extract_match_criterion_ipv4_info(match_criterion : Dict) -> Ipv4Info:

    ipv4_info = Ipv4Info()

    for type_value in match_criterion['match-type']:

        match_type = type_value['type']

        value = type_value['value'][0]

        if match_type == 'ietf-network-slice-service:source-ip-prefix':

            ipv4_info.src_prefix = value

        elif match_type == 'ietf-network-slice-service:destination-ip-prefix':

            ipv4_info.dst_prefix = value

        elif match_type == 'ietf-network-slice-service:source-tcp-port':

            ipv4_info.src_port = value

        elif match_type == 'ietf-network-slice-service:destination-tcp-port':

            ipv4_info.dst_port = value

        elif match_type == 'ietf-network-slice-service:vlan':

            ipv4_info.vlan = value

    return ipv4_info