Add streaming example for expression measurement

expression-measurement/streaming/python-streaming-example/.env.example

@@ -0,0 +1 @@
+HUME_API_KEY=your_api_key_here

expression-measurement/streaming/python-streaming-example/.gitignore

@@ -0,0 +1,3 @@
+.env*.local
+.env
+*.egg-info/

expression-measurement/streaming/python-streaming-example/README.md

@@ -0,0 +1,44 @@
+<div align="center">
+  <img src="https://storage.googleapis.com/hume-public-logos/hume/hume-banner.png">
+  <h1>Expression Measurement | Sample Python Implementation</h1>
+  <p>
+    <strong>Analyze language using Hume's Python SDK.</strong>
+  </p>
+</div>
+
+## Overview
+This is a sample implementation of Hume's Expression Measurement using the Websocket interface in Hume's Python SDK. This example using the "language" model only, which is for analyzing the emotional content of of text.
+
+See the [python-top-emotions batch example](../../batch/python-top-emotions) for how to process pre-recorded media files (images) using the *Batch* API, and see the [next-js-streaming-example](../next-js-streaming-example) for an example of how to use the Streaming API on the web to process live data coming from a webcam or microphone.
+
+## Setup
+
+1. Copy the `.env.example` file to `.env` and add your Hume API key:
+   ```
+   cp .env.example .env
+   ```
+
+2. Edit `.env` and replace `your_api_key_here` with your actual Hume API key
+
+3. Create a virtual environment and install dependencies using `uv`:
+   ```
+   uv venv
+   uv pip install -e .
+   ```
+
+## Usage
+
+Run the example with `uv`:
+```
+uv run main.py
+```
+
+Or using standard Python:
+```
+python main.py
+```
+
+Type text and press Enter to analyze the emotions in the text. The example will display the emotion scores sorted from highest to lowest.
+
+Type `exit` or press Ctrl+C to exit the application.
+

expression-measurement/streaming/python-streaming-example/main.py

@@ -0,0 +1,103 @@
+#!/usr/bin/env python3
+"""
+Simple Hume Expression Measurement Streaming API example using the language model.
+This example takes text input from stdin and outputs emotion summaries to stdout.
+"""
+
+import asyncio
+import os
+import sys
+from typing import List, Tuple, TypedDict
+from dotenv import load_dotenv
+
+from hume import AsyncHumeClient
+from hume.expression_measurement.stream.stream.types.config import Config
+from hume.expression_measurement.stream.stream.types.stream_language import StreamLanguage
+from hume.expression_measurement.stream.stream.types.stream_model_predictions import StreamModelPredictions
+from hume.expression_measurement.stream.stream.types.subscribe_event import SubscribeEvent
+
+load_dotenv()
+
+API_KEY = os.getenv("HUME_API_KEY")
+if not API_KEY:
+    print("Error: HUME_API_KEY environment variable not set")
+    print("Please create a .env file with your API key")
+    sys.exit(1)
+
+class Result(TypedDict):
+    text: str | None
+    scores: List[Tuple[str, float]]
+
+def process_emotion_scores(event: StreamModelPredictions) -> List[Result]:
+    if not hasattr(event, 'language') or event.language is None or event.language.predictions is None or len(event.language.predictions) == 0:
+        raise ValueError("Unexpected: event does not contain language data")
+
+    ret: List[Result] = []
+    for prediction in event.language.predictions:
+        ret.append({
+            "scores": sorted(
+              [(item.name, item.score) for item in prediction.emotions or [] if item.name and item.score is not None],
+              key=lambda x: x[1],
+              reverse=True),
+            "text": prediction.text,
+        })
+    return ret
+
+def print_emotion_summary(result: Result) -> None:
+    if not result:
+        print("No emotions detected")
+        return
+
+    print("\nEmotion Analysis:")
+    print("-" * 40)
+    for emotion, score in result['scores'][:5]:  # Show top 5 emotions
+        print(f"{emotion.ljust(15)}: {score:.4f}")
+    print("-" * 40)
+
+async def streaming_example() -> None:
+    """Main function to run the streaming example"""
+    client = AsyncHumeClient(api_key=API_KEY)
+
+    language_config = StreamLanguage(granularity="sentence")
+    config = Config(language=language_config)
+
+    print("Hume Language Emotion Streaming Example")
+    print("-" * 60)
+
+    async with client.expression_measurement.stream.connect(
+        options={"config": config},
+    ) as socket:
+        print("\nEnter text and press Enter to analyze emotions")
+
+        while True:
+            try:
+                # Get user input
+                text = input("\nType 'exit' to quit\n> ")
+                if text.lower() == 'exit':
+                    break
+
+                if not text.strip():
+                    continue
+
+                response = await socket.send_text(text=text)
+
+                if (not response or not isinstance(response, StreamModelPredictions)):
+                    print("Error: Invalid response from server")
+                    continue
+                results = process_emotion_scores(response)
+                for result in results:
+                    print_emotion_summary(result)
+
+            except KeyboardInterrupt:
+                print("\nExiting...")
+                break
+
+def main():
+    """Entry point for the script"""
+    try:
+        asyncio.run(streaming_example())
+    except KeyboardInterrupt:
+        print("\nExiting...")
+
+if __name__ == "__main__":
+    main()

expression-measurement/streaming/python-streaming-example/pyproject.toml

@@ -0,0 +1,13 @@
+[project]
+name = "hume-stream-simple"
+version = "0.1.0"
+description = "Simple Hume streaming example using language model"
+requires-python = ">=3.9"
+dependencies = [
+    "hume",
+    "python-dotenv",
+]
+
+[build-system]
+requires = ["setuptools", "wheel"]
+build-backend = "setuptools.build_meta"