A script to download events from Cloud using the API.

get_events_from_Cloud.py

import glob
import logging
import os
import shutil
from pathlib import Path
import pandas as pd
import requests

API_URL = "https://api.cloud.pupil-labs.com/v2"

log = logging.getLogger(__name__)
logging.basicConfig(level=logging.INFO)

def api_get(path: str) -> dict:
    url = f"{API_URL}/{path}"
    if requests.get(url, headers={"api-key": API_KEY}).json()["status"] == "success":
        return requests.get(url, headers={"api-key": API_KEY}).json()["result"]
    else:
        error = requests.get(url, headers={"api-key": API_KEY}).json()["message"]
        log.error(error)
        raise (Exception(error))

def get_workspace_recordings(workspace_id: str) -> dict:
    log.info(f"fetching recordings in project:{workspace_id}")
    project_recordings = api_get(f"/workspaces/{workspace_id}/recordings")
    return project_recordings
    

def download_events(recording_id: str, workspace_id: str, download_path: str, **kwargs):
    os.makedirs(download_path, exist_ok=True)
    events= api_get(f"/workspaces/{workspace_id}/recordings/{recording_id}/events")
    filename= "events"
    for k, v in kwargs.items():
        filename = filename + f"_{v}"
    filename= filename +".csv"
    if len(events)==0:
        log.warning(f"No events found in recording {recording_id}")
    else:
        pd.DataFrame(events).to_csv(download_path / filename, index=False)
        log.info(filename)

if __name__ == "__main__":
    import argparse
    parser= argparse.ArgumentParser()
    parser.add_argument("--workspace_id", type=str, default=None, help="The ID of the workspace workspace_id")
    parser.add_argument("--download_path", type=str, default=None, help="Where to download the results download_path")
    parser.add_argument("--API_KEY", type=str, default=None, help="The API key to use"),
    args= parser.parse_args()

    API_KEY= args.API_KEY
    if API_KEY is None:
        raise Exception("--API_KEY is None. Please provide an API_KEY")
    workspace_id= args.workspace_id
    if workspace_id is None:
        raise Exception("--workspace_id is None. Please provide a workspace_id")
    download_path= Path(args.download_path) 
    if download_path is None:
        raise Exception("--download_path is None. Please provide a download_path")

    recs= get_workspace_recordings(workspace_id=workspace_id)
    log.info(f"Found {len(recs)} recordings in project {workspace_id}")
    for i, rec in enumerate(recs):
        log.info(f"Downloading events from recording {i} out of {len(recs)}: {rec['id']}")
        wearer_name= api_get(f"/workspaces/{workspace_id}/wearers/{rec['wearer_id']}")['name']
        download_events(recording_id=rec["id"], workspace_id=workspace_id, download_path=download_path, wearer_name=wearer_name, recording_name=rec["name"])