first major commit of MCP module (8756ba41) · Commits · OOP / code / ai2

MCP_module/Dockerfile

0 → 100644

+0 −0

Empty file added.

MCP_module/Dummy_backend/Dockerfile

0 → 100644

+0 −0

Empty file added.

MCP_module/Dummy_backend/README.md

0 → 100644

+103 −0

Original line number	Diff line number	Diff line

		---

		# Dummy CAMARA Backend


		## Overview

		This repository contains a dummy backend implementation of CAMARA API endpoints, used for testing, experimentation or integration demos.

		---

		* Uses Connexion framework to automatically handle routing based on the provided OpenAPI spec (`openapi.yaml`).
		* Organizes logic into a modular `controllers/` directory for clean separation of API endpoints.
		* Provides automatic Swagger UI documentation for easy testing of API endpoints.
		* Loads configuration parameters from `.env` file.

		## 📁 Project Structure

		```
		Dummy_backend/

		├── controllers/
		\| \| experimental/
		│ │ └── qod_controller.py
		│ │ └── ....
		│ ├── v1/
		│ │ └── example_controller.py
		│ ├── v2/
		│ │ └── example_controller.py
		\| \| └── example_controller.py
		├── openapi.yaml
		├── app.py
		├── .env
		├── requirements.txt
		└── Dockerfile

		```

		---

		## ⚙️ Installation

		### Option 1

		Follow these steps to set up and run the API locally.


		```
		git clone https://labs.etsi.org/rep/oop/code/ai2.git
		cd MCP_module/Dummy_backend
		python3 -m venv venv (Tested with python versions 3.10x and 3.12x)
		source venv/bin/activate # On Windows: venv\Scripts\activate
		pip install -r requirements.txt

		Create a `.env` file in the project root (if not already present) and set your API configuration:
		API_HOST=
		API_PORT=
		```


		## 🚀 Usage

		### Option 1
		```
		python app.py
		```


		Once the server is running, open your browser and visit:

		* Swagger UI: [http://127.0.0.1:8081/ui/](http://127.0.0.1:8081/ui/)
		to explore and test the endpoints interactively.
		* The url may very in case of the corresponding environment variables

		---

		## 🤝 Contributing

		To contribute:

		1. Fork the repository.
		2. Create a new branch for your feature or fix:

		```
		git checkout -b feature/my-feature
		```
		3. Commit your changes:

		```
		git commit -m "Add new endpoint for dummy CAMARA data"
		```
		4. Push your branch and open a Pull Request.

		Please make sure to:

		* Update or add OpenAPI definitions in `openapi.yaml` with additional API end-points.
		* Create a new folder under `controllers/` (e.g., controllers/v3/).
		* Point each `operationId` to the new controller module.

		---

MCP_module/Dummy_backend/app.py

0 → 100644

+21 −0

Original line number	Diff line number	Diff line
		import os
		import connexion
		from dotenv import load_dotenv


		load_dotenv()
		# Load environment variables with defaults

		HOST = os.getenv("API_HOST", "127.0.0.1")
		PORT = int(os.getenv("API_PORT", 8081))

		# Create Connexion app
		app = connexion.App(__name__, specification_dir="./")
		app.add_api("openapi.yaml", strict_validation=True)

		if __name__ == "__main__":
		print("Starting API server...")
		print(f"Swagger UI available at: http://{HOST}:{PORT}/ui/")
		# print(f"OpenAPI JSON available at: http://{HOST}:{PORT}/openapi.json")
		app.run(host=HOST, port=PORT)

MCP_module/Dummy_backend/controllers/experimental/qod_controller.py

0 → 100644

+174 −0

Original line number	Diff line number	Diff line
		from datetime import datetime, timedelta
		from typing import Dict, Any
		import uuid

		# In-memory session storage
		sessions_db: Dict[str, Dict[str, Any]] = {}

		class QoSStatus:
		REQUESTED = "REQUESTED"
		AVAILABLE = "AVAILABLE"
		UNAVAILABLE = "UNAVAILABLE"

		def create_session(body: Dict[str, Any]) -> tuple:
		"""
		POST /sessions
		Create a new QoS session
		"""
		try:
		session_id = str(uuid.uuid4())
		device = body.get("device", {})
		app_server = body.get("applicationServer", {})
		qos_profile = body.get("qosProfile", "QOS_L")
		sink = body.get("sink")
		duration = body.get("duration", 3600)

		# Validate required fields
		if not app_server or not sink:
		return {
		"status": 422,
		"code": "MISSING_FIELD",
		"message": "applicationServer and sink are required"
		}, 422

		now = datetime.utcnow()
		expires_at = now + timedelta(seconds=duration)

		session = {
		"sessionId": session_id,
		"device": device,
		"applicationServer": app_server,
		"devicePorts": body.get("devicePorts", {"ranges": [], "ports": []}),
		"applicationServerPorts": body.get("applicationServerPorts", {"ranges": [], "ports": []}),
		"qosProfile": qos_profile,
		"sink": sink,
		"sinkCredential": body.get("sinkCredential", {}),
		"duration": duration,
		"qosStatus": QoSStatus.REQUESTED,
		"createdAt": now.isoformat() + "Z",
		"startedAt": now.isoformat() + "Z",
		"expiresAt": expires_at.isoformat() + "Z",
		}

		sessions_db[session_id] = session

		# Response excludes device
		response = {
		"sessionId": session_id,
		"applicationServer": app_server,
		"qosProfile": qos_profile,
		"sink": sink,
		"duration": duration,
		"qosStatus": QoSStatus.REQUESTED
		}

		return response, 201

		except Exception as e:
		return {
		"status": 400,
		"code": "INVALID_ARGUMENT",
		"message": f"Client specified an invalid argument: {str(e)}"
		}, 400

		def get_session(sessionId: str) -> tuple:
		"""
		GET /sessions/{sessionId}
		Retrieve QoS session information
		"""
		session = sessions_db.get(sessionId)
		if not session:
		return {
		"status": 404,
		"code": "NOT_FOUND",
		"message": "Session not found"
		}, 404

		response = {
		"sessionId": session["sessionId"],
		"duration": session["duration"],
		"applicationServer": session["applicationServer"],
		"qosProfile": session["qosProfile"],
		"sink": session["sink"],
		"startedAt": session["startedAt"],
		"expiresAt": session["expiresAt"],
		"qosStatus": session["qosStatus"]
		}
		return response, 200

		def delete_session(sessionId: str) -> tuple:
		"""
		DELETE /sessions/{sessionId}
		Delete a QoS session
		"""
		session = sessions_db.get(sessionId)
		if not session:
		return {
		"status": 404,
		"code": "NOT_FOUND",
		"message": "Session not found"
		}, 404

		# Mark as UNAVAILABLE before deletion if it was AVAILABLE
		if session["qosStatus"] == QoSStatus.AVAILABLE:
		session["qosStatus"] = QoSStatus.UNAVAILABLE
		# Normally, send a notification callback here

		del sessions_db[sessionId]
		return "", 204

		def retrieve_sessions(body: Dict[str, Any]) -> tuple:
		"""
		POST /retrieve-sessions
		Retrieve QoS sessions associated with a specific device
		"""
		try:
		device = body.get("device", {})
		if not device:
		return {
		"status": 422,
		"code": "MISSING_FIELD",
		"message": "Device field is required"
		}, 422

		phone_number = device.get("phoneNumber")
		network_id = device.get("networkAccessIdentifier")
		ipv4_address = device.get("ipv4Address", {}).get("publicAddress")
		ipv6_address = device.get("ipv6Address")

		# Search for sessions that match any device identifier
		matched_sessions = []
		for session in sessions_db.values():
		session_device = session.get("device", {})
		if (
		session_device.get("phoneNumber") == phone_number
		or session_device.get("networkAccessIdentifier") == network_id
		or session_device.get("ipv4Address", {}).get("publicAddress") == ipv4_address
		or session_device.get("ipv6Address") == ipv6_address
		):
		matched_sessions.append({
		"applicationServer": session["applicationServer"],
		"qosProfile": session["qosProfile"],
		"sink": session["sink"],
		"sessionId": session["sessionId"],
		"duration": session["duration"],
		"startedAt": session["startedAt"],
		"expiresAt": session["expiresAt"],
		"qosStatus": session["qosStatus"]
		})

		if not matched_sessions:
		return {
		"status": 404,
		"code": "NOT_FOUND",
		"message": "No sessions found for the specified device"
		}, 404

		return matched_sessions, 200

		except Exception as e:
		return {
		"status": 400,
		"code": "INVALID_ARGUMENT",
		"message": f"Client specified an invalid argument: {str(e)}"
		}, 400
		No newline at end of file