Module `panopticon.trace`

Classes to generate a tracefile in the right format

Expand source code

#!/bin/env python3

"""Classes to generate a tracefile in the right format"""

from __future__ import annotations

import io
import json
import os
import threading
from dataclasses import asdict, dataclass, field
from enum import Enum
from time import perf_counter_ns
from typing import Any, Dict, Optional

from panopticon.version import version


class Trace:
    def __init__(self):
        self._events = []

    def add_event(self, event: TraceEvent):
        self._events.append(event)

    def __str__(self) -> str:
        return json.dumps(
            {
                "traceEvents": [asdict(x) for x in self._events],
                "displayTimeUnit": "ns",
                "otherData": {"version": f"Panopticon {version}"},
            },
            indent="  ",
        )


class StreamingTrace(Trace):
    """Streams data to a file without keeping it in memory"""

    def __init__(self, stream: io.IOBase):
        self._out = stream
        self._out.write("[\n")  # Opening brace
        self._out.flush()

    def add_event(self, event: TraceEvent):
        json.dump(asdict(event), self._out)
        self._out.write(",\n")
        self._out.flush()

    def __str__(self) -> str:
        if hasattr(self._out, "getvalue") and callable(
            getattr(self._out, "getvalue")
        ):
            return self._out.getvalue()
        else:
            return f"StreamingTrace ({self._out})"


class _SerializableEnum(str, Enum):
    ...


class Phase:
    class Duration(_SerializableEnum):
        START = "B"
        END = "E"

    class Instant(_SerializableEnum):
        INSTANT = "i"

    class Counter(_SerializableEnum):
        COUNTER = "c"

    class Complete(_SerializableEnum):
        INSTANT = "X"

    class Async(_SerializableEnum):
        START = "b"
        INSTANT = "n"
        END = "e"

    class Flow(_SerializableEnum):
        START = "s"
        INSTANT = "t"
        END = "f"

    class Object(_SerializableEnum):
        NEW = "N"
        SNAPSHOT = "O"
        DESTROY = "D"


@dataclass
class TraceEvent:
    name: str
    cat: str
    ph: str
    args: Optional[Dict[str, Any]] = None
    ts: int = field(init=False)
    pid: int = field(init=False)
    tid: int = field(init=False)

    def __post_init__(self):
        self.ts = perf_counter_ns() / 1000
        self.pid = os.getpid()
        self.tid = _get_thread_id()


@dataclass
class DurationTraceEvent(TraceEvent):
    ph: Phase.Duration


class InstantScope(_SerializableEnum):
    GLOBAL = "g"
    PROCESS = "p"
    THREAD = "t"


@dataclass
class InstantTraceEvent(TraceEvent):

    ph: Phase.Instant = Phase.Instant.INSTANT
    s: InstantScope = InstantScope.THREAD


class FlowBindingPoint(_SerializableEnum):
    ENCLOSING = "e"
    NEXT = "n"


@dataclass
class FlowTraceEvent(TraceEvent):
    id: int = 0  # ick
    ph: Phase.Flow = Phase.Flow.START
    bp: FlowBindingPoint = FlowBindingPoint.ENCLOSING


def _get_thread_id() -> int:
    try:
        return threading.get_native_id()
    except AttributeError:
        return threading.get_ident()

Classes

class DurationTraceEvent (name: str, cat: str, ph: Phase.Duration, args: Optional[Dict[str, Any]] = None)

DurationTraceEvent(name: 'str', cat: 'str', ph: 'Phase.Duration', args: 'Optional[Dict[str, Any]]' = None)

Expand source code

class DurationTraceEvent(TraceEvent):
    ph: Phase.Duration

Ancestors

TraceEvent

Class variables

var ph : Phase.Duration

class FlowBindingPoint (value, names=None, *, module=None, qualname=None, type=None, start=1)

An enumeration.

Expand source code

class FlowBindingPoint(_SerializableEnum):
    ENCLOSING = "e"
    NEXT = "n"

Ancestors

panopticon.trace._SerializableEnum
builtins.str
enum.Enum

Class variables

var ENCLOSING
var NEXT

class FlowTraceEvent (name: str, cat: str, ph: Phase.Flow = Flow.START, args: Optional[Dict[str, Any]] = None, id: int = 0, bp: FlowBindingPoint = FlowBindingPoint.ENCLOSING)

FlowTraceEvent(name: 'str', cat: 'str', ph: 'Phase.Flow' = , args: 'Optional[Dict[str, Any]]' = None, id: 'int' = 0, bp: 'FlowBindingPoint' = )

Expand source code

class FlowTraceEvent(TraceEvent):
    id: int = 0  # ick
    ph: Phase.Flow = Phase.Flow.START
    bp: FlowBindingPoint = FlowBindingPoint.ENCLOSING

Ancestors

TraceEvent

Class variables

var bp : FlowBindingPoint
var id : int
var ph : Phase.Flow

class InstantScope (value, names=None, *, module=None, qualname=None, type=None, start=1)

An enumeration.

Expand source code

class InstantScope(_SerializableEnum):
    GLOBAL = "g"
    PROCESS = "p"
    THREAD = "t"

Ancestors

panopticon.trace._SerializableEnum
builtins.str
enum.Enum

Class variables

var GLOBAL
var PROCESS
var THREAD

class InstantTraceEvent (name: str, cat: str, ph: Phase.Instant = Instant.INSTANT, args: Optional[Dict[str, Any]] = None, s: InstantScope = InstantScope.THREAD)

InstantTraceEvent(name: 'str', cat: 'str', ph: 'Phase.Instant' = , args: 'Optional[Dict[str, Any]]' = None, s: 'InstantScope' = )

Expand source code

class InstantTraceEvent(TraceEvent):

    ph: Phase.Instant = Phase.Instant.INSTANT
    s: InstantScope = InstantScope.THREAD

Ancestors

TraceEvent

Class variables

var ph : Phase.Instant
var s : InstantScope

class Phase

Expand source code

class Phase:
    class Duration(_SerializableEnum):
        START = "B"
        END = "E"

    class Instant(_SerializableEnum):
        INSTANT = "i"

    class Counter(_SerializableEnum):
        COUNTER = "c"

    class Complete(_SerializableEnum):
        INSTANT = "X"

    class Async(_SerializableEnum):
        START = "b"
        INSTANT = "n"
        END = "e"

    class Flow(_SerializableEnum):
        START = "s"
        INSTANT = "t"
        END = "f"

    class Object(_SerializableEnum):
        NEW = "N"
        SNAPSHOT = "O"
        DESTROY = "D"

Class variables

var Async: An enumeration.
var Complete: An enumeration.
var Counter: An enumeration.
var Duration: An enumeration.
var Flow: An enumeration.
var Instant: An enumeration.
var Object: An enumeration.

class StreamingTrace (stream: io.IOBase)

Streams data to a file without keeping it in memory

Expand source code

class StreamingTrace(Trace):
    """Streams data to a file without keeping it in memory"""

    def __init__(self, stream: io.IOBase):
        self._out = stream
        self._out.write("[\n")  # Opening brace
        self._out.flush()

    def add_event(self, event: TraceEvent):
        json.dump(asdict(event), self._out)
        self._out.write(",\n")
        self._out.flush()

    def __str__(self) -> str:
        if hasattr(self._out, "getvalue") and callable(
            getattr(self._out, "getvalue")
        ):
            return self._out.getvalue()
        else:
            return f"StreamingTrace ({self._out})"

Ancestors

Trace

Methods

def add_event(self, event: TraceEvent)

Expand source code

def add_event(self, event: TraceEvent):
    json.dump(asdict(event), self._out)
    self._out.write(",\n")
    self._out.flush()

class Trace

Expand source code

class Trace:
    def __init__(self):
        self._events = []

    def add_event(self, event: TraceEvent):
        self._events.append(event)

    def __str__(self) -> str:
        return json.dumps(
            {
                "traceEvents": [asdict(x) for x in self._events],
                "displayTimeUnit": "ns",
                "otherData": {"version": f"Panopticon {version}"},
            },
            indent="  ",
        )

Subclasses

StreamingTrace

Methods

def add_event(self, event: TraceEvent)

Expand source code

def add_event(self, event: TraceEvent):
    self._events.append(event)

class TraceEvent (name: str, cat: str, ph: str, args: Optional[Dict[str, Any]] = None)

TraceEvent(name: 'str', cat: 'str', ph: 'str', args: 'Optional[Dict[str, Any]]' = None)

Expand source code

class TraceEvent:
    name: str
    cat: str
    ph: str
    args: Optional[Dict[str, Any]] = None
    ts: int = field(init=False)
    pid: int = field(init=False)
    tid: int = field(init=False)

    def __post_init__(self):
        self.ts = perf_counter_ns() / 1000
        self.pid = os.getpid()
        self.tid = _get_thread_id()

Class variables

var args : Union[Dict[str, Any], NoneType]
var cat : str
var name : str
var ph : str
var pid : int
var tid : int
var ts : int