src/models.py

"""
OFP Data Models
Implements Open Floor Protocol envelope and event structures following v1.0.0 specifications
"""

from dataclasses import dataclass, field
from typing import List, Dict, Optional, Any
from datetime import datetime, timezone
import json
import uuid


@dataclass
class Identification:
    """Assistant identification information"""
    speaker_uri: str
    service_url: str
    conversational_name: str
    organization: Optional[str] = None
    role: Optional[str] = None
    synopsis: Optional[str] = None


@dataclass
class DialogEvent:
    """Dialog event following OFP Dialog Event Object v1.0.2"""
    id: str
    speaker_uri: str
    span: Dict[str, str]
    features: Dict[str, Any]

    @staticmethod
    def create_text_event(speaker_uri: str, text: str, event_id: Optional[str] = None) -> 'DialogEvent':
        """Create a text-based dialog event"""
        return DialogEvent(
            id=event_id or f"de:{uuid.uuid4()}",
            speaker_uri=speaker_uri,
            span={"startTime": datetime.now(timezone.utc).isoformat().replace('+00:00', 'Z')},
            features={
                "text": {
                    "mimeType": "text/plain",
                    "tokens": [{"value": text}]
                }
            }
        )

    def to_dict(self) -> Dict:
        """Convert to dictionary for serialization"""
        return {
            "id": self.id,
            "speakerUri": self.speaker_uri,
            "span": self.span,
            "features": self.features
        }


@dataclass
class Event:
    """OFP Event structure for inter-agent messages"""
    event_type: str
    to: Optional[Dict[str, Any]] = None
    parameters: Optional[Dict[str, Any]] = None

    def to_dict(self) -> Dict:
        """Convert to dictionary for serialization"""
        result = {"eventType": self.event_type}
        if self.to:
            result["to"] = self.to
        if self.parameters:
            result["parameters"] = self.parameters
        return result


@dataclass
class Envelope:
    """OFP Envelope following Inter-agent Message v1.0.0"""
    schema: Dict[str, str]
    conversation: Dict[str, Any]
    sender: Dict[str, str]
    events: List[Dict[str, Any]]

    @staticmethod
    def from_json(json_str: str) -> 'Envelope':
        """Parse OFP envelope from JSON string"""
        data = json.loads(json_str)
        ofp = data.get('openFloor', {})
        return Envelope(
            schema=ofp.get('schema', {}),
            conversation=ofp.get('conversation', {}),
            sender=ofp.get('sender', {}),
            events=ofp.get('events', [])
        )

    @staticmethod
    def from_dict(data: Dict) -> 'Envelope':
        """Parse OFP envelope from dictionary"""
        ofp = data.get('openFloor', data)  # Support both wrapped and unwrapped
        return Envelope(
            schema=ofp.get('schema', {}),
            conversation=ofp.get('conversation', {}),
            sender=ofp.get('sender', {}),
            events=ofp.get('events', [])
        )

    def to_payload(self) -> Dict:
        """Convert to JSON payload for transmission"""
        return {
            "openFloor": {
                "schema": self.schema,
                "conversation": self.conversation,
                "sender": self.sender,
                "events": self.events
            }
        }

    def to_json(self) -> str:
        """Convert to JSON string"""
        return json.dumps(self.to_payload(), indent=2)


def validate_envelope(envelope: Envelope) -> bool:
    """Validate OFP envelope structure"""
    try:
        # Check required fields
        if not envelope.schema or 'version' not in envelope.schema:
            return False

        if not envelope.conversation or 'id' not in envelope.conversation:
            return False

        if not envelope.sender or 'speakerUri' not in envelope.sender:
            return False

        if not isinstance(envelope.events, list):
            return False

        # Validate each event
        for event in envelope.events:
            if not isinstance(event, dict) or 'eventType' not in event:
                return False

        return True
    except Exception:
        return False


def create_envelope(conversation_id: str, speaker_uri: str, events: List[Dict]) -> Envelope:
    """Helper function to create a valid OFP envelope"""
    return Envelope(
        schema={"version": "1.0.0"},
        conversation={"id": conversation_id},
        sender={"speakerUri": speaker_uri},
        events=events
    )