utils.py

import json
import os
import requests
import uuid

ACTIVE_MODE = True
LEARN_ENDPOINT = "https://api-sg-demo.adaptemy.io/learn"
CONTENT_OBJECTS_ENDPOINT = "https://api-sg-demo.adaptemy.io/content-objects"
DEFAULT_BEARER_TOKEN = "*****" 


def get_bearer_token():
    return (
        os.getenv("BEARER_TOKEN")
        or os.getenv("HF_BEARER_TOKEN")
        or DEFAULT_BEARER_TOKEN
    )


def get_headers():
    return {
        "Authorization": f"Bearer {get_bearer_token()}",
        "Content-Type": "application/json",
        "Accept": "application/json",
    }

def new_guid():
    return str(uuid.uuid4())

def xAPI_deploy_request(endpoint, payload, headers=None):
    request_headers = headers or get_headers()

    response = requests.post(
        endpoint,
        headers=request_headers,
        json=payload
    )

    print("Status code (content object post):", response.status_code)
    try:
        print(json.dumps(response.json(), indent=2))
    except Exception:
        print(response.text)

def xAPI_user_request(endpoint, payload, headers=None):
    request_headers = headers or get_headers()

    response = requests.post(
        endpoint,
        headers=request_headers,
        json=payload
    )

    print("Status code (experience post):", response.status_code)
    if response.status_code == 200:
        try:
            print("Sucessfull submmited the learner experience:", json.dumps(payload, indent=2))
            return( f"✅ Submitted **{len(payload)}** learner experience record(s).")
        except Exception as e:
            print("error due to:", e)
            return(f"error due to: {e}")
    else:
        print("Failed to submit learner experience.")
        print("Response:", response.text)
        return(f"Failed to submit learner experience : {response.text}")

def xAPI_validate(params, endpoint = CONTENT_OBJECTS_ENDPOINT, headers=None):
    request_headers = headers or get_headers()

    response = requests.get(
        endpoint,
        headers=request_headers,
        params=params
    )

    print("Status code (content object post):", response.status_code)
    try:
        print(json.dumps(response.json(), indent=2))
    except Exception:
        print(response.text)

def xr_experience_event_formatting(
    learner_name: str,
    duration: str,
    xr_content_object_id: str,
    session_id: str,
):
    """
    Formulates a learner's XR experience event to Adaptemy xAPI as per the defined schema.

    Args:
        learner_name: The name of the learner.
        duration: The temporal duration of the experience.
        xr_content_object_id: The ID of the XR multisensory content object.
        session_id: The ID of the learning session.
    """

    xr_experience_events = {
        "actor": {
            "account": {
                "homepage": "https://api-sg-demo.adaptemy.io/organization/23",
                "name": learner_name
            }
        },
        "object": {
            "id": xr_content_object_id,
        },
        "verb": {
            "id": "http://activitystrea.ms/schema/1.0/experienced"
        },
        "result": {
            "completion": True,
            "duration": duration
        },
        "context": {
            "platform": "heat-xrel",
            "registration": session_id,
            "active": ACTIVE_MODE
        }
    }

    return xr_experience_events



def post_xr_content_object(
    description_text: str,
    name_text: str,
    media_type: str,
    scene: str,
    media_device: str,
    concept_guids: list[str] = None
):
    """
    Formulates and posts an XR multisensory content object to Adaptemy xAPI.

    Args:
        description_text: Description of the XR experience.
        name_text: Short name of the XR experience.
        media_type: Type of media (e.g., "sensory-xr", "hologram-xr").
        scene: Scene description for media details.
        media_device: Device description for media details.
        concept_guids: Optional list of concept GUIDs related to the content.

    Returns:
        The ID of the posted XR multisensory content object.
    """
    xr_content_guid = new_guid()
    xr_content_object_id = f"https://api.adaptemy.io/content/{xr_content_guid}"

    extensions_data = {
        "https://api.adaptemy.io/contentmetadata/active": ACTIVE_MODE,
        "https://api.adaptemy.io/contentmetadata/learning-phase": "inquiry-based-learning",
        "https://api.adaptemy.io/contentmetadata/scene": scene,
        "https://api.adaptemy.io/contentmetadata/media-details": {
            "type": media_type,
            "device": media_device
        }
    }
    if concept_guids:
        extensions_data["https://api.adaptemy.io/concept"] = concept_guids

    xr_content_objects = [{
        "id": xr_content_object_id,
        "organizationId": 23,
        "definition": {
            "description": {
                "en-US": description_text
            },
            "extensions": extensions_data,
            "name": {
                "en-US": name_text
            },
            "type": "http://adlnet.gov/expapi/activities/media"
        },
        "isEnabled": True
    }]

    xAPI_deploy_request(CONTENT_OBJECTS_ENDPOINT, xr_content_objects)

    return xr_content_object_id