Skip to content

messaging

framewise_meet_client.messaging

MessageSender

Manages sending messages to the server.

Source code in framewise_meet_client/messaging.py 
 28
 29
 30
 31
 32
 33
 34
 35
 36
 37
 38
 39
 40
 41
 42
 43
 44
 45
 46
 47
 48
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
class MessageSender:
    """Manages sending messages to the server."""

    def __init__(self, connection):
        """Initialize the message sender.

        Args:
            connection: WebSocketConnection instance
        """
        self.connection = connections

    def _prepare_message(self, message_class: Type[T], **kwargs) -> T:
        """Prepare a message with standard fields.

        Args:
            message_class: The message class to instantiate
            **kwargs: Additional fields for the message

        Returns:
            An instance of the message class
        """
        return message_class(**kwargs)

    async def _send_message(self, message: Any) -> None:
        """Send a message over the WebSocket connection.

        Args:
            message: Message object to send (will be converted to dict)

        Raises:
            ConnectionError: If the message cannot be sent
        """
        try:
            # Convert Pydantic model to dict if it's a model
            if hasattr(message, "model_dump"):
                message_dict = message.model_dump()
            else:
                message_dict = message

            await self.connection.send_json(message_dict)
            logger.debug(f"Sent message: {json.dumps(message_dict)[:100]}...")
        except Exception as e:
            logger.error(f"Error sending message: {str(e)}")
            raise ConnectionError(f"Failed to send message: {str(e)}")

    def send_generated_text(
        self,
        text: str,
        is_generation_end: bool = False,
        loop: asyncio.AbstractEventLoop = None,
    ) -> None:
        """Send generated text to the server.

        Args:
            text: The generated text
            is_generation_end: Whether this is the end of generation
            loop: Event loop to use for coroutine execution (uses current loop if None)
        """
        content = GeneratedTextContent(text=text, is_generation_end=is_generation_end)
        message = self._prepare_message(GeneratedTextMessage, content=content)

        if loop:
            asyncio.run_coroutine_threadsafe(self._send_message(message), loop)
        else:
            asyncio.create_task(self._send_message(message))

    def send_custom_ui_element(
        self, ui_type: str, data: Dict[str, Any], loop: asyncio.AbstractEventLoop = None
    ) -> None:
        """Send a custom UI element to the server.

        This is a generic method for sending custom UI elements. For specific UI elements,
        consider creating dedicated helper methods for better type safety and easier usage.

        Args:
            ui_type: Type of UI element - must be supported by the Framewise Meet client
            data: Data specific to the UI element
            loop: Event loop to use for coroutine execution (uses current loop if None)

        Examples:
            ```python
            # Send a custom chart component
            sender.send_custom_ui_element(
                ui_type="chart",
                data={
                    "chart_type": "bar",
                    "title": "Meeting Participation",
                    "data": [10, 20, 35, 15],
                    "labels": ["Week 1", "Week 2", "Week 3", "Week 4"]
                }
            )
            ```

        Notes:
            For creating new custom UI element types, refer to the documentation on
            adding custom UI components.
        """
        # Create a generic custom UI element
        custom_element = {"type": ui_type, "data": data}
        message = self._prepare_message(CustomUIElementMessage, content=custom_element)

        if loop:
            asyncio.run_coroutine_threadsafe(self._send_message(message), loop)
        else:
            asyncio.create_task(self._send_message(message))

    def send_mcq_question(
        self,
        question_id: str,
        question: str,
        options: List[str],
        loop: asyncio.AbstractEventLoop = None,
        image_path: Optional[str] = None,
    ) -> None:
        """Send an MCQ question as a custom UI element.

        Args:
            question_id: Unique identifier for the question
            question: The question text
            options: List of option texts
            loop: Event loop to run the coroutine in
            image_path: Optional path to an image to display with the question
        """
        # Create a proper MCQ question element
        mcq_data = MCQQuestionData(
            id=question_id, question=question, options=options, image_path=image_path
        )
        mcq_element = MCQQuestionElement(type="mcq_question", data=mcq_data)
        message = self._prepare_message(CustomUIElementMessage, content=mcq_element)

        if loop:
            asyncio.run_coroutine_threadsafe(self._send_message(message), loop)
        else:
            asyncio.create_task(self._send_message(message))

    def send_notification(
        self,
        message: str,
        level: str = "info",
        duration: int = 8000,
        loop: Optional[asyncio.AbstractEventLoop] = None,
    ) -> None:
        """Send a notification to be displayed in the UI.

        Args:
            message: The notification message
            level: Notification level (info, warning, error, success)
            duration: Duration to show the notification in milliseconds
            loop: Event loop to run the coroutine in
        """
        notification_data = NotificationData(
            message=message, level=level, duration=duration
        )
        notification_element = NotificationElement(
            type="notification_element", data=notification_data
        )
        msg = self._prepare_message(
            CustomUIElementMessage, content=notification_element
        )

        if loop:
            asyncio.run_coroutine_threadsafe(self._send_message(msg), loop)
        else:
            asyncio.create_task(self._send_message(msg))

    def send_error(
        self,
        error_message: str,
        error_code: Optional[str] = None,
        loop: Optional[asyncio.AbstractEventLoop] = None,
    ) -> None:
        """Send an error message to the server.

        Args:
            error_message: The error message
            error_code: Optional error code
            loop: Event loop to run the coroutine in
        """
        message = self._prepare_message(
            ErrorResponse, error=error_message, error_code=error_code
        )

        if loop:
            asyncio.run_coroutine_threadsafe(self._send_message(message), loop)
        else:
            asyncio.create_task(self._send_message(message))

__init__(connection)

Initialize the message sender.

Parameters:

Name Type Description Default
connection

WebSocketConnection instance

 required
Source code in framewise_meet_client/messaging.py 
31
32
33
34
35
36
37
def __init__(self, connection):
    """Initialize the message sender.

    Args:
        connection: WebSocketConnection instance
    """
    self.connection = connections

send_generated_text(text, is_generation_end=False, loop=None)

Send generated text to the server.

Parameters:

Name Type Description Default
text str

The generated text

 required
is_generation_end bool

Whether this is the end of generation

 False
loop AbstractEventLoop

Event loop to use for coroutine execution (uses current loop if None)

 None
Source code in framewise_meet_client/messaging.py 
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
def send_generated_text(
    self,
    text: str,
    is_generation_end: bool = False,
    loop: asyncio.AbstractEventLoop = None,
) -> None:
    """Send generated text to the server.

    Args:
        text: The generated text
        is_generation_end: Whether this is the end of generation
        loop: Event loop to use for coroutine execution (uses current loop if None)
    """
    content = GeneratedTextContent(text=text, is_generation_end=is_generation_end)
    message = self._prepare_message(GeneratedTextMessage, content=content)

    if loop:
        asyncio.run_coroutine_threadsafe(self._send_message(message), loop)
    else:
        asyncio.create_task(self._send_message(message))

send_custom_ui_element(ui_type, data, loop=None)

Send a custom UI element to the server.

This is a generic method for sending custom UI elements. For specific UI elements, consider creating dedicated helper methods for better type safety and easier usage.

Parameters:

Name Type Description Default
ui_type str

Type of UI element - must be supported by the Framewise Meet client

 required
data Dict[str, Any]

Data specific to the UI element

 required
loop AbstractEventLoop

Event loop to use for coroutine execution (uses current loop if None)

 None

Examples:

# Send a custom chart component
sender.send_custom_ui_element(
    ui_type="chart",
    data={
        "chart_type": "bar",
        "title": "Meeting Participation",
        "data": [10, 20, 35, 15],
        "labels": ["Week 1", "Week 2", "Week 3", "Week 4"]
    }
)
Notes

For creating new custom UI element types, refer to the documentation on adding custom UI components.

Source code in framewise_meet_client/messaging.py 
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
def send_custom_ui_element(
    self, ui_type: str, data: Dict[str, Any], loop: asyncio.AbstractEventLoop = None
) -> None:
    """Send a custom UI element to the server.

    This is a generic method for sending custom UI elements. For specific UI elements,
    consider creating dedicated helper methods for better type safety and easier usage.

    Args:
        ui_type: Type of UI element - must be supported by the Framewise Meet client
        data: Data specific to the UI element
        loop: Event loop to use for coroutine execution (uses current loop if None)

    Examples:
        ```python
        # Send a custom chart component
        sender.send_custom_ui_element(
            ui_type="chart",
            data={
                "chart_type": "bar",
                "title": "Meeting Participation",
                "data": [10, 20, 35, 15],
                "labels": ["Week 1", "Week 2", "Week 3", "Week 4"]
            }
        )
        ```

    Notes:
        For creating new custom UI element types, refer to the documentation on
        adding custom UI components.
    """
    # Create a generic custom UI element
    custom_element = {"type": ui_type, "data": data}
    message = self._prepare_message(CustomUIElementMessage, content=custom_element)

    if loop:
        asyncio.run_coroutine_threadsafe(self._send_message(message), loop)
    else:
        asyncio.create_task(self._send_message(message))

send_mcq_question(question_id, question, options, loop=None, image_path=None)

Send an MCQ question as a custom UI element.

Parameters:

Name Type Description Default
question_id str

Unique identifier for the question

 required
question str

The question text

 required
options List[str]

List of option texts

 required
loop AbstractEventLoop

Event loop to run the coroutine in

 None
image_path Optional[str]

Optional path to an image to display with the question

 None
Source code in framewise_meet_client/messaging.py 
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
def send_mcq_question(
    self,
    question_id: str,
    question: str,
    options: List[str],
    loop: asyncio.AbstractEventLoop = None,
    image_path: Optional[str] = None,
) -> None:
    """Send an MCQ question as a custom UI element.

    Args:
        question_id: Unique identifier for the question
        question: The question text
        options: List of option texts
        loop: Event loop to run the coroutine in
        image_path: Optional path to an image to display with the question
    """
    # Create a proper MCQ question element
    mcq_data = MCQQuestionData(
        id=question_id, question=question, options=options, image_path=image_path
    )
    mcq_element = MCQQuestionElement(type="mcq_question", data=mcq_data)
    message = self._prepare_message(CustomUIElementMessage, content=mcq_element)

    if loop:
        asyncio.run_coroutine_threadsafe(self._send_message(message), loop)
    else:
        asyncio.create_task(self._send_message(message))

send_notification(message, level='info', duration=8000, loop=None)

Send a notification to be displayed in the UI.

Parameters:

Name Type Description Default
message str

The notification message

 required
level str

Notification level (info, warning, error, success)

 'info'
duration int

Duration to show the notification in milliseconds

 8000
loop Optional[AbstractEventLoop]

Event loop to run the coroutine in

 None
Source code in framewise_meet_client/messaging.py 
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
def send_notification(
    self,
    message: str,
    level: str = "info",
    duration: int = 8000,
    loop: Optional[asyncio.AbstractEventLoop] = None,
) -> None:
    """Send a notification to be displayed in the UI.

    Args:
        message: The notification message
        level: Notification level (info, warning, error, success)
        duration: Duration to show the notification in milliseconds
        loop: Event loop to run the coroutine in
    """
    notification_data = NotificationData(
        message=message, level=level, duration=duration
    )
    notification_element = NotificationElement(
        type="notification_element", data=notification_data
    )
    msg = self._prepare_message(
        CustomUIElementMessage, content=notification_element
    )

    if loop:
        asyncio.run_coroutine_threadsafe(self._send_message(msg), loop)
    else:
        asyncio.create_task(self._send_message(msg))

send_error(error_message, error_code=None, loop=None)

Send an error message to the server.

Parameters:

Name Type Description Default
error_message str

The error message

 required
error_code Optional[str]

Optional error code

 None
loop Optional[AbstractEventLoop]

Event loop to run the coroutine in

 None
Source code in framewise_meet_client/messaging.py 
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
def send_error(
    self,
    error_message: str,
    error_code: Optional[str] = None,
    loop: Optional[asyncio.AbstractEventLoop] = None,
) -> None:
    """Send an error message to the server.

    Args:
        error_message: The error message
        error_code: Optional error code
        loop: Event loop to run the coroutine in
    """
    message = self._prepare_message(
        ErrorResponse, error=error_message, error_code=error_code
    )

    if loop:
        asyncio.run_coroutine_threadsafe(self._send_message(message), loop)
    else:
        asyncio.create_task(self._send_message(message))