MEC application.ipynb

    "                result[attr] = value.to_dict()\n",
    "            elif isinstance(value, dict):\n",
    "                result[attr] = dict(map(\n",
    "                    lambda item: (item[0], item[1].to_dict())\n",
    "                    if hasattr(item[1], 'to_dict') else item,\n",
    "                    value.items()\n",
    "                ))\n",
    "            else:\n",
    "                result[attr] = value\n",
    "        if issubclass(EndPointInfoAddress, dict):\n",
    "            for key, value in self.items():\n",
    "                result[key] = value\n",
    "        return result\n",
    "    def to_str(self):\n",
    "        return pprint.pformat(self.to_dict())\n",
    "    def __repr__(self):\n",
    "        return self.to_str()\n",
    "    def __eq__(self, other):\n",
    "        if not isinstance(other, EndPointInfoAddress):\n",
    "            return False\n",
    "        return self.__dict__ == other.__dict__\n",
    "    def __ne__(self, other):\n",
    "        return not self == other\n",
    "\n",
    "class EndPointInfoAddresses(object):\n",
    "    swagger_types = {'_addresses': 'list[EndPointInfoAddress]'}\n",
    "    attribute_map = {'_addresses': 'addresses'}\n",
    "    def __init__(self, addresses:list):  # noqa: E501\n",
    "        self._addresses = None\n",
    "        self.addresses = addresses\n",
    "    @property\n",
    "    def addresses(self):\n",
    "        return self._addresses\n",
    "    @addresses.setter\n",
    "    def addresses(self, addresses):\n",
    "        self._addresses = addresses\n",
    "    def to_dict(self):\n",
    "        result = {}\n",
    "        for attr, _ in six.iteritems(self.swagger_types):\n",
    "            value = getattr(self, attr)\n",
    "            if isinstance(value, list):\n",
    "                result[attr] = list(map(lambda x: x.to_dict() if hasattr(x, 'to_dict') else x,value))\n",
    "            elif hasattr(value, 'to_dict'):\n",
    "                result[attr] = value.to_dict()\n",
    "            elif isinstance(value, dict):\n",
    "                result[attr] = dict(map(\n",
    "                    lambda item: (item[0], item[1].to_dict())\n",
    "                    if hasattr(item[1], 'to_dict') else item,\n",
    "                    value.items()\n",
    "                ))\n",
    "            else:\n",
    "                result[attr] = value\n",
    "        if issubclass(EndPointInfoAddresses, dict):\n",
    "            for key, value in self.items():\n",
    "                result[key] = value\n",
    "        return result\n",
    "    def to_str(self):\n",
    "        return pprint.pformat(self.to_dict())\n",
    "    def __repr__(self):\n",
    "        return self.to_str()\n",
    "    def __eq__(self, other):\n",
    "        if not isinstance(other, EndPointInfoAddresses):\n",
    "            return False\n",
    "        return self.__dict__ == other.__dict__\n",
    "    def __ne__(self, other):\n",
    "        return not self == other\n",
    "\n",
    "class EndPointInfoAlternative(object):\n",
    "    swagger_types = {'alternative': 'object'}\n",
    "    attribute_map = {'alternative': 'alternative'}\n",
    "    def __init__(self, alternative=None):  # noqa: E501\n",
    "        self._alternative = None\n",
    "        self.discriminator = None\n",
    "        self.alternative = alternative\n",
    "    @property\n",
    "    def alternative(self):\n",
    "        return self._alternative\n",
    "    @alternative.setter\n",
    "    def alternative(self, alternative):\n",
    "        if alternative is None:\n",
    "            raise ValueError(\"Invalid value for `alternative`, must not be `None`\")  # noqa: E501\n",
    "        self._alternative = alternative\n",
    "    def to_dict(self):\n",
    "        result = {}\n",
    "        for attr, _ in six.iteritems(self.swagger_types):\n",
    "            value = getattr(self, attr)\n",
    "            if isinstance(value, list):\n",
    "                result[attr] = list(map(\n",
    "                    lambda x: x.to_dict() if hasattr(x, \"to_dict\") else x,\n",
    "                    value\n",
    "                ))\n",
    "            elif hasattr(value, \"to_dict\"):\n",
    "                result[attr] = value.to_dict()\n",
    "            elif isinstance(value, dict):\n",
    "                result[attr] = dict(map(\n",
    "                    lambda item: (item[0], item[1].to_dict())\n",
    "                    if hasattr(item[1], \"to_dict\") else item,\n",
    "                    value.items()\n",
    "                ))\n",
    "            else:\n",
    "                result[attr] = value\n",
    "        if issubclass(EndPointInfoAlternative, dict):\n",
    "            for key, value in self.items():\n",
    "                result[key] = value\n",
    "        return result\n",
    "    def to_str(self):\n",
    "        return pprint.pformat(self.to_dict())\n",
    "    def __repr__(self):\n",
    "        return self.to_str()\n",
    "    def __eq__(self, other):\n",
    "        if not isinstance(other, EndPointInfoAlternative):\n",
    "            return False\n",
    "        return self.__dict__ == other.__dict__\n",
    "    def __ne__(self, other):\n",
    "        return not self == other\n"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "The function below is creating an application MEC services\n",
    "\n",
    "**Note:** This is call application MEC service in opposition of a standardized MEC service exposed by the MEC Sanbox such as MEC 013, MEC 030...\n",
    "\n",
    "**Reference:** ETSI GS MEC 011 V3.2.1 (2024-04) Clause 8.2.6.3.4 POST"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "def create_mec_service(sandbox_name: str, app_inst_id: swagger_client.models.application_info.ApplicationInfo) -> object:\n",
    "    \"\"\"\n",
    "    Request to create a new application MEC service\n",
    "    :param sandbox_name: The MEC Sandbox instance to use\n",
    "    :param app_inst_id: The MEC application instance identifier\n",
    "    :param sub_id: The subscription identifier\n",
    "    :return The HTTP response, the response status and the headers on success, None otherwise\n",
    "    :see ETSI GS MEC 011 V3.2.1 (2024-04) Clause 8.2.6.3.4 POST\n",
    "    \"\"\"\n",
    "    global MEC_PLTF, CALLBACK_URI, logger, service_api\n",
    "\n",
    "    logger.debug('>>> create_mec_service')\n",
    "    try:\n",
    "        url = '/{sandbox_name}/{mec_pltf}/mec_service_mgmt/v1/applications/{app_inst_id}/services'\n",
    "        logger.debug('create_mec_service: url: ' + url)\n",
    "        path_params = {}\n",
    "        path_params['sandbox_name'] = sandbox_name\n",
    "        path_params['mec_pltf'] = MEC_PLTF\n",
    "        path_params['app_inst_id'] = app_inst_id.id\n",
    "        # HTTP header `Accept`\n",
    "        header_params = {}\n",
    "        header_params['Accept'] = 'application/json'  # noqa: E501\n",
    "        # HTTP header `Content-Type`\n",
    "        header_params['Content-Type'] = 'application/json'  # noqa: E501\n",
    "        # Body request\n",
    "        callback = CALLBACK_URI + '/statistic/v1/quantity'\n",
    "        transport_info = TransportInfo(id=str(uuid.uuid4()), name='HTTP REST API', type='REST_HTTP', protocol='HTTP', version='2.0', security=SecurityInfo(), endpoint=OneOfTransportInfoEndpoint())\n",
    "        transport_info.endpoint.uris=[EndPointInfoUris(callback)]\n",
    "        category_ref = CategoryRef(href=callback, id=str(uuid.uuid4()), name='Demo', version='1.0.0')\n",
    "        appServiceInfo = ServiceInfo(ser_name='demo6 MEC Service', ser_category=category_ref, version='1.0.0', state='ACTIVE',transport_info=transport_info, serializer='JSON')\n",
    "        (result, status, headers) = service_api.call_api(url, 'POST', header_params=header_params, path_params=path_params, body=appServiceInfo, async_req=False)\n",
    "        return (result, status, headers['Location'])\n",
    "    except ApiException as e:\n",
    "        logger.error('Exception when calling call_api: %s\\n' % e)\n",
    "    return None\n",
    "    # End of function create_mec_service"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "The function below is deleting an application MEC services.\n",
    "\n",
    "**Reference:** ETSI GS MEC 011 V3.2.1 (2024-04) Clause 8.2.7.3.5 DELETE"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "def delete_mec_service(resource_url: str) -> int:\n",
    "    \"\"\"\n",
    "    Request to create a new application MEC service\n",
    "    :param sandbox_name: The MEC Sandbox instance to use\n",
    "    :param app_inst_id: The MEC application instance identifier\n",
    "    :param sub_id: The subscription identifier\n",
    "    :return 0 on success, -1 otherwise\n",
    "    :see ETSI GS MEC 011 V3.2.1 (2024-04) Clause 8.2.7.3.5 DELETE\n",
    "    \"\"\"\n",
    "    global logger\n",
    "\n",
    "    logger.debug('>>> delete_mec_subscription: resource_url: ' + resource_url)\n",
    "    try:\n",
    "        res = urllib3.util.parse_url(resource_url)\n",
    "        if res is None:\n",
    "            logger.error('delete_mec_subscription: Failed to paerse URL')\n",
    "            return -1\n",
    "        header_params = {}\n",
    "        # HTTP header `Accept`\n",
    "        header_params['Accept'] = 'application/json'  # noqa: E501\n",
    "        # HTTP header `Content-Type`\n",
    "        header_params['Content-Type'] = 'application/json'  # noqa: E501\n",
    "        service_api.call_api(res.path, 'DELETE', header_params=header_params, async_req=False)\n",
    "        return 0\n",
    "    except ApiException as e:\n",
    "        logger.error('Exception when calling call_api: %s\\n' % e)\n",
    "    return -1\n",
    "    # End of function delete_mec_service"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "def process_main():\n",
    "    \"\"\"\n",
    "    This is the second sprint of our skeleton of our MEC application:\n",
    "        - Mec application setup\n",
    "        - Create new MEC service\n",
    "        - Send a request to our MEC service\n",
    "        - Delete newly created MEC service\n",
    "        - Mec application termination\n",
    "    \"\"\" \n",
    "    global LISTENER_IP, LISTENER_PORT, CALLBACK_URI, logger\n",
    "\n",
    "    logger.debug('Starting at ' + time.strftime('%Y%m%d-%H%M%S'))\n",
    "    logger.debug('\\t pwd= ' + os.getcwd())\n",
    "\n",
    "    # Start notification server in a daemonized thread\n",
    "    httpd = start_notification_server()\n",
    "\n",
    "    # Setup the MEC application\n",
    "    sandbox_name, app_inst_id, sub_id = mec_app_setup()\n",
    "\n",
    "    # Create the MEC service\n",
    "    result, status, mec_service_resource = create_mec_service(sandbox_name, app_inst_id)\n",
    "    if status != 201:\n",
    "        logger.error('Failed to create MEC service')\n",
    "    else:\n",
    "        logger.info('mec_service_resource: %s', mec_service_resource)\n",
    "\n",
    "    # Any processing here\n",
    "    logger.info('body: ' + str(result.data))\n",
    "    data = json.loads(result.data)\n",
    "    logger.info('data: %s', str(data))\n",
    "    logger.info('=============> Execute the command: curl %s/statistic/v1/quantity', CALLBACK_URI)\n",
    "    time.sleep(30)\n",
    "\n",
    "    # Stop notification server\n",
    "    stop_notification_server(httpd)\n",
    "\n",
    "    # Delete the MEC servce\n",
    "    delete_mec_service(mec_service_resource)\n",
    "\n",
    "    # Terminate the MEC application\n",
    "    mec_app_termination(sandbox_name, app_inst_id, sub_id)\n",
    "\n",
    "    logger.debug('Stopped at ' + time.strftime('%Y%m%d-%H%M%S'))\n",
    "    # End of function process_main\n",
    "\n",
    "if __name__ == '__main__':\n",
    "    process_main()\n"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "# Annexes\n",
    "\n",
    "## Annex A: How to use an existing MEC sandbox instance\n",
    "\n",
    "This case is used when the MEC Sandbox API is not used. The procedure is the following:\n",
    "- Log to the MEC Sandbox using a WEB browser\n",
    "- Select a network scenario\n",
    "- Create a new application instance\n",
    "\n",
    "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\n"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### Bibliography\n",
    "\n",
    "1. ETSI GS MEC 002 (V2.2.1) (01-2022): \"Multi-access Edge Computing (MEC); Phase 2: Use Cases and Requirements\".\n",
    "2. ETSI GS MEC 010-1 (V1.1.1) (10-2017): \"Mobile Edge Computing (MEC); Mobile Edge Management; Part 1: System, host and platform management\".\n",
    "3. ETSI GS MEC 010-2 (V2.2.1) (02-2022): \"Multi-access Edge Computing (MEC); MEC Management; Part 2: Application lifecycle, rules and requirements management\".\n",
    "4. ETSI GS MEC 011 (V3.1.1) (09-2022): \"Multi-access Edge Computing (MEC); Edge Platform Application Enablement\".\n",
    "5. ETSI GS MEC 012 (V2.2.1) (02-2022): \"Multi-access Edge Computing (MEC); Radio Network Information API\".\n",
    "6. ETSI GS MEC 013 (V2.2.1) (01-2022): \"Multi-access Edge Computing (MEC); Location API\".\n",
    "7. ETSI GS MEC 014 (V2.1.1) (03-2021): \"Multi-access Edge Computing (MEC); UE Identity API\".\n",
    "8. ETSI GS MEC 015 (V2.1.1) (06-2020): \"Multi-Access Edge Computing (MEC); Traffic Management APIs\".\n",
    "9. ETSI GS MEC 016 (V2.2.1) (04-2020): \"Multi-access Edge Computing (MEC); Device application interface\".\n",
    "10. ETSI GS MEC 021 (V2.2.1) (02-2022): \"Multi-access Edge Computing (MEC); Application Mobility Service API\".\n",
    "11. ETSI GS MEC 028 (V2.3.1) (07-2022): \"Multi-access Edge Computing (MEC); WLAN Access Information API\".\n",
    "12. ETSI GS MEC 029 (V2.2.1) (01-2022): \"Multi-access Edge Computing (MEC); Fixed Access Information API\".\n",
    "13. ETSI GS MEC 030 (V3.2.1) (05-2022): \"Multi-access Edge Computing (MEC); V2X Information Service API\".\n",
    "14. ETSI GR MEC-DEC 025 (V2.1.1) (06-2019): \"Multi-access Edge Computing (MEC); MEC Testing Framework\".\n",
    "15. ETSI GR MEC 001 (V3.1.1) (01-2022): \"Multi-access Edge Computing (MEC); Terminology\".\n",
    "16. ETSI GR MEC 003 (V3.1.1): Multi-access Edge Computing (MEC); \n",
    "Framework and Reference Architecture\n",
    "17. [The Wiki MEC web site](https://www.etsi.org/technologies/multi-access-edge-computing)\n"
   ]
  }
 ],
 "metadata": {
  "kernelspec": {
   "display_name": "Python 3 (ipykernel)",
   "language": "python",
   "name": "python3"
  },
  "language_info": {
   "codemirror_mode": {
    "name": "ipython",
    "version": 3
   },
   "file_extension": ".py",
   "mimetype": "text/x-python",
   "name": "python",
   "nbconvert_exporter": "python",
   "pygments_lexer": "ipython3",
   "version": "3.11.10"
  }
 },
 "nbformat": 4,
 "nbformat_minor": 4
}