Commit ab916217 authored by Yann Garcia's avatar Yann Garcia
Browse files

Minor bug fixes on demo9

parent 3474c470

examples/demo9/python/notebook/MEC_And_oneM2M_Tutorial.ipynb

+1 −1
Original line number Diff line number Diff line 
%% Cell type:markdown id: tags:



# How to develop an ETSI MEC application to exploit both ETSI MEC Sandbox and oneM2M platforms, based on ETSI GS MEC 046



This tutorial introduces the step by step procedure to create a basic MEC application following ETSI MEC standards and oneM2M standards to manage devices/sensors/actuators... (MEC 046).

This tutorial introduces the step by step procedure to create a basic MEC application following ETSI MEC standards and oneM2M standards to manage devices/sensors/actuators... (MEC 033, MEC 046).

It uses two platforms:

1. The ETSI MEC Sandbox and,

2. a oneM2M platform.



In this tutorial, we will use the [ACME oneM2M platform](https://acmecse.net/) with some changes to register to the ETSI MEC platform.



<div class="alert alert-block alert-danger">

    <b>Note:</b> These source code examples are simplified and ignore return codes and error checks to a large extent. We do this to highlight how to use the MEC Sandbox API and the different MEC satndards and reduce unrelated code.

A real-world application will of course properly check every return value and exit correctly at the first serious error.

</div>



%% Cell type:markdown id: tags:



Here is the list of the questions and topics to clarify from the ETSI MEC perspective to support an efficient support of oneM2M features

<div class="alert alert-block alert-danger">

    <b>Note:</b> FIXMEs

    - How to register MEC/oneM2M platforms together (currently the oneM2M platform is registering to the ETSI MEC sandbox platform)

    - How to setup the testbed for the tutorial

    - How to link oneM2M resources (AE, container, flex-container...) and SAREF onthologies with existing MEC standards

    -

</div>



%% Cell type:markdown id: tags:



## What is a ETSI MEC application



For more details on how to develop an ETSI MEC application, please refer to [The Wiki MEC web site](https://www.etsi.org/technologies/multi-access-edge-computing) and the [ETSI MEC application tutorial](http://mec-platform2.etsi.org:9999/notebooks/work/notebook/MEC%20application.ipynb).



%% Cell type:markdown id: tags:



## The basics to develop an ETSI MEC application



All the code below are helpers functions required to start the development our ETSI MEC application.

For more details, please refer to the[ETSI MEC application tutorial](http://mec-platform2.etsi.org:9999/notebooks/work/notebook/MEC%20application.ipynb).



<div class="alert alert-warning" role="alert">

    <b>Note:</b> The sub-paragraph 'Putting everything together' is a specific paragraph where all the newly features introduced in the main paragraph are put together to create an executable block of code. It is possible to skip this block of code by removing the comment character (#) on first line of this block of code.

</div>



%% Cell type:code id: tags:



``` python

import os

os.chdir(os.path.join(os.getcwd(), '../mecapp'))

print(os.getcwd())

from __future__ import division # Import floating-point division (1/4=0.25) instead of Euclidian division (1/4=0)



import os

import sys

import re

import logging

import threading

import time

import json

import uuid



import pprint



import six



import swagger_client

from swagger_client.rest import ApiException



from http import HTTPStatus

from http.server import BaseHTTPRequestHandler, HTTPServer



try:

    import urllib3

except ImportError:

    raise ImportError('Swagger python client requires urllib3.')



MEC_SANDBOX_URL     = 'https://mec-platform.etsi.org'                       # MEC Sandbox host/base URL

MEC_SANDBOX_API_URL = 'https://mec-platform.etsi.org/sandbox-api/v1'        # MEC Sandbox API host/base URL

PROVIDER            = 'Jupyter2024'                                          # Login provider value - To skip authorization: 'github'

MEC_PLTF            = 'mep1'                                                 # MEC plateform name. Linked to the network scenario

LOGGER_FORMAT       = '%(asctime)s - %(name)s - %(levelname)s - %(message)s' # Logging format

STABLE_TIME_OUT     = 10                                                     # Timer to wait for MEC Sndbox reaches its stable state (K8S pods in running state)

LOGIN_TIMEOUT       = 3 #30                                                  # Timer to wait for user to authorize from GITHUB

LISTENER_IP         = '0.0.0.0'                                              # Listener IPv4 address for notification callback calls

LISTENER_PORT       = 31111                                                  # Listener IPv4 port for notification callback calls. Default: 36001

CALLBACK_URI        = 'http://mec-platform.etsi.org:31111/sandbox/v1'

OM2M_REGISTRATION   = 30                                                     # Timer to wait for user to register sensors and devices to oneM2M platform



# Initialize the logger

logger = logging.getLogger(__name__)

logger.setLevel(logging.DEBUG)

logging.basicConfig(filename='/tmp/' + time.strftime('%Y%m%d-%H%M%S') + '.log')

l = logging.StreamHandler()

l.setFormatter(logging.Formatter('%(asctime)s - %(name)s - %(levelname)s - %(message)s'))

logger.addHandler(l)



# Setup the HTTP REST API configuration to be used to send request to MEC Sandbox API

configuration               = swagger_client.Configuration()

configuration.host          = MEC_SANDBOX_API_URL

configuration.verify_ssl    = True

configuration.debug         = True

configuration.logger_format = LOGGER_FORMAT



# Create an instance of ApiClient

sandbox_api = swagger_client.ApiClient(configuration, 'Content-Type', 'application/json')



# Setup the HTTP REST API configuration to be used to send request to MEC Services

configuration1               = swagger_client.Configuration()

configuration1.host          = MEC_SANDBOX_URL

configuration1.verify_ssl    = True

configuration1.debug         = True

configuration1.logger_format = LOGGER_FORMAT



# Create an instance of ApiClient

service_api = swagger_client.ApiClient(configuration1, 'Content-Type', 'application/json')



# Initialize the global variables

nw_scenarios     = []    # The list of available network scenarios

nw_scenario_idx  = -1    # The network scenario idx to activate (deactivate)

app_inst_id      = None  # The requested application instance identifier

got_notification = False # Set to true if a POST notification is received



def process_login() -> str:

    """

    Authenticate and create a new MEC Sandbox instance.

    :return: The sandbox instance identifier on success, None otherwise

    """

    global PROVIDER, logger



    logger.debug('>>> process_login')



    try:

        auth = swagger_client.AuthorizationApi(sandbox_api)

        oauth = auth.login(PROVIDER, async_req = False)

        logger.debug('process_login (step1): oauth: ' + str(oauth))

        # Wait for the MEC Sandbox is running

        logger.debug('=======================> DO AUTHORIZATION WITH CODE : ' + oauth.user_code)

        logger.debug('=======================> DO AUTHORIZATION HERE :      ' + oauth.verification_uri)

        if oauth.verification_uri == "":

            time.sleep(LOGIN_TIMEOUT) # Skip scecurity, wait for a few seconds

        else:

            time.sleep(10 * LOGIN_TIMEOUT) # Wait for Authirization from user side

        namespace = auth.get_namespace(oauth.user_code)

        logger.debug('process_login (step2): result: ' + str(namespace))

        return namespace.sandbox_name

    except ApiException as e:

        logger.error('Exception when calling AuthorizationApi->login: %s\n' % e)



    return None

    # End of function process_login



def process_logout(sandbox_name: str) -> int:

    """

    Delete the specified MEC Sandbox instance.

    :param sandbox_name: The MEC Sandbox to delete

    :return: 0 on success, -1 otherwise

    """

    global logger



    logger.debug('>>> process_logout: sandbox=' + sandbox_name)



    try:

        auth = swagger_client.AuthorizationApi(sandbox_api)

        result = auth.logout(sandbox_name, async_req = False)  # noqa: E501

        return 0

    except ApiException as e:

        logger.error('Exception when calling AuthorizationApi->logout: %s\n' % e)

    return -1

    # End of function process_logout





def get_network_scenarios(sandbox_name: str) -> list:

    """

    Retrieve the list of the available network scenarios.

    :param sandbox_name: The MEC Sandbox instance to use

    :return: The list of the available network scenarios on success, None otherwise

    """

    global PROVIDER, logger, sandbox_api, configuration



    logger.debug('>>> get_network_scenarios: sandbox=' + sandbox_name)



    try:

        nw = swagger_client.SandboxNetworkScenariosApi(sandbox_api)

        result = nw.sandbox_network_scenarios_get(sandbox_name, async_req = False)  # noqa: E501

        logger.debug('get_network_scenarios: result: ' + str(result))

        return result

    except ApiException as e:

        logger.error('Exception when calling SandboxNetworkScenariosApi->sandbox_network_scenarios_get: %s\n' % e)



    return None

    # End of function get_network_scenarios



def select_network_scenario_based_on_criteria(criterias_list: list) -> int:

    """

    Select the network scenario to activate based of the provided list of criterias.

    :param criterias_list: The list of criterias to select the correct network scenario

    :return: 0 on success, -1 otherwise

    :remark: The current mask is:

    [{'id': '4g-5g-macro-v2x'}, {'id': '4g-5g-macro-v2x-fed'}, {'id': '4g-5g-mc-v2x-fed-iot'}, {'id': '4g-5g-wifi-macro'}, {'id': '4g-macro'}, {'id': '4g-wifi-macro'}, {'id': 'dual-mep-4g-5g-wifi-macro'}, {'id': 'dual-mep-short-path'}]

            |                                 |                              |                             |                        |                       |                              |                                |

            ||--------------------------------                               |                             |                        |                       |                              |                                |

            ||||-------------------------------------------------------------                              |                        |                       |                              |                                |

            |||||------------------------------------------------------------------------------------------                         |                       |                              |                                |

            ||||||------------------------------------------------------------------------------------------------------------------                        |                              |                                |

            |||||||------------------------------------------------------------------------------------------------------------------------------------------------------------------------                                 |

            ||||||| --------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------

    0000000011111111b

    """

    global PROVIDER, logger



    logger.debug('>>> select_network_scenario_based_on_criteria: criterias_list=' + str(criterias_list))



    if len(criterias_list) == 0:

        return 0 # The index of the '4g-5g-macro-v2x' network scenario - Hard coded



    selector = 0xFF

    for item in criterias_list:

        if item == 'v2x':

            selector ^= 0x1F

        elif item == 'fed':

            selector ^= 0x9F

        elif item == 'iot':

            selector ^= 0xDF

        elif item == 'wifi':

            selector ^= 0xEF

        elif item == 'dual':

            selector ^= 0xFE

        # End of 'for' statement

    logger.debug('select_network_scenario_based_on_criteria: selector=' + str(selector))

    selected = 0 # The index of the '4g-5g-macro-v2x' network scenario - Hard coded

    if (selector & 0x40) == 0x40:

       selected = 1

    elif (selector & 0x20) == 0x20:

       selected = 2

    elif (selector & 0x10) == 0x10:

       selected = 3

    elif (selector & 0x08) == 0x08:

       selected = 4

    elif (selector & 0x04) == 0x04:

       selected = 5

    elif (selector & 0x02) == 0x02:

       selected = 6

    elif (selector & 0x01) == 0x01:

       selected = 7

    logger.debug('select_network_scenario_based_on_criteria: selected=' + str(selected))



    return selected

    # End of function select_network_scenario_based_on_criteria



def activate_network_scenario(sandbox_name: str) -> int:

    """

    Activate the specified network scenario.

    :param sandbox_name: The MEC Sandbox instance to use

    :return: 0 on success, -1 otherwise

    """

    global logger, sandbox_api, nw_scenarios, nw_scenario_idx



    logger.debug('>>> activate_network_scenario: ' + sandbox_name)



    nw_scenario_idx = select_network_scenario_based_on_criteria(['iot'])

    if nw_scenario_idx == -1:

        logger.error('activate_network_scenario: Failed to select a network scenarion')

        return -1



    try:

        nw = swagger_client.SandboxNetworkScenariosApi(sandbox_api)

        nw.sandbox_network_scenario_post(sandbox_name, nw_scenarios[nw_scenario_idx].id, async_req = False)  # noqa: E501

        return 0

    except ApiException as e:

        logger.error('Exception when calling SandboxNetworkScenariosApi->activate_network_scenario: %s\n' % e)



    return -1

    # End of function activate_network_scenario



def deactivate_network_scenario(sandbox: str) -> int:

    """

    Deactivate the current network scenario.

    :param sandbox: The MEC Sandbox instance to use

    :return: 0 on success, -1 otherwise

    """

    global logger, sandbox_api, nw_scenarios, nw_scenario_idx



    logger.debug('>>> deactivate_network_scenario: ' + sandbox)



    try:

        nw = swagger_client.SandboxNetworkScenariosApi(sandbox_api)

        nw.sandbox_network_scenario_delete(sandbox, nw_scenarios[nw_scenario_idx].id, async_req = False)  # noqa: E501

        return 0

    except ApiException as e:

        logger.error('Exception when calling SandboxNetworkScenariosApi->deactivate_network_scenario: %s\n' % e)



    return -1

    # End of function deactivate_network_scenario



def request_application_instance_id(sandbox_name: str) -> swagger_client.models.ApplicationInfo:

    """

    Request the creation of a new MEC application instance identifier.

    It is like the MEC application was instanciated by the MEC platform and it is executed locally.

    :param sandbox_name: The MEC Sandbox instance to use

    :return: The MEC application instance identifier on success, None otherwise

    :see ETSI GS MEC 011 V3.2.1 (2024-04) Clause 5.2.2 MEC application start-up

    """

    global MEC_PLTF, logger, sandbox_api, configuration



    logger.debug('>>> request_application_instance_id: ' + sandbox_name)



    # Create a instance of our MEC application

    try:

        a = swagger_client.models.ApplicationInfo(id=str(uuid.uuid4()), name='JupyterMecApp', node_name=MEC_PLTF, type='USER')  # noqa: E501

        nw = swagger_client.SandboxAppInstancesApi(sandbox_api)

        result = nw.sandbox_app_instances_post(a, sandbox_name, async_req = False)  # noqa: E501

        logger.debug('request_application_instance_id: result: ' + str(result))

        return result

    except ApiException as e:

        logger.error('Exception when calling SandboxAppInstancesApi->sandbox_app_instances_post: %s\n' % e)



    return None

    # End of function request_application_instance_id



def delete_application_instance_id(sandbox_name: str, app_inst_id: str) -> int:

    """

    Request the deletion of a MEC application.

    :param sandbox: The MEC Sandbox instance to use

    :param app_inst_id: The MEC application instance identifier

    :return: 0 on success, -1 otherwise

    """

    global logger, sandbox_api, configuration



    logger.debug('>>> delete_application_instance_id: ' + sandbox_name)

    logger.debug('>>> delete_application_instance_id: ' + app_inst_id)



    try:

        nw = swagger_client.SandboxAppInstancesApi(sandbox_api)

        nw.sandbox_app_instances_delete(sandbox_name, app_inst_id, async_req = False)  # noqa: E501

        return 0

    except ApiException as e:

        logger.error('Exception when calling SandboxAppInstancesApi->sandbox_app_instances_delete: %s\n' % e)



    return -1

    # End of function deletet_application_instance_id



def get_applications_list(sandbox_name: str) -> list:

    """

    Request the list of the MEC application available on the MEC Platform.

    :param sandbox: The MEC Sandbox instance to use

    :return: 0 on success, -1 otherwise

    """

    global logger, sandbox_api, configuration



    logger.debug('>>> get_applications_list: ' + sandbox_name)



    try:

        nw = swagger_client.SandboxAppInstancesApi(sandbox_api)

        result = nw.sandbox_app_instances_get(sandbox_name, async_req = False)  # noqa: E501

        logger.debug('get_applications_list: result: ' + str(result))

        return result

    except ApiException as e:

        logger.error('Exception when calling SandboxAppInstancesApi->get_applications_list: %s\n' % e)

    return None

    # End of function delete_application_instance_id



def send_ready_confirmation(sandbox_name: str, app_inst_id: swagger_client.models.application_info.ApplicationInfo) -> int:

    """

    Send the ready_confirmation to indicate that the MEC application is active.

    :param sandbox_name: The MEC Sandbox instance to use

    :param app_inst_id: The MEC application instance identifier

    :return: 0 on success, -1 otherwise

    :see ETSI GS MEC 011 V3.2.1 (2024-04) Clause 5.2.2 MEC application start-up - Step 4c

    """

    global MEC_PLTF, logger, service_api



    logger.debug('>>> send_ready_confirmation: ' + app_inst_id.id)

    try:

        url = '/{sandbox_name}/{mec_pltf}/mec_app_support/v2/applications/{app_inst_id}/confirm_ready'

        logger.debug('send_ready_confirmation: url: ' + url)

        path_params = {}

        path_params['sandbox_name'] = sandbox_name

        path_params['mec_pltf'] = MEC_PLTF

        path_params['app_inst_id'] = app_inst_id.id

        header_params = {}

        # HTTP header `Accept`

        header_params['Accept'] = 'application/json'  # noqa: E501

        # HTTP header `Content-Type`

        header_params['Content-Type'] = 'application/json'  # noqa: E501

        # JSON indication READY

        dict_body = {}

        dict_body['indication'] = 'READY'

        service_api.call_api(url, 'POST', header_params=header_params, path_params = path_params, body=dict_body, async_req=False)

        return 0

    except ApiException as e:

        logger.error('Exception when calling call_api: %s\n' % e)

    return -1

    # End of function send_ready_confirmation



def send_subscribe_termination(sandbox_name: str, app_inst_id: swagger_client.models.application_info.ApplicationInfo) -> object:

    """

    Subscribe to the MEC application termination notifications.

    :param sandbox_name: The MEC Sandbox instance to use

    :param app_inst_id: The MEC application instance identifier

    :return: The HTTP respone, the subscription ID and the resource URL on success, None otherwise

    :see ETSI GS MEC 011 V3.2.1 (2024-04) Clause 5.2.6b Receiving event notifications on MEC application instance termination

    """

    global MEC_PLTF, logger, service_api



    logger.debug('>>> send_subscribe_termination: ' + app_inst_id.id)

    try:

        url = '/{sandbox_name}/{mec_pltf}/mec_app_support/v2/applications/{app_inst_id}/subscriptions'

        logger.debug('send_subscribe_termination: url: ' + url)

        path_params = {}

        path_params['sandbox_name'] = sandbox_name

        path_params['mec_pltf'] = MEC_PLTF

        path_params['app_inst_id'] = app_inst_id.id

        header_params = {}

        # HTTP header `Accept`

        header_params['Accept'] = 'application/json'  # noqa: E501

        # HTTP header `Content-Type`

        header_params['Content-Type'] = 'application/json'  # noqa: E501

        # Body

        dict_body = {}

        dict_body['subscriptionType'] = 'AppTerminationNotificationSubscription'

        dict_body['callbackReference'] = CALLBACK_URI + '/mec011/v2/termination' # FIXME To be parameterized

        dict_body['appInstanceId'] = app_inst_id.id

        (result, status, headers) = service_api.call_api(url, 'POST', header_params=header_params, path_params = path_params, body=dict_body, async_req=False)

        return (result, extract_sub_id(headers['Location']), headers['Location'])

    except ApiException as e:

        logger.error('Exception when calling call_api: %s\n' % e)

    return (None, status, None)

    # End of function send_subscribe_termination



def extract_sub_id(resource_url: str) -> str:

    """

    Extract the subscription identifier from the specified subscription URL.

    :param resource_url: The subscription URL

    :return: The subscription identifier on success, None otherwise

    """

    global logger



    logger.debug('>>> extract_sub_id: resource_url: ' + resource_url)



    res = urllib3.util.parse_url(resource_url)

    if res is not None and res.path is not None and res.path != '':

        id = res.path.rsplit('/', 1)[-1]

        if id is not None:

            return id

    return None

    # End of function extract_sub_id



def delete_subscribe_termination(sandbox_name: str, app_inst_id: swagger_client.models.application_info.ApplicationInfo, sub_id: str) -> int:

    """

    Delete the subscrition to the AppTermination notification.

    :param sandbox_name: The MEC Sandbox instance to use

    :param app_inst_id: The MEC application instance identifier

    :param sub_id: The subscription identifier

    :return: 0 on success, -1 otherwise

    """

    global MEC_PLTF, logger, service_api



    logger.debug('>>> delete_subscribe_termination: ' + app_inst_id.id)

    try:

        url = '/{sandbox_name}/{mec_pltf}/mec_app_support/v2/applications/{app_inst_id}/subscriptions/{sub_id}'

        logger.debug('delete_subscribe_termination: url: ' + url)

        path_params = {}

        path_params['sandbox_name'] = sandbox_name

        path_params['mec_pltf'] = MEC_PLTF

        path_params['app_inst_id'] = app_inst_id.id

        path_params['sub_id'] = sub_id

        header_params = {}

        # HTTP header `Accept`

        header_params['Accept'] = 'application/json'  # noqa: E501

        # HTTP header `Content-Type`

        header_params['Content-Type'] = 'application/json'  # noqa: E501

        service_api.call_api(url, 'DELETE', header_params=header_params, path_params = path_params, async_req=False)

        return 0

    except ApiException as e:

        logger.error('Exception when calling call_api: %s\n' % e)

    return -1

    # End of function delete_subscribe_termination



def send_termination_confirmation(sandbox_name: str, app_inst_id: swagger_client.models.application_info.ApplicationInfo) -> int:

    """

    Send the confirm_termination to indicate that the MEC application is terminating gracefully.

    :param sandbox_name: The MEC Sandbox instance to use

    :param app_inst_id: The MEC application instance identifier

    :return: 0 on success, -1 otherwise

    :see ETSI GS MEC 011 V3.2.1 (2024-04) Clause 5.2.3 MEC application graceful termination/stop

    """

    global MEC_PLTF, logger, service_api



    logger.debug('>>> send_termination_confirmation: ' + app_inst_id.id)

    try:

        url = '/{sandbox_name}/{mec_pltf}/mec_app_support/v2/applications/{app_inst_id}/confirm_termination'

        logger.debug('send_termination_confirmation: url: ' + url)

        path_params = {}

        path_params['sandbox_name'] = sandbox_name

        path_params['mec_pltf'] = MEC_PLTF

        path_params['app_inst_id'] = app_inst_id.id

        header_params = {}

        # HTTP header `Accept`

        header_params['Accept'] = 'application/json'  # noqa: E501

        # HTTP header `Content-Type`

        header_params['Content-Type'] = 'application/json'  # noqa: E501

        # JSON indication READY

        dict_body = {}

        dict_body['operationAction'] = 'TERMINATING'

        service_api.call_api(url, 'POST', header_params=header_params, path_params = path_params, body=dict_body, async_req=False)

        return 0

    except ApiException as e:

        logger.error('Exception when calling call_api: %s\n' % e)

    return -1

    # End of function send_termination_confirmation



def mec_app_setup():

    """

    This function provides the steps to setup a MEC application:

        - Login

        - Print sandbox identifier

        - Print available network scenarios

        - Activate a network scenario

        - Request for a new application instance identifier

        - Send READY confirmation

        - Subscribe to AppTermination Notification

    :return The MEC Sandbox instance, the MEC application instance identifier and the subscription identifier on success, None otherwise

    """

    global logger, nw_scenarios



    # Login

    sandbox = process_login()

    if sandbox is None:

        logger.error('Failed to instanciate a MEC Sandbox')

        return None

    # Wait for the MEC Sandbox is running

    time.sleep(STABLE_TIME_OUT) # Wait for k8s pods up and running



    # Print available network scenarios

    nw_scenarios = get_network_scenarios(sandbox)

    if nw_scenarios is None:

        logger.error('Failed to retrieve the list of network scenarios')

    elif len(nw_scenarios) != 0:

        # Wait for the MEC Sandbox is running

        time.sleep(STABLE_TIME_OUT) # Wait for k8s pods up and running

    else:

        logger.info('nw_scenarios: No scenario available')



    # Activate a network scenario based on a list of criterias (hard coded!!!)

    if activate_network_scenario(sandbox) == -1:

        logger.error('Failed to activate network scenario')

    else:

        # Wait for the MEC services are running

        time.sleep(2 * STABLE_TIME_OUT) # Wait for k8s pods up and running



    # Request for a new application instance identifier

    app_inst_id = request_application_instance_id(sandbox)

    if app_inst_id == None:

        logger.error('Failed to request an application instance identifier')

    else:

        # Wait for the MEC services are terminated

        time.sleep(STABLE_TIME_OUT)



    # Send READY confirmation

    sub_id = None

    if send_ready_confirmation(sandbox, app_inst_id) == -1:

        logger.error('Failed to send confirm_ready')

    else:

        # Subscribe to AppTerminationNotificationSubscription

        result, sub_id, res_url = send_subscribe_termination(sandbox, app_inst_id)

        if sub_id == None:

            logger.error('Failed to do the subscription')



    return (sandbox, app_inst_id, sub_id)

    # End of function mec_app_setup



def mec_app_termination(sandbox_name: str, app_inst_id:swagger_client.models.ApplicationInfo, sub_id: str):

    """

    This function provides the steps to setup a MEC application:

        - Login

        - Print sandbox identifier

        - Print available network scenarios

        - Activate a network scenario

        - Request for a new application instance identifier

        - Send READY confirmation

        - Subscribe to AppTermination Notification

    :param sandbox_name: The MEC Sandbox instance to use

    :param app_inst_id: The MEC application instance identifier

    :param sub_id: The subscription identifier

    """

    global logger



    # Delete AppTerminationNotification subscription

    if sub_id is not None:

        if delete_subscribe_termination(sandbox_name, app_inst_id, sub_id) == -1:

            logger.error('Failed to delete the application instance identifier')



    # Delete the application instance identifier

    if delete_application_instance_id(sandbox_name, app_inst_id.id) == -1:

        logger.error('Failed to delete the application instance identifier')

    else:

        # Wait for the MEC services are terminated

        time.sleep(STABLE_TIME_OUT)



    # Deactivate a network scenario based on a list of criterias (hard coded!!!)

    if deactivate_network_scenario(sandbox_name) == -1:

        logger.error('Failed to deactivate network scenario')

    else:

        # Wait for the MEC services are terminated

        time.sleep(2 * STABLE_TIME_OUT)



    # Logout

    process_logout(sandbox_name)

    # End of function mec_app_termination

```



%% Cell type:markdown id: tags:



# Create our first ETSI MEC application: the skeleton of our MEC application



The purpose of this section is to create a first skeloton of our final MEC application by retrieving the oneM2M platform identifier following the procedure described in MEC 033.



## Accessing the ETSI MEC Sandbox



The cell code above include a section where the ETSI MEC Sandbox setting are already filled to make this tutorial running well (see constants MEC_SANDBOX_URL and MEC_SANDBOX_API_URL).



## Hosting the oneM2M platform



All the details about the oneM2M platform are provided during the registration to the ETSI MEC Sanbox registration process.

The tricky point is that the oneM2M platform shall address the same ETSI MEC Sandbox configured in the cell code above (see constants MEC_SANDBOX_URL and MEC_SANDBOX_API_URL).



## Wait for the oneM2M platform to register to the ETSI MEC platform



Here, we just going to poll every 5 seconds to verify if the oneM2M platform is registered to the ETSI MEC platform.

After a minutes, this function returns with than error.



%% Cell type:code id: tags:



``` python

def check_is_om2m_platform_registered(sandbox_name: str, timeout: int) -> str:

    """

    To retrieves the first regitered oneM2M platform

    :param sandbox_name: The MEC Sandbox instance to use

    :return The oneM2M platform identifier on success, an empty string otherwise

    :see ETSI GS MEC 033 V3.1.1 (2022-12) Clause 5.3.2 Registered IoT platforms query

    """

    global MEC_PLTF, logger, service_api



    logger.debug('>>> check_is_om2m_platform_registered')

    expiry = 0

    while expiry < timeout: # x seconds

        try:

            url = '/{sandbox_name}/{mec_pltf}/iots/v1/registered_iot_platforms'

            logger.debug('check_is_om2m_platform_registered: url: ' + url)

            path_params = {}

            path_params['sandbox_name'] = sandbox_name

            path_params['mec_pltf'] = MEC_PLTF

            header_params = {}

            # HTTP header `Accept`

            header_params['Accept'] = 'application/json'  # noqa: E501

            # HTTP header `Content-Type`

            header_params['Content-Type'] = 'application/json'  # noqa: E501

            (result, status, headers) = service_api.call_api(url, 'GET', header_params=header_params, path_params=path_params, async_req=False)

            if status == 200:

                # Extract the oneM2M platform identifier

                logger.info('body: ' + str(result.data))

                data = json.loads(result.data)

                logger.info('data: ' + str(data))

                if len(data) > 0:

                    return data[0]['iotPlatformId']

                else:

                    logger.error('check_is_om2m_platform_registered: Empty response\n')

            else:

                logger.error('Failed to do discover, #tries=' + str(tries))

        except ApiException as e:

            logger.error('Exception when calling call_api: %s\n' % e)



        expiry += 10

        time.sleep(10) # Wait for 10 seconds

        # End of 'while' statement

    return ""

    # End of function check_is_om2m_platform_registered

```



%% Cell type:markdown id: tags:



## Putting everything together



It is time now to create the our third iteration of our ETSI MEC application.



The sequence is the following:

- Create an ETSI MEC application instance

- Wait for a oneM2M platform to register to the ETSI MEC platform

- Display the oneM2M platform identifier

- Terminate the ETSI MEC application instance



%% Cell type:code id: tags:



``` python

%%script echo skipping

# Uncomment the line above to skip execution of this cell

def process_main():

    """

    This is the second sprint of our skeleton of our MEC application:

        - Create an ETSI MEC application instance

        - Wait for a oneM2M platform to register to the ETSI MEC platform (1 minute max.)

        - Display the oneM2M platform identifier

        - Terminate the ETSI MEC application instance

    """

    global logger, nw_scenarios



    logger.debug('Starting at ' + time.strftime('%Y%m%d-%H%M%S'))

    logger.debug('\t pwd= ' + os.getcwd())



    # Setup the MEC platform and the MEC application

    (sandbox_name, app_inst_id, sub_id) = mec_app_setup()



    # Wait for a oneM2M platform to register to the ETSI MEC platform

    om2m_pltf_id = check_is_om2m_platform_registered(sandbox_name, 240)

    if om2m_pltf_id != "":

        logger.info('om2m_pltf_id: %s', str(om2m_pltf_id))

    else:

        logger.info('Timeout')



    # Terminate the MEC application

    mec_app_termination(sandbox_name, app_inst_id, sub_id)



    logger.debug('Stopped at ' + time.strftime('%Y%m%d-%H%M%S'))

    # End of function process_main



if __name__ == '__main__':

    process_main()

```



%% Cell type:markdown id: tags:



# Second round: Discover the list of sensors and devices available



The purpose of this second application is to discover all the sencors and devices availble on the oneM2M platform.



## What is sensors and devices discovery



The role of the discovery is to retrieve the list of all sensors and devices, based on some filter criteria.

These filter criteria are (ETSI GS MEC 046 V3.1.1 (2024-04) Clause 5.3.2 Sensor Discovery Lookup):

- Type of sensor;

- A list of properties that the sensor can sense;

- A list of characteristics that sensors and device shall match;

- An geographical area that sensors and device shall match.



%% Cell type:code id: tags:



``` python

def om2m_discover_devices(sandbox_name: str, type_: str) -> dict:

    """

    This function performs a discovery of all available sensors and devices based on type

    :param sandbox_name: The MEC Sandbox instance to use

    :param type_: The MEC Sandbox instance to use

    :return The discovered sensors & devices on success, an None otherwise

    :see ETSI GS MEC 046 V3.1.1 (2024-04) Clause 5.3.2 Sensor Discovery Lookup

    :see ETSI GS MEC 046 V3.1.1 (2024-04) Clause 7.3.3.1 GET

    """

    global MEC_PLTF, logger, service_api



    try:

        url = '/{sandbox_name}/{mec_pltf}/sens/v1/queries/sensor_discovery'

        logger.debug('om2m_discover_devices: url: ' + url)

        path_params = {}

        path_params['sandbox_name'] = sandbox_name

        path_params['mec_pltf'] = MEC_PLTF

        # Query parameters

        query_params = []

        query_params.append(('type', type_))

        query_params.append(('sensorPropertyList', 'saref:any'))

        header_params = {}

        # HTTP header `Accept`

        header_params['Accept'] = 'application/json'  # noqa: E501

        # HTTP header `Content-Type`

        header_params['Content-Type'] = 'application/json'  # noqa: E501

        (result, status, headers) = service_api.call_api(url, 'GET', header_params=header_params, path_params=path_params, query_params=query_params, async_req=False)

        if status == 200:

            # Extract the oneM2M platform identifier

            logger.info('body: ' + str(result.data))

            data = json.loads(result.data)

            logger.info('data: ' + str(data))

            return data

    except ApiException as e:

        logger.error('Exception when calling call_api: %s\n' % e)

        return None

    # End of function om2m_discover_devices

```



%% Cell type:markdown id: tags:



## Preconditions



To be efficient, some sensors and devices shall be registered on the oneM2M platform.

Here is a little project simulating a basic smart garden based on ESP32. To use it, you do not need any material. This project uses the well known Wokwi playground to simulate the ESP32. It provides you a free license to play with it and this is enough for our needs.



You are just required to use Visual Code Studio or Cursor IDE with PlatformIO extention installed. Next, follow the instructions from the README file to build, configure and execute this project.



The references below will point you to the project code

- The [ESP32 project](https://gitlab.com/garcia.yann/lolin32esp32project-01.git)

- The [wokwi web site](https://wokwi.com/)



Assuming this ESP32 project is deployed, the preconditions() function below is just (for 30 seconds) waiting for you to execute it for 30 seconds.



%% Cell type:code id: tags:



``` python

def preconditions():

    """

    This function performs a discovery of all available sensors and devices based on type

    """

    global OM2M_REGISTRATION



    logger.debug('=======================> Please run script to register sensors & devices')

    time.sleep(OM2M_REGISTRATION) # Wait

    # End of function preconditions

```



%% Cell type:markdown id: tags:



## Putting everything together

It is time now to create the our third iteration of our MEC application.



The sequence is the followin

- Create an ETSI MEC application instance

- Wait for a oneM2M platform to register to the ETSI MEC platform (1 minute max.)

- Discover the list of sensors and devices of type

- Display the result of the discovery

- Terminate the ETSI MEC application instance

g:



%% Cell type:code id: tags:



``` python

%%script echo skipping

# Uncomment the line above to skip execution of this cell

def process_main():

    """

    This is the second sprint of our skeleton of our MEC application:

        - Create an ETSI MEC application instance

        - Wait for a oneM2M platform to register to the ETSI MEC platform (1 minute max.)

        - Discover the list of sensors and devices of type

        - Display the result of the discovery

        - Terminate the ETSI MEC application instance

    """

    global logger, nw_scenarios



    logger.debug('Starting at ' + time.strftime('%Y%m%d-%H%M%S'))

    logger.debug('\t pwd= ' + os.getcwd())



    # Setup the MEC application

    (sandbox_name, app_inst_id, sub_id) = mec_app_setup()



    # Wait for a oneM2M platform to register to the ETSI MEC platform

    om2m_pltf_id = check_is_om2m_platform_registered(sandbox_name, 240)

    if om2m_pltf_id != "":

        logger.info('om2m_pltf_id: %s', str(om2m_pltf_id))

    else:

        logger.warning('Timeout')



    # Ask user to register sensors & devices to the oneM2M platform

    preconditions()



    # Do the discovery

    tries = 1

    sensors = om2m_discover_devices(sandbox_name, 'CNT')

    while tries < 5 and sensors is None:

        tries += 1

        logger.error('Failed to do discover, #tries=' + str(tries))

        time.sleep(10) # wait for x seconds

        sensors = om2m_discover_devices(sandbox_name, 'CNT')

    # End of 'while' statement

    if not sensors is None:

        logger.info('oneM2M sensors and devices discovery: ' + str(sensors))

    else:

        logger.error('Failed to do discover')



    # Terminate the MEC application

    mec_app_termination(sandbox_name, app_inst_id, sub_id)



    logger.debug('Stopped at ' + time.strftime('%Y%m%d-%H%M%S'))

    # End of function process_main



if __name__ == '__main__':

    process_main()

```



%% Cell type:markdown id: tags:



# Third round: Retrieve information of a specific sensor/device



We are now able to develop an ETSI MEC application which is able to discover a list of sensors/devices based on different criteria.

The next step is to retrieve information of a specific sensor we discovered.



## Selecting a sensors/devices



Having a list of sensors/devices, the next step is to select one of them to retrieve information about it.



In this section, we wl iluse a function to do this selection based on criterao



**Note:** For the time being, this function returns the first sensor/device in the list. It would be a good exercise to refine this function ;).



%% Cell type:code id: tags:



``` python

def select_sensors_based_on_criteria(sensors: list, criterias: dict) -> str:

    """

    Select a sensor/device based of the provided list of criterias.

    :param criterias_list: The list of criterias to select the correct sensor

                            - sensorType

    :return: The sensor/device identifier on success, an empty string otherwise

    """

    global logger



    # Set default values

    type_ = 'FLX'

    # TODO To be continued



    # Update default values

    if not criterias is None and len(criterias) > 0:

        if 'sensorType' in criterias.keys():

            type_ = criterias['sensorType']

        # TODO To be continued



    # Let's find the sensor matching criteria

    for item in sensors:

        #logger.debug('select_sensors_based_on_criteria: Processing item: ' + str(item))

        if item['sensorType'] != type_:

            continue

        return item['sensorIdentifier']

    # End of 'for' statement

    return ''

    # End of function select_sensors_based_on_criteria

```



%% Cell type:markdown id: tags:



## Retrieving information of a specific sensor



The purpose here is the to retrieve sensor data desscription based on ETSI GS MEC 046 V3.1.1 (2024-04) Clause 5.3.6 Sensor Data Lookup



%% Cell type:code id: tags:



``` python

def om2m_get_sensor_data(sandbox_name: str, sensor_id: str) -> dict:

    """

    This function retrieves the information of a specific sensor/device

    :param sandbox_name: The MEC Sandbox instance to use

    :param sensor_id: The sensor identifier

    :return The discovered sensors & devices on success, an None otherwise

    :see ETSI GS MEC 046 V3.1.1 (2024-04) Clause 5.3.6 Sensor Data Lookup

    :see ETSI GS MEC 046 V3.1.1 (2024-04) Clause 7.9.3.1 GET

    """

    global MEC_PLTF, logger, service_api



    try:

        url = '/{sandbox_name}/{mec_pltf}/sens/v1/queries/sensor_data'

        logger.debug('om2m_discover_devices: url: ' + url)

        path_params = {}

        path_params['sandbox_name'] = sandbox_name

        path_params['mec_pltf'] = MEC_PLTF

        # Query parameters

        query_params = []

        query_params.append(('sensorIdentifier', sensor_id))

        header_params = {}

        # HTTP header `Accept`

        header_params['Accept'] = 'application/json'  # noqa: E501

        # HTTP header `Content-Type`

        header_params['Content-Type'] = 'application/json'  # noqa: E501

        (result, status, headers) = service_api.call_api(url, 'GET', header_params=header_params, path_params=path_params, query_params=query_params, async_req=False)

        if status == 200:

            # Extract the oneM2M platform identifier

            logger.info('body: ' + str(result.data))

            data = json.loads(result.data)

            logger.info('data: ' + str(data))

            return data

    except ApiException as e:

        logger.error('Exception when calling call_api: %s\n' % e)

        return None

    # End of function om2m_get_sensor_data

```



%% Cell type:markdown id: tags:



## Putting everything together

It is time now to create the our third iteration of our MEC application.



The sequence is the following:

- Create an ETSI MEC application instance

- Wait for a oneM2M platform to register to the ETSI MEC platform (1 minute max.)

- Discover the list of sensors and devices of type

- Select a specific sensor/device

- Retrive its SensorData descritpion (ETSI GS MEC 046 V3.1.1 (2024-04) Clause 6.2.3 Type: SensorData)

- Display the result of the discovery

- Terminate the ETSI MEC application instance



%% Cell type:code id: tags:



``` python

#%%script echo skipping

# Uncomment the line above to skip execution of this cell

def process_main():

    """

    This is the second sprint of our skeleton of our MEC application:

        - Create an ETSI MEC application instance

        - Wait for a oneM2M platform to register to the ETSI MEC platform (1 minute max.)

        - Discover the list of sensors and devices of type

        - Display the result of the discovery

        - Terminate the ETSI MEC application instance

    """

    global logger, nw_scenarios



    logger.debug('Starting at ' + time.strftime('%Y%m%d-%H%M%S'))

    logger.debug('\t pwd= ' + os.getcwd())



    # Setup the MEC application

    (sandbox_name, app_inst_id, sub_id) = mec_app_setup()



    # Wait for a oneM2M platform to register to the ETSI MEC platform

    om2m_pltf_id = check_is_om2m_platform_registered(sandbox_name, 240)

    if om2m_pltf_id != "":

        logger.info('om2m_pltf_id: %s', str(om2m_pltf_id))

    else:

        logger.warning('Timeout')



    # Ask user to register sensors & devices to the oneM2M platform

    preconditions()



    # Do the discovery

    tries = 1

    sensors = om2m_discover_devices(sandbox_name, 'CNT')

    while tries < 3 and sensors is None:

        tries += 1

        logger.error('Failed to do discover, #tries=' + str(tries))

        time.sleep(15) # wait for 10s

        sensors = om2m_discover_devices(sandbox_name, 'CNT')

    # End of 'while' statement

    if not sensors is None:

        logger.info('oneM2M sensors and devices discovery: ' + str(sensors))

        sensor_id = select_sensors_based_on_criteria(sensors, None)

        if sensor_id != '':

            info = om2m_get_sensor_data(sandbox_name, sensor_id)

            if not info is None:

                desc = next((item for item in sensors if item['sensorIdentifier'] == sensor_id), None)

                logger.info('oneM2M sensor descr: ' + str(desc))

                logger.info('oneM2M sensor info: ' + str(info))

            else:

                logger.error('Failed to do retrieve info for ' + sensor_id)

        else:

            logger.error('Failed to find a sensor/device matching criteria')

    else:

        logger.error('Failed to do discover')



    # Terminate the MEC application

    mec_app_termination(sandbox_name, app_inst_id, sub_id)



    logger.debug('Stopped at ' + time.strftime('%Y%m%d-%H%M%S'))

    # End of function process_main



if __name__ == '__main__':

    process_main()

```



%% Cell type:markdown id: tags:



# Fourth round: How to use subscriptions



The purpose here is to use subscription on sensors/devices to be notified when a event occurs, such as an update of the sensor/device, a new measurement...

According to ETSI GS MEC 046 V3.1.1 (2024-04) Clause 5.3 Sequence diagrams, there are several kind of subscription:

- Sensor discovery subscription (clause 5.3.3 Sensor Discovery Subscription);

- Sensor Status subscription (clause 5.3.5 Sensor Status Subscription);

- Sensor Data subscription (clause 5.3.7 Sensor Data Subscription).



In any case, the mechanism is the same:

1. Our MEC application shall create a subscription request to get a subscription identifier

2. Our MEC application receives notification (see next section below)

3. Our Our MEC application cancel the subscription when terminating



The cell below provides the code to create our subscription.



%% Cell type:markdown id: tags:



#### Notification support



To recieve notification, our MEC application is required to support an HTTP listenener to recieve POST requests from the MEC Sandbox and reply to them: this is the notification mechanism.



The class HTTPRequestHandler (see cell below) provides the suport of such mechanism.



%% Cell type:code id: tags:



``` python

class HTTPServer_RequestHandler(BaseHTTPRequestHandler):

    """

    Minimal implementation of an HTTP server (http only).

    """



    def do_GET(self):

        logger.info('>>> do_GET: ' + self.path)



        ctype = self.headers.get('content-type')

        logger.info('do_GET: ' + ctype)



        message = ''

        if self.path == '/sandbox/v1/statistic/v1/quantity':

            logger.info('do_GET: Computing statistic quantities for application MEC service')

            # TODO Add logit to our MEC service

            message = '{"time":20180124,"avg": 0.0,"max": 0.0,"min": 0.0,"stddev": 0.0 }'

        else:

             # Send error message

            message = '{"title":"Unknown URI","type":"do_GET.parser","status":404 }'

        logger.info('do_GET: message: ' + message)



        # Send response status code

        self.send_response(HTTPStatus.OK)



        # Send headers

        self.send_header('Content-type','text/plain; charset=utf-8')

        self.send_header('Content-length', str(len(message)))

        self.end_headers()



        # Write content as utf-8 data

        self.wfile.write(message.encode('utf8'))

        return

        # End of function do_GET



    def do_POST(self):

        global got_notification



        logger.info('>>> do_POST: ' + self.path)



        ctype = self.headers.get('content-type')

        logger.info('do_POST: ' + ctype)



        content_len = int(self.headers.get('Content-Length'))

        if content_len != 0:

            body = self.rfile.read(content_len).decode('utf8')

            logger.info('do_POST: body:' + str(type(body)))

            logger.info('do_POST: body:' + str(body))

            data = json.loads(str(body))

            logger.info('do_POST: data: %s', str(data))



        self.send_response(HTTPStatus.NOT_IMPLEMENTED)

        self.end_headers()

        got_notification = True

        return

        # End of function do_POST



    def do_PUT(self):

        logger.info('>>> do_PUT: ' + self.path)



        ctype = self.headers.get('content-type')

        logger.info('do_PUT: ' + ctype)



        self.send_response(HTTPStatus.NOT_IMPLEMENTED)

        self.end_headers()

        return

        # End of function do_PUT



    def do_PATCH(self):

        logger.info('>>> do_PATCH: ' + self.path)



        ctype = self.headers.get('content-type')

        logger.info('do_PATCH: ' + ctype)



        self.send_response(HTTPStatus.NOT_IMPLEMENTED)

        self.end_headers()

        return

        # End of function do_PATCH

    # End of class HTTPRequestHandler



    def do_DELETE(self):

        logger.info('>>> do_DELETE: ' + self.path)



        ctype = self.headers.get('content-type')

        logger.info('do_DELETE: ' + ctype)



        self.send_response(HTTPStatus.NOT_IMPLEMENTED)

        self.end_headers()

        return

        # End of function do_DELETE

    # End of class HTTPRequestHandler



def start_notification_server() -> HTTPServer:

    """

    Start the notification server

    :return The instance of the HTTP server

    """

    global LISTENER_PORT, got_notification, logger



    logger.debug('>>> start_notification_server on port ' + str(LISTENER_PORT))

    got_notification = False

    server_address = ('', LISTENER_PORT)

    httpd = HTTPServer(server_address, HTTPServer_RequestHandler)

    # Start notification server in a daemonized thread

    notification_server = threading.Thread(target = httpd.serve_forever, name='notification_server')

    notification_server.daemon = True

    notification_server.start()

    return httpd

    # End of function HTTPRequestHandler



def stop_notification_server(httpd: HTTPServer):

    """

    Stop the notification server

    :param The instance of the HTTP server

    """

    global logger



    logger.debug('>>> stop_notification_server')

    httpd.server_close()

    httpd=None

    # End of function HTTPRequestHandler

```



%% Cell type:code id: tags:



``` python

def subscribe_for_user_events(sandbox_name: str) -> object:

    """

    Subscriptions for notifications related to location.

    :param sandbox_name: The MEC Sandbox instance to use

    :return: The HTTP respone, the subscription ID and the resource URL on success, None otherwise

    :see ETSI GS MEC 013 V3.1.1 Clause 7.5 Resource: user_subscriptions

    """

    global MEC_PLTF, logger, service_api



    logger.debug('>>> subscribe_for_user_events: ' + sandbox_name)

    try:

        url = '/{sandbox_name}/{mec_pltf}//location/v3/subscriptions/users'

        logger.debug('subscribe_for_user_events: url: ' + url)

        path_params = {}

        path_params['sandbox_name'] = sandbox_name

        path_params['mec_pltf'] = MEC_PLTF

        header_params = {}

        # HTTP header `Accept`

        header_params['Accept'] = 'application/json'  # noqa: E501

        # HTTP header `Content-Type`

        header_params['Content-Type'] = 'application/json'  # noqa: E501

        # Body

        dict_body = {}

        dict_body['subscriptionType'] = 'UserLocationEventSubscription'

        dict_body['callbackReference'] = CALLBACK_URI + '/mec013/v3/location' # FIXME To be parameterized

        dict_body['address'] = '10.100.0.1' # FIXME To be parameterized

        dict_body['clientCorrelator'] = "12345"

        dict_body['locationEventCriteria'] = [ "ENTERING_AREA_EVENT", "LEAVING_AREA_EVENT"]

        m = {}

        m["userLocationEventSubscription"] = dict_body

        (result, status, headers) = service_api.call_api(url, 'POST', header_params=header_params, path_params = path_params, body=m, async_req=False)

        return (result, extract_sub_id(headers['Location']), headers['Location'])

    except ApiException as e:

        logger.error('Exception when calling call_api: %s\n' % e)

    return (None, None, None)

    # End of function subscribe_for_user_are_event

```



%% Cell type:markdown id: tags:



### Putting everything together



It is time now to create the our third iteration of our MEC application.



The sequence is the following:

- Login

- Activate a network scenario

- Create subscription

- Wait for notification

- Delete our application instance identifier

- Deactivate a network scenario

- Logout



%% Cell type:code id: tags:



``` python

%%script echo skipping

# Uncomment the line above to skip execution of this cell

def process_main():

    """

    This is the second sprint of our skeleton of our MEC application:

        - Login

        - Activate a network scenario

        - Create subscription

        - Wait for notification

        - Delete our application instance identifier

        - Deactivate a network scenario

        - Logout

    """

    global logger, nw_scenarios, got_notification



    logger.debug('Starting at ' + time.strftime('%Y%m%d-%H%M%S'))

    logger.debug('\t pwd= ' + os.getcwd())



    # Start notification server in a daemonized thread

    httpd = start_notification_server()



    # Setup the MEC application

    (sandbox_name, app_inst_id, sub_id) = mec_app_setup()



    # Create subscription

    result, status, headers = subscribe_for_user_events(sandbox_name)

    logger.info('UE location information: status: %s', str(status))

    if status != 200:

        logger.error('Failed to get UE location information')

    else:

        logger.info('UE location information: ' + str(result.data))



    # Getting UE location lookup

    result, status = get_ue_location(sandbox_name, '10.100.0.1')

    logger.info('UE location information: status: ' + str(status))

    if status != 200:

        logger.error('Failed to get UE location information')

    else:

        logger.info('UE location information: ' + str(result.data))



    # Wait for the notification

    counter = 0

    while not got_notification and counter < 30:

        logger.info('Waiting for subscription...')

        time.sleep(STABLE_TIME_OUT)

        counter += 1

        # End of 'while' statement



    # Stop notification server

    stop_notification_server(httpd)



    # Terminate the MEC application

    mec_app_termination(sandbox_name, app_inst_id, sub_id)



    logger.debug('Stopped at ' + time.strftime('%Y%m%d-%H%M%S'))

    # End of function process_main



if __name__ == '__main__':

    process_main()

```



%% Cell type:markdown id: tags:



# Fith round: Update sensor value



%% Cell type:markdown id: tags:



# Annexes



## Annex A: How to use an existing MEC sandbox instance



This case is used when the MEC Sandbox API is not used. The procedure is the following:

- Log to the MEC Sandbox using a WEB browser

- Select a network scenario

- Create a new application instance



When it is done, the newly created application instance is used by your application when required. This application instance is usually passed to your application in the command line or using a configuration file



%% Cell type:markdown id: tags:



### Bibliography



1. ETSI GS MEC 002 (V2.2.1) (01-2022): "Multi-access Edge Computing (MEC); Phase 2: Use Cases and Requirements".

2. ETSI GS MEC 033 V3.1.1 (2022-12): "Multi-access Edge Computing (MEC); IoT APII".

3. ETSI GS MEC 046V3.1.1 (2024-04): "Multi-access Edge Computing (MEC); Sensor-sharing API".s

go-apps/meep-iot-pltf/meep-acme-in-cse/entrypoint.sh

+18 −17
Original line number Diff line number Diff line 
#!/bin/bash

set -e



echo "MEEP_HOST_URL: ${MEEP_HOST_URL}"

SERVER_IP="${MEEP_HOST_URL#http://}"; MEEP_HOST_URL="${MEEP_HOST_URL#https://}"



echo "MEEP_HOST_URL: ${MEEP_HOST_URL}"

echo "MEC_PLATFORM=$MEC_PLATFORM"

echo "MEEP_SANDBOX_NAME=$MEEP_SANDBOX_NAME"
@@ -20,21 +20,22 @@ echo "MQTT_PORT: ${MQTT_PORT}" 
echo "MQTT_USERNAME: ${MQTT_USERNAME}"

echo "MQTT_PASSWORD: ${MQTT_PASSWORD}"



#SERVICE_NAME="meep-mosquitto"

#NAMESPACE=$(cat /var/run/secrets/kubernetes.io/serviceaccount/namespace)

#TOKEN=$(cat /var/run/secrets/kubernetes.io/serviceaccount/token)

#MOSQUITTO_NODE_PORT=$(curl -sSk \

#  -H "Authorization: Bearer $TOKEN" \

#  https://kubernetes.default.svc/api/v1/namespaces/$NAMESPACE/services/$SERVICE_NAME \

#  | jq -r '.spec.ports[0].nodePort')

#echo "External NodePort exposed for service [$SERVICE_NAME] in namespace [$NAMESPACE] is: $MOSQUITTO_NODE_PORT"



#SERVICE_NAME="meep-acme-in-cse"

#NODE_PORT=$(curl -sSk \

#  -H "Authorization: Bearer $TOKEN" \

#  https://kubernetes.default.svc/api/v1/namespaces/$NAMESPACE/services/$SERVICE_NAME \

#  | jq -r '.spec.ports[0].nodePort')

#echo "External NodePort exposed for service [$SERVICE_NAME] in namespace [$NAMESPACE] is: $NODE_PORT"

# Retrieve the internal meep-mosquitto IP address

SERVICE_NAME="meep-cloud-mosquitto"

NAMESPACE=$(cat /var/run/secrets/kubernetes.io/serviceaccount/namespace)

TOKEN=$(cat /var/run/secrets/kubernetes.io/serviceaccount/token)

MOSQUITTO_NODE_IP_ADDRESS=$(curl -sSk \

  -H "Authorization: Bearer $TOKEN" \

  https://kubernetes.default.svc/api/v1/namespaces/$NAMESPACE/services/$SERVICE_NAME \

  | jq -r '.spec.clusterIP')

echo "Internal IP exposed for service [$SERVICE_NAME] in namespace [$NAMESPACE] is: $MOSQUITTO_NODE_IP_ADDRESS"



# Retrieve the internal meep-mosquitto port id

MOSQUITTO_NODE_PORT=$(curl -sSk \

  -H "Authorization: Bearer $TOKEN" \

  https://kubernetes.default.svc/api/v1/namespaces/$NAMESPACE/services/$SERVICE_NAME \

  | jq -r '.spec.ports[0].targetPort')

echo "Internal NodePort exposed for service [$SERVICE_NAME] in namespace [$NAMESPACE] is: $MOSQUITTO_NODE_PORT"



if [[ ! -z "${MEEP_MEP_NAME}" ]]; then

    svcPath="${MEEP_SANDBOX_NAME}/${MEEP_MEP_NAME}"
@@ -62,7 +63,7 @@ ENABLE_WS=${ENABLE_WS:-false} 
# MQTT Configuration

MQTT_ENABLE=${MQTT_ENABLE:-true}

MQTT_HOST=${MOSQUITTO_NODE_IP_ADDRESS:-"meep-cloud-mosquito"}

MQTT_PORT=${MOSQUITTO_NODE_PORT:-1883}

MQTT_PORT=${MOSQUITTO_NODE_PORT:-443}

MQTT_USERNAME=${MQTT_USERNAME:-"acme-mn-cse"}

MQTT_PASSWORD=${MQTT_PASSWORD:-"mqtt"}

go-apps/meep-iot/server/meep-iot.go

+2 −2
Original line number Diff line number Diff line
@@ -952,13 +952,13 @@ func registereddevicesPOST(w http.ResponseWriter, r *http.Request) { 
	log.Info("registereddevicesPOST: ", deviceInfo)



	// Sanity checks

	if len(deviceInfo.DeviceId) == 0 {

	if deviceInfo.DeviceId == "" {

		err = errors.New("DeviceId field shall be present")

		errHandlerProblemDetails(w, err.Error(), http.StatusBadRequest)

		return

	}

	if len(deviceInfo.RequestedMecTrafficRule) == 0 {

		if len(deviceInfo.RequestedIotPlatformId) == 0 && len(deviceInfo.RequestedUserTransportId) == 0 {

		if deviceInfo.RequestedIotPlatformId == "" && deviceInfo.RequestedUserTransportId == "" {

			err = errors.New("Invalid traffic rule provided")

			errHandlerProblemDetails(w, err.Error(), http.StatusBadRequest)

			return

go-packages/meep-iot-mgr/iot-mgr.go

+34 −17
Original line number Diff line number Diff line
@@ -538,38 +538,55 @@ func (tm *IotMgr) createDeviceWithIotPlatformId(device DeviceInfo, requestedIotP 
		return deviceResp, err

	}

	if _, ok := registeredIotPlatformsAEMap[requestedIotPlatformId]; !ok {

		err = errors.New("Devaice cannot be created witout oneM2M AE parent")

		err = errors.New("Devaice cannot be created without oneM2M AE parent")

		return deviceResp, err

	}



	if registeredIotPlatformsMap[requestedIotPlatformId].oneM2M != nil && device.Enabled == true {

	// FIXME FSCOM How to manage these fields from DeviceInfo

	if registeredIotPlatformsMap[requestedIotPlatformId].oneM2M != nil && device.Enabled {

		log.Info("createDeviceWithIotPlatformId: Create device on IoT platform", device)

		var sensor = sssmgr.SensorDiscoveryInfo{

			SensorIdentifier: device.DeviceId,

			SensorType:       "CNT", // FIXME FSCOM How to retrieve this info

			SensorType:       "FLX", // FIXME FSCOM How to retrieve this info

			SensorPosition:   nil,

			IotPlatformId:    requestedIotPlatformId,

		}

		sensor.Flex = make(map[string]interface{}, 0)

		sensor.Flex["type"] = "mec:iotDev"

		sensor.Flex["cnd"] = "org.onem2m.common.mec.device.iotDevice"

		sensor.Flex["devId"] = device.DeviceId

		sensor.Flex["rIPId"] = requestedIotPlatformId

		sensor.Flex["enabd"] = device.Enabled

		if device.Gpsi != "" {

			sensor.Flex["gpsi"] = device.Gpsi

		}

		if device.Pei != "" {

			sensor.Flex["pei"] = device.Pei

		}

		if device.Supi != "" {

			sensor.Flex["supi"] = device.Supi

		}

		if device.Msisdn != "" {

			sensor.Flex["msisdn"] = device.Msisdn

		}

		if device.Imei != "" {

			sensor.Flex["imei"] = device.Imei

		}

		if device.Imsi != "" {

			sensor.Flex["imsi"] = device.Imsi

		}

		if device.Iccid != "" {

			sensor.Flex["iccid"] = device.Iccid

		}



		if len(device.DeviceMetadata) != 0 {

			sensor.SensorCharacteristicList = make([]sssmgr.SensorCharacteristic, len(device.DeviceMetadata))

			for i, c := range device.DeviceMetadata {

				sensor.SensorCharacteristicList[i] = sssmgr.SensorCharacteristic{CharacteristicName: c.Key, CharacteristicValue: c.Value}

			} // End of 'for' statement

		}

		// FIXME FSCOM How to manage these fields from DeviceInfo

		// 	DeviceAuthenticationInfo string

		// 	Gpsi                     string

		// 	Pei                      string

		// 	Supi                     string

		// 	Msisdn                   string

		// 	Imei                     string

		// 	Imsi                     string

		// 	Iccid                    string

		// 	RequestedMecTrafficRule  []TrafficRuleDescriptor

		// 	//DeviceSpecificMessageFormats *DeviceSpecificMessageFormats

		// 	//DownlinkInfo *DownlinkInfo

		// 	ClientCertificate string

		// }

		log.Info("createDeviceWithIotPlatformId: Call OneM2M_create: ", sensor)

		log.Info("createDeviceWithIotPlatformId: registeredIotPlatformsMap: ", registeredIotPlatformsMap)

		sensor, err := registeredIotPlatformsMap[requestedIotPlatformId].oneM2M.OneM2M_create(sensor, registeredIotPlatformsAEMap[requestedIotPlatformId].SensorIdentifier)

		if err != nil {

			return deviceResp, err