Skip to content

Handlers

Sum Handler

datadivr.handlers

Message handlers for DataDivr.

Classes

HandlerType

Bases: Enum

Type of handler to register.

Source code in datadivr/handlers/registry.py 
11
12
13
14
15
16
class HandlerType(Enum):
    """Type of handler to register."""

    SERVER = auto()
    CLIENT = auto()
    BOTH = auto()

Functions

get_handlers(handler_type=HandlerType.SERVER)

Get registered handlers for the specified type.

Parameters:

Name Type Description Default
handler_type HandlerType

Type of handlers to retrieve (SERVER or CLIENT)

 SERVER
Source code in datadivr/handlers/registry.py 
24
25
26
27
28
29
30
31
32
33
34
35
def get_handlers(
    handler_type: HandlerType = HandlerType.SERVER,
) -> dict[str, Callable[[WebSocketMessage], Awaitable[WebSocketMessage | None]]]:
    """
    Get registered handlers for the specified type.

    Args:
        handler_type: Type of handlers to retrieve (SERVER or CLIENT)
    """
    if handler_type == HandlerType.SERVER:
        return _server_handlers
    return _client_handlers

get_node_info_handler(message) async

Handle requests to get information about a specific node.

Source code in datadivr/handlers/custom_handlers.py 
26
27
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
@websocket_handler("get_node_info", HandlerType.SERVER)
async def get_node_info_handler(message: WebSocketMessage) -> WebSocketMessage:
    """Handle requests to get information about a specific node."""
    node_index = message.payload.get("index") if message.payload else None
    current_project = ProjectManager.get_current_project()

    if current_project is None:
        return WebSocketMessage(
            event_name="get_node_info_result", payload={"error": "No project is currently open"}, to=message.from_id
        )

    if node_index is None:
        return WebSocketMessage(
            event_name="get_node_info_result", payload={"error": "Node index not provided"}, to=message.from_id
        )

    try:
        node_data = (
            current_project.nodes_data.get_attributes_by_index(node_index) if current_project.nodes_data else None
        )

        if node_data is None:
            return WebSocketMessage(
                event_name="get_node_info_result", payload={"error": "Node data not found"}, to=message.from_id
            )

        # Convert all float32 values to float
        node_data = {k: float(v) if isinstance(v, np.float32) else v for k, v in node_data.items()}

        return WebSocketMessage(event_name="get_node_info_result", payload=node_data, to=message.from_id)
    except IndexError as e:
        return WebSocketMessage(event_name="get_node_info_result", payload={"error": str(e)}, to=message.from_id)
    except Exception as e:
        return WebSocketMessage(event_name="get_node_info_result", payload={"error": str(e)}, to=message.from_id)

handle_sum_result(message) async

Handle the result of a sum calculation on the client side.

Parameters:

Name Type Description Default
message WebSocketMessage

A WebSocketMessage containing the sum result in its payload

 required
Source code in datadivr/handlers/builtin/sum_handler.py 
43
44
45
46
47
48
49
50
51
@websocket_handler("sum_handler_result", HandlerType.CLIENT)
async def handle_sum_result(message: WebSocketMessage) -> None:
    """Handle the result of a sum calculation on the client side.

    Args:
        message: A WebSocketMessage containing the sum result in its payload
    """
    print(f"*** handle_sum_result(): {message.from_id}: '{message.payload}'")
    return None

msg_handler(message) async

Handle generic text messages on the client side.

Parameters:

Name Type Description Default
message WebSocketMessage

A WebSocketMessage containing a text message

 required
Source code in datadivr/handlers/builtin/sum_handler.py 
54
55
56
57
58
59
60
61
62
@websocket_handler("msg", HandlerType.CLIENT)
async def msg_handler(message: WebSocketMessage) -> None:
    """Handle generic text messages on the client side.

    Args:
        message: A WebSocketMessage containing a text message
    """
    print(f">> {message.from_id}({message.event_name}): '{message.message}'")
    return None

sum_handler(message) async

Handle requests to calculate the sum of a list of numbers.

This handler processes both server and client-side sum calculation requests. It expects a payload containing a list of numbers and returns their sum.

Parameters:

Name Type Description Default
message WebSocketMessage

A WebSocketMessage containing a payload with a "numbers" key holding a list of numeric values

 required

Returns:

Name Type Description
WebSocketMessage WebSocketMessage

A message containing the sum result or an error message
Example payload

{"numbers": [1, 2, 3, 4, 5]}

Source code in datadivr/handlers/builtin/sum_handler.py 
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
@websocket_handler("sum_event", HandlerType.SERVER)
@websocket_handler("sum_event_client", HandlerType.CLIENT)
async def sum_handler(message: WebSocketMessage) -> WebSocketMessage:
    """Handle requests to calculate the sum of a list of numbers.

    This handler processes both server and client-side sum calculation requests.
    It expects a payload containing a list of numbers and returns their sum.

    Args:
        message: A WebSocketMessage containing a payload with a "numbers" key
                holding a list of numeric values

    Returns:
        WebSocketMessage: A message containing the sum result or an error message

    Example payload:
        {"numbers": [1, 2, 3, 4, 5]}
    """
    try:
        payload = message.payload
        if not isinstance(payload, dict):
            return create_error_message("Invalid payload format", message.from_id)

        numbers = payload.get("numbers")
        if not isinstance(numbers, list):
            return create_error_message("Payload must contain a list of numbers", message.from_id)

        result = sum(float(n) for n in numbers)
        return WebSocketMessage(
            event_name="sum_handler_result",
            payload=result,
            to=message.from_id,
        )
    except Exception as e:
        return create_error_message(f"Error: {e!s}", message.from_id)

websocket_handler(event_name, handler_type=HandlerType.SERVER)

Decorator to register a websocket handler function.

Parameters:

Name Type Description Default
event_name str

The event name to register the handler for.

 required
handler_type HandlerType

Where this handler should be registered (SERVER, CLIENT, or BOTH)

 SERVER
Example

@websocket_handler("sum_event", HandlerType.BOTH) async def sum_handler(message: WebSocketMessage) -> Optional[WebSocketMessage]: ...

Source code in datadivr/handlers/registry.py 
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
def websocket_handler(
    event_name: str, handler_type: HandlerType = HandlerType.SERVER
) -> Callable[[Callable[..., Awaitable[WebSocketMessage | None]]], Callable[..., Awaitable[WebSocketMessage | None]]]:
    """
    Decorator to register a websocket handler function.

    Args:
        event_name: The event name to register the handler for.
        handler_type: Where this handler should be registered (SERVER, CLIENT, or BOTH)

    Example:
        @websocket_handler("sum_event", HandlerType.BOTH)
        async def sum_handler(message: WebSocketMessage) -> Optional[WebSocketMessage]:
            ...
    """

    def decorator(
        func: Callable[..., Awaitable[WebSocketMessage | None]],
    ) -> Callable[..., Awaitable[WebSocketMessage | None]]:
        @wraps(func)
        async def wrapper(*args: Any, **kwargs: Any) -> WebSocketMessage | None:
            return await func(*args, **kwargs)

        if handler_type in (HandlerType.SERVER, HandlerType.BOTH):
            _server_handlers[event_name] = wrapper
        if handler_type in (HandlerType.CLIENT, HandlerType.BOTH):
            _client_handlers[event_name] = wrapper

        return wrapper

    return decorator

Modules

builtin

Built-in message handlers.

Modules
sum_handler
Classes Functions
handle_sum_result(message) async

Handle the result of a sum calculation on the client side.

Parameters:

Name Type Description Default
message WebSocketMessage

A WebSocketMessage containing the sum result in its payload

 required
Source code in datadivr/handlers/builtin/sum_handler.py 
43
44
45
46
47
48
49
50
51
@websocket_handler("sum_handler_result", HandlerType.CLIENT)
async def handle_sum_result(message: WebSocketMessage) -> None:
    """Handle the result of a sum calculation on the client side.

    Args:
        message: A WebSocketMessage containing the sum result in its payload
    """
    print(f"*** handle_sum_result(): {message.from_id}: '{message.payload}'")
    return None
msg_handler(message) async

Handle generic text messages on the client side.

Parameters:

Name Type Description Default
message WebSocketMessage

A WebSocketMessage containing a text message

 required
Source code in datadivr/handlers/builtin/sum_handler.py 
54
55
56
57
58
59
60
61
62
@websocket_handler("msg", HandlerType.CLIENT)
async def msg_handler(message: WebSocketMessage) -> None:
    """Handle generic text messages on the client side.

    Args:
        message: A WebSocketMessage containing a text message
    """
    print(f">> {message.from_id}({message.event_name}): '{message.message}'")
    return None
sum_handler(message) async

Handle requests to calculate the sum of a list of numbers.

This handler processes both server and client-side sum calculation requests. It expects a payload containing a list of numbers and returns their sum.

Parameters:

Name Type Description Default
message WebSocketMessage

A WebSocketMessage containing a payload with a "numbers" key holding a list of numeric values

 required

Returns:

Name Type Description
WebSocketMessage WebSocketMessage

A message containing the sum result or an error message
Example payload

{"numbers": [1, 2, 3, 4, 5]}

Source code in datadivr/handlers/builtin/sum_handler.py 
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
@websocket_handler("sum_event", HandlerType.SERVER)
@websocket_handler("sum_event_client", HandlerType.CLIENT)
async def sum_handler(message: WebSocketMessage) -> WebSocketMessage:
    """Handle requests to calculate the sum of a list of numbers.

    This handler processes both server and client-side sum calculation requests.
    It expects a payload containing a list of numbers and returns their sum.

    Args:
        message: A WebSocketMessage containing a payload with a "numbers" key
                holding a list of numeric values

    Returns:
        WebSocketMessage: A message containing the sum result or an error message

    Example payload:
        {"numbers": [1, 2, 3, 4, 5]}
    """
    try:
        payload = message.payload
        if not isinstance(payload, dict):
            return create_error_message("Invalid payload format", message.from_id)

        numbers = payload.get("numbers")
        if not isinstance(numbers, list):
            return create_error_message("Payload must contain a list of numbers", message.from_id)

        result = sum(float(n) for n in numbers)
        return WebSocketMessage(
            event_name="sum_handler_result",
            payload=result,
            to=message.from_id,
        )
    except Exception as e:
        return create_error_message(f"Error: {e!s}", message.from_id)

custom_handlers

Classes
Functions
get_node_info_handler(message) async

Handle requests to get information about a specific node.

Source code in datadivr/handlers/custom_handlers.py 
26
27
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
@websocket_handler("get_node_info", HandlerType.SERVER)
async def get_node_info_handler(message: WebSocketMessage) -> WebSocketMessage:
    """Handle requests to get information about a specific node."""
    node_index = message.payload.get("index") if message.payload else None
    current_project = ProjectManager.get_current_project()

    if current_project is None:
        return WebSocketMessage(
            event_name="get_node_info_result", payload={"error": "No project is currently open"}, to=message.from_id
        )

    if node_index is None:
        return WebSocketMessage(
            event_name="get_node_info_result", payload={"error": "Node index not provided"}, to=message.from_id
        )

    try:
        node_data = (
            current_project.nodes_data.get_attributes_by_index(node_index) if current_project.nodes_data else None
        )

        if node_data is None:
            return WebSocketMessage(
                event_name="get_node_info_result", payload={"error": "Node data not found"}, to=message.from_id
            )

        # Convert all float32 values to float
        node_data = {k: float(v) if isinstance(v, np.float32) else v for k, v in node_data.items()}

        return WebSocketMessage(event_name="get_node_info_result", payload=node_data, to=message.from_id)
    except IndexError as e:
        return WebSocketMessage(event_name="get_node_info_result", payload={"error": str(e)}, to=message.from_id)
    except Exception as e:
        return WebSocketMessage(event_name="get_node_info_result", payload={"error": str(e)}, to=message.from_id)
handle_client_overview(message) async

Handle client overview messages.

Source code in datadivr/handlers/custom_handlers.py 
11
12
13
14
15
16
17
18
19
20
21
22
23
@websocket_handler("client_overview", HandlerType.CLIENT)
async def handle_client_overview(message: WebSocketMessage) -> None:
    """Handle client overview messages."""
    if not message.payload:
        return

    overview = message.payload
    logger.info(
        "Client Overview Update: ",
        clients=len(overview["client_ids"]),
        ids=overview["client_ids"],
        time=overview["timestamp"],
    )

registry

Classes
HandlerType

Bases: Enum

Type of handler to register.

Source code in datadivr/handlers/registry.py 
11
12
13
14
15
16
class HandlerType(Enum):
    """Type of handler to register."""

    SERVER = auto()
    CLIENT = auto()
    BOTH = auto()
Functions
get_handlers(handler_type=HandlerType.SERVER)

Get registered handlers for the specified type.

Parameters:

Name Type Description Default
handler_type HandlerType

Type of handlers to retrieve (SERVER or CLIENT)

 SERVER
Source code in datadivr/handlers/registry.py 
24
25
26
27
28
29
30
31
32
33
34
35
def get_handlers(
    handler_type: HandlerType = HandlerType.SERVER,
) -> dict[str, Callable[[WebSocketMessage], Awaitable[WebSocketMessage | None]]]:
    """
    Get registered handlers for the specified type.

    Args:
        handler_type: Type of handlers to retrieve (SERVER or CLIENT)
    """
    if handler_type == HandlerType.SERVER:
        return _server_handlers
    return _client_handlers
websocket_handler(event_name, handler_type=HandlerType.SERVER)

Decorator to register a websocket handler function.

Parameters:

Name Type Description Default
event_name str

The event name to register the handler for.

 required
handler_type HandlerType

Where this handler should be registered (SERVER, CLIENT, or BOTH)

 SERVER
Example

@websocket_handler("sum_event", HandlerType.BOTH) async def sum_handler(message: WebSocketMessage) -> Optional[WebSocketMessage]: ...

Source code in datadivr/handlers/registry.py 
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
def websocket_handler(
    event_name: str, handler_type: HandlerType = HandlerType.SERVER
) -> Callable[[Callable[..., Awaitable[WebSocketMessage | None]]], Callable[..., Awaitable[WebSocketMessage | None]]]:
    """
    Decorator to register a websocket handler function.

    Args:
        event_name: The event name to register the handler for.
        handler_type: Where this handler should be registered (SERVER, CLIENT, or BOTH)

    Example:
        @websocket_handler("sum_event", HandlerType.BOTH)
        async def sum_handler(message: WebSocketMessage) -> Optional[WebSocketMessage]:
            ...
    """

    def decorator(
        func: Callable[..., Awaitable[WebSocketMessage | None]],
    ) -> Callable[..., Awaitable[WebSocketMessage | None]]:
        @wraps(func)
        async def wrapper(*args: Any, **kwargs: Any) -> WebSocketMessage | None:
            return await func(*args, **kwargs)

        if handler_type in (HandlerType.SERVER, HandlerType.BOTH):
            _server_handlers[event_name] = wrapper
        if handler_type in (HandlerType.CLIENT, HandlerType.BOTH):
            _client_handlers[event_name] = wrapper

        return wrapper

    return decorator

options: show_root_heading: true show_source: true