briefcase-python 2.4.1

Python bindings for Briefcase AI
Documentation 
1
2
3
4
5
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
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

"""
Central event emission helpers.

emit() dispatches a BriefcaseEvent to all configured sinks
(webhook + event_bus) according to BriefcaseConfig.

Helper factories:
  emit_low_confidence(decision, confidence, threshold)
  emit_drift_detected(decision, details)
"""

from __future__ import annotations

import asyncio
from typing import Any, Dict, Optional

from briefcase.config import BriefcaseConfig
from briefcase.events.types import BriefcaseEvent


async def emit(event: BriefcaseEvent) -> None:
    """Dispatch *event* to webhook and/or event_bus if configured."""
    config = BriefcaseConfig.get()

    # --- Webhook ---
    if config.webhook_url:
        from briefcase.events.webhook import WebhookEmitter

        emitter = WebhookEmitter(
            url=config.webhook_url,
            secret=config.webhook_secret or "",
            subscribed_events=config.subscribed_events or [],
        )
        await emitter.emit(event)

    # --- Event bus (e.g. KafkaPublisher) ---
    if config.event_bus is not None:
        bus = config.event_bus
        if asyncio.iscoroutinefunction(getattr(bus, "emit", None)):
            await bus.emit(event)
        elif hasattr(bus, "publish"):
            publish = bus.publish
            if asyncio.iscoroutinefunction(publish):
                await publish(event)
            else:
                publish(event)


async def emit_low_confidence(
    decision: Any,
    confidence: float,
    threshold: float,
) -> None:
    """Emit a 'decision.low_confidence' event."""
    decision_id = (
        getattr(decision, "decision_id", None)
        or getattr(decision, "id", None)
        or ""
    )
    event = BriefcaseEvent(
        event_type="decision.low_confidence",
        decision_id=str(decision_id),
        payload={
            "confidence": confidence,
            "threshold": threshold,
        },
    )
    await emit(event)


async def emit_drift_detected(
    decision: Any,
    details: Optional[Dict[str, Any]] = None,
) -> None:
    """Emit a 'drift.detected' event."""
    decision_id = (
        getattr(decision, "decision_id", None)
        or getattr(decision, "id", None)
        or ""
    )
    event = BriefcaseEvent(
        event_type="drift.detected",
        decision_id=str(decision_id),
        payload=details or {},
    )
    await emit(event)