TraceView API#

def add_spans(self, spans: Iterable[Any]) -> None:
    """Add (or update, by ``span_id``) OpenTelemetry SDK spans or span dicts."""
    with self._lock:
        for span in spans:
            data = _normalize_span(span) if isinstance(span, dict) else span_to_dict(span)
            self._by_id[data["span_id"]] = data
    self._sync()

@contextmanager
def capture_weave(self) -> Iterator[None]:
    """Show the W&B Weave ``@weave.op`` calls made inside this ``with`` block, live.

    Needs ``weave.init(...)`` first; the calls still go to W&B as usual.
    Weave has no public call hook, so while the block runs this wraps the
    client's ``create_call``/``finish_call``.
    """
    import weave

    client = weave.get_client()
    if client is None:
        raise RuntimeError("no Weave client: call weave.init(...) first")
    create_call, finish_call = client.create_call, client.finish_call

    def create_call_and_record(*args: Any, **kwargs: Any) -> Any:
        call = create_call(*args, **kwargs)
        self._record_call(call)
        return call

    def finish_call_and_record(call: Any, *args: Any, **kwargs: Any) -> Any:
        result = finish_call(call, *args, **kwargs)
        self._record_call(call)
        return result

    client.create_call, client.finish_call = create_call_and_record, finish_call_and_record
    try:
        yield
    finally:
        client.create_call, client.finish_call = create_call, finish_call
        self._sync()  # send any update the throttle held back

def clear(self) -> None:
    """Drop all spans and reset the selection, replay cursor, and collapsed rows."""
    with self._lock:
        self._by_id = {}
        self._running = {}
        self._trace_ids = set()
    self.spans = []
    self.trace_id = ""
    self.selected_span_id = None
    self.cursor_ms = None
    self.collapsed = []

@classmethod
def from_otlp(
    cls, source: str | Path | dict[str, Any] | list[dict[str, Any]], **kwargs: Any
) -> "TraceView":
    """Load OTLP/JSON: a file path (a JSON document or JSON lines), a dict, or a list of dicts."""
    if isinstance(source, (str, Path)):
        text = Path(source).read_text()
        try:
            docs = [json.loads(text)]
        except json.JSONDecodeError:  # the OTLP file exporter writes one doc per line
            docs = [json.loads(line) for line in text.splitlines() if line.strip()]
    elif isinstance(source, dict):
        docs = [source]
    else:
        docs = list(source)
    spans = [span for doc in docs for span in spans_from_otlp(doc)]
    return cls(spans, **kwargs)

@classmethod
def from_spans(cls, spans: Iterable[Any], **kwargs: Any) -> "TraceView":
    """Build from OpenTelemetry SDK spans, e.g. ``InMemorySpanExporter.get_finished_spans()``."""
    return cls([span_to_dict(span) for span in spans], **kwargs)

@classmethod
def from_weave(cls, calls: Iterable[Any], **kwargs: Any) -> "TraceView":
    """Build from W&B Weave calls, e.g. ``client.get_calls(filter={"trace_ids": [...]})``.

    Accepts the SDK's call objects or plain dicts (service API JSON); no
    ``weave`` install needed for the dict form.
    """
    return cls([call_to_dict(call) for call in calls], **kwargs)

def span_processor(self) -> Any:
    """An OpenTelemetry ``SpanProcessor`` that streams spans into this widget.

    Add it to a ``TracerProvider`` with ``provider.add_span_processor(...)``.
    Updates are sent from the thread that starts or ends a span; spans that
    finish on background threads appear with the next update from the
    notebook's own thread.
    """
    from opentelemetry.sdk.trace import SpanProcessor

    view = self

    class TraceViewProcessor(SpanProcessor):
        def on_start(self, span: Any, parent_context: Any = None) -> None:
            view._record(span)

        def on_end(self, span: Any) -> None:
            view._record(span)

    return TraceViewProcessor()