tcod/trinity/service/data_juicer/server/server.py

import argparse
import json
import traceback
import uuid

from flask import Flask, jsonify, make_response, request

from trinity.service.data_juicer.client import (
    deserialize_arrow_to_dataset,
    serialize_dataset_to_arrow,
)
from trinity.service.data_juicer.server.session import DataJuicerSession
from trinity.service.data_juicer.server.utils import DJConfig

app = Flask(__name__)
sessions = {}


@app.route("/health", methods=["GET"])
def health_check():
    """Health check endpoint to verify server is running."""
    return jsonify({"status": "ok"}), 200


@app.route("/create", methods=["POST"])
def create():
    """Create a new data juicer session.

    Args:
        config (dict): Configuration parameters for the session.

    For example, the config should look like this:

    .. code-block:: python

        {
            "operators": [
                {
                    "operator1_name": {
                        "arg1": "value1",
                        "arg2": "value2"
                    }
                },
                {
                    "operator2_name": {
                        "arg1": "value1",
                        "arg2": "value2"
                    }
                }
            ],
            "config_path": "path/to/data_juicer_config.yaml",
        }
    """
    config = request.json
    try:
        config = DJConfig.model_validate(config)
    except Exception as e:
        return jsonify({"error": f"Failed to parse config: {e}"}), 400

    session_id = str(uuid.uuid4())
    sessions[session_id] = DataJuicerSession(config)

    return jsonify({"session_id": session_id, "message": "Session created successfully."})


@app.route("/process_experience", methods=["POST"])
def process_experience():
    """
    Process uploaded experiences for a given session.
    Expects a multipart/form-data POST with arrow bytes and session_id.
    Returns processed experiences and metrics as arrow bytes and JSON.
    """
    session_id = request.form.get("session_id")
    if not session_id or session_id not in sessions:
        return jsonify({"error": "Session ID not found."}), 404

    # Check for file in request
    if "arrow_data" not in request.files:
        return jsonify({"error": "No file uploaded."}), 400
    arrow_file = request.files["arrow_data"]
    arrow_bytes = arrow_file.read()
    ds = deserialize_arrow_to_dataset(arrow_bytes)

    # Process experiences using session
    session = sessions[session_id]
    try:
        # from_hf_datasets and to_hf_datasets should be imported from trinity.common.experience
        processed_ds, metrics = session.process_experience(ds)
    except Exception:
        return jsonify({"error": f"Experience processing failed:\n{traceback.format_exc()}"}), 500

    # Serialize processed experiences to parquet in-memory
    return_bytes = serialize_dataset_to_arrow(processed_ds)

    # Return arrow bytes and metrics as response
    response = make_response(return_bytes)
    response.headers["X-Metrics"] = json.dumps(metrics)
    response.headers["Content-Type"] = "application/octet-stream"
    response.headers["Content-Disposition"] = "attachment; filename=processed.arrow"
    return response


@app.route("/process_task", methods=["POST"])
def process_task():
    """Process a task for a given session.
    Different from process_experience which process a batch of experiences,
    this endpoint process a whole taskset.
    """
    data = request.json
    session_id = data.get("session_id")
    if not session_id or session_id not in sessions:
        return jsonify({"error": "Session ID not found."}), 404

    session = sessions[session_id]
    try:
        metrics = session.process_task()
        return jsonify({"metrics": metrics})
    except Exception:
        return jsonify({"error": f"Task processing failed:\n{traceback.format_exc()}"}), 500


@app.route("/close", methods=["POST"])
def close():
    """Close a data juicer session."""
    data = request.json
    session_id = data.get("session_id")
    if not session_id or session_id not in sessions:
        return jsonify({"error": "Session ID not found."}), 404

    del sessions[session_id]
    return jsonify({"message": "Session closed successfully."})


def main(host="localhost", port=5005, debug=False):
    app.run(host=host, port=port, debug=debug)


if __name__ == "__main__":
    parser = argparse.ArgumentParser(description="Data Processing Server")
    parser.add_argument("--host", type=str, default="localhost", help="Host address")
    parser.add_argument("--port", type=int, default=5005, help="Port number")
    parser.add_argument("--debug", action="store_true", help="Enable debug mode")
    args = parser.parse_args()
    main(host=args.host, port=args.port, debug=args.debug)