# Copyright (c) 2024-2026 Karana Dynamics Pty Ltd. All rights reserved.
#
# NOTICE TO USER:
#
# This source code and/or documentation (the "Licensed Materials") is
# the confidential and proprietary information of Karana Dynamics Inc.
# Use of these Licensed Materials is governed by the terms and conditions
# of a separate software license agreement between Karana Dynamics and the
# Licensee ("License Agreement"). Unless expressly permitted under that
# agreement, any reproduction, modification, distribution, or disclosure
# of the Licensed Materials, in whole or in part, to any third party
# without the prior written consent of Karana Dynamics is strictly prohibited.
#
# THE LICENSED MATERIALS ARE PROVIDED "AS IS" WITHOUT WARRANTY OF ANY KIND.
# KARANA DYNAMICS DISCLAIMS ALL WARRANTIES, EXPRESS OR IMPLIED, INCLUDING
# BUT NOT LIMITED TO WARRANTIES OF MERCHANTABILITY, NON-INFRINGEMENT, AND
# FITNESS FOR A PARTICULAR PURPOSE.
#
# IN NO EVENT SHALL KARANA DYNAMICS BE LIABLE FOR ANY DAMAGES WHATSOEVER,
# INCLUDING BUT NOT LIMITED TO LOSS OF PROFITS, DATA, OR USE, EVEN IF
# ADVISED OF THE POSSIBILITY OF SUCH DAMAGES, WHETHER IN CONTRACT, TORT,
# OR OTHERWISE ARISING OUT OF OR IN CONNECTION WITH THE LICENSED MATERIALS.
#
# U.S. Government End Users: The Licensed Materials are a "commercial item"
# as defined at 48 C.F.R. 2.101, and are provided to the U.S. Government
# only as a commercial end item under the terms of this license.
#
# Any use of the Licensed Materials in individual or commercial software must
# include, in the user documentation and internal source code comments,
# this Notice, Disclaimer, and U.S. Government Use Provision.
"""Tools for viewing Python object referrers with vis-network.js."""
from collections import deque
from collections.abc import Callable
import gc
import types
import weakref
import Karana.Core as kc
from ._datatypes import (
ArrowOptions,
ArrowStyle,
Edge,
EdgeColorOptions,
EdgeOptions,
NetworkGraph,
NetworkOptions,
Node,
NodeColorOptions,
NodeFontOptions,
NodeOptions,
SmoothOptions,
)
from ._server import GraphServer
__all__ = ["PythonReferenceGraphServer"]
_TARGET_COLOR = "#FFB4A2"
_DIRECT_REFERRER_COLOR = "#90CAF9"
_INDIRECT_REFERRER_COLOR = "#DDEAF7"
_COLLECTED_COLOR = "#BDBDBD"
_EDGE_COLOR = "#607D8B"
_HIGHLIGHT_COLOR = "#FF00FF"
_MAX_TEXT_LENGTH = 160
def _safeRepr(obj: object, max_length: int = _MAX_TEXT_LENGTH) -> str:
"""Return a bounded representation that tolerates broken repr methods."""
try:
text = repr(obj)
except Exception as error:
text = f"<repr failed: {type(error).__name__}: {error}>"
if len(text) > max_length:
return f"{text[: max_length - 3]}..."
return text
def _qualifiedTypeName(obj: object) -> str:
"""Return the object's module-qualified type name."""
object_type = type(obj)
return f"{object_type.__module__}.{object_type.__qualname__}"
def _objectLabel(obj: object, *, target_label: str | None = None) -> str:
"""Create a compact node label for an arbitrary Python object."""
if target_label is not None:
label = target_label
elif isinstance(obj, kc.Base):
label = f"{obj.typeString()}\n{obj.name()}"
elif isinstance(obj, types.FrameType):
label = f"frame\n{obj.f_code.co_name}"
elif isinstance(obj, types.ModuleType):
label = f"module\n{obj.__name__}"
elif isinstance(obj, dict):
namespace = obj.get("__name__")
if isinstance(namespace, str) and "__builtins__" in obj:
label = f"namespace\n{namespace}"
else:
label = f"dict ({len(obj)} items)"
elif isinstance(obj, (list, tuple, set, frozenset, deque)):
label = f"{type(obj).__name__} ({len(obj)} items)"
else:
label = type(obj).__name__
return label
def _objectTitle(obj: object) -> str:
"""Create detailed tooltip text without retaining the object."""
lines = [
_qualifiedTypeName(obj),
f"python_id=0x{id(obj):x}",
]
if isinstance(obj, kc.Base):
lines.extend(
(
f"kdflex_id={obj.id()}",
f"name={obj.name()}",
)
)
if isinstance(obj, types.FrameType):
lines.extend(
(
f"function={obj.f_code.co_name}",
f"file={obj.f_code.co_filename}:{obj.f_lineno}",
)
)
else:
lines.append(f"repr={_safeRepr(obj)}")
return "\n".join(lines)
def _nodeFor(
obj: object,
*,
depth: int,
max_depth: int,
target_label: str | None = None,
) -> Node:
"""Create a node without storing the referenced Python object."""
if depth == 0:
color = _TARGET_COLOR
shape = "diamond"
elif depth == 1:
color = _DIRECT_REFERRER_COLOR
shape = "box"
else:
color = _INDIRECT_REFERRER_COLOR
shape = "box"
return Node(
id=id(obj),
label=_objectLabel(obj, target_label=target_label),
title=_objectTitle(obj),
options=NodeOptions(
shape=shape,
level=max_depth - depth,
borderWidth=2,
color=NodeColorOptions(
background=color,
border=color,
highlight={"border": _HIGHLIGHT_COLOR, "background": color},
),
font=NodeFontOptions(color="#FFFFFF", size=12, face="Arial"),
),
)
def _matchingNames(mapping: dict, referent: object) -> list[str]:
"""Return bounded key names whose values are the referent."""
names = []
for key, value in mapping.items():
if value is referent:
names.append(_safeRepr(key, max_length=50))
if len(names) == 4:
break
return names
def _relationLabel(owner: object, referent: object) -> str:
"""Describe how an owner directly references the referent."""
if isinstance(owner, dict):
names = _matchingNames(owner, referent)
return ", ".join(names) if names else "value"
if isinstance(owner, (list, tuple, deque)):
indices = [str(index) for index, value in enumerate(owner) if value is referent][:4]
return f"index {', '.join(indices)}" if indices else "element"
if isinstance(owner, (set, frozenset)):
return "member"
if isinstance(owner, types.FrameType):
names = _matchingNames(owner.f_locals, referent)
return ", ".join(names) if names else "local"
if isinstance(owner, types.CellType):
return "closure cell"
try:
attributes = _matchingNames(vars(owner), referent)
except TypeError:
attributes = []
return ", ".join(attributes) if attributes else "reference"
def _isInternalReferrer(obj: object, ignored_ids: set[int]) -> bool:
"""Exclude traversal artifacts and this module's own execution frames."""
if id(obj) in ignored_ids:
return True
if isinstance(obj, types.FrameType):
return obj.f_globals.get("__name__") == __name__
if isinstance(obj, dict):
return obj.get("__name__") == __name__
return False
def _isTerminalOwner(obj: object) -> bool:
"""Avoid expanding interpreter roots that produce low-value fan-out."""
if isinstance(obj, (types.FrameType, types.ModuleType)):
return True
return isinstance(obj, dict) and isinstance(obj.get("__name__"), str) and "__builtins__" in obj
def _targetHandle(target: object) -> Callable[[], object | None]:
"""Create a non-owning target handle appropriate for the object."""
if isinstance(target, kc.Base):
return kc.CppWeakRefBase(target)
try:
return weakref.ref(target)
except TypeError as error:
raise TypeError(
"PythonReferenceGraphServer requires a weak-referenceable target. "
"kdFlex Base objects and ordinary user-defined class instances are "
"supported; built-in containers such as list and dict are not."
) from error
# The graph server stores traversal limits and a non-owning target handle.
class PythonReferenceGraphServer(GraphServer):
"""GraphServer showing which Python objects retain a target object.
Every edge points from an owner to the object it directly references.
Traversal proceeds away from the target through ``gc.get_referrers``.
"""
def __init__(
self,
target: object,
*,
port: int = 0,
title: str = "Python reference owners",
target_label: str | None = None,
max_depth: int = 2,
max_nodes: int = 100,
network_options: NetworkOptions | None = None,
):
"""Create and bind a PythonReferenceGraphServer.
Parameters
----------
target : object
Weak-referenceable object whose Python referrers should be shown.
kdFlex Base objects use ``CppWeakRefBase`` automatically.
port : int
Port to bind. Port 0 selects an arbitrary free port.
title : str
Graph title.
target_label : str | None
Optional display label for the target node.
max_depth : int
Maximum number of referrer links to traverse from the target.
max_nodes : int
Maximum number of nodes, including the target.
network_options : NetworkOptions | None
visjs network options. Default options are used when omitted.
"""
if max_depth < 0:
raise ValueError("max_depth must be greater than or equal to zero")
if max_nodes < 1:
raise ValueError("max_nodes must be greater than or equal to one")
self._target_ref = _targetHandle(target)
self._target_id = id(target)
self._title = title
self._target_label = target_label
self._max_depth = max_depth
self._max_nodes = max_nodes
self._network_options = NetworkOptions() if network_options is None else network_options
self.nodes: dict[int | str, Node] = {}
self.edges: dict[str, Edge] = {}
self.truncated = False
self.graph = NetworkGraph()
self.initial_graph_pending = True
# Do not call gc.get_referrers() while target is still present in this
# constructor's temporary call arguments. The first client connection
# performs the full traversal after those objects have been released.
self.nodes = {
id(target): _nodeFor(
target,
depth=0,
max_depth=self._max_depth,
target_label=self._target_label,
)
}
self._updateGraph()
super().__init__(graph=self.graph, port=port, buttons=[])
def _ensureInitialGraph(self):
"""Build the first full graph after constructor temporaries are gone."""
if not self.initial_graph_pending:
return
self._buildGraph()
self.initial_graph_pending = False
[docs]
def onConnect(self, client_id: int):
"""Build a clean initial graph before initializing the first client."""
self._ensureInitialGraph()
super().onConnect(client_id)
def _buildGraph(self, target: object | None = None):
"""Rebuild the graph without retaining any traversed objects."""
if target is None:
target = self._target_ref()
if target is None:
collected_node = Node(
id=f"collected-{self._target_id}",
label=self._target_label or "Target collected",
title=f"Former python_id=0x{self._target_id:x}",
options=NodeOptions(
shape="diamond",
color=NodeColorOptions(
background=_COLLECTED_COLOR,
border=_COLLECTED_COLOR,
),
),
)
self.nodes = {collected_node.id: collected_node}
self.edges = {}
self.truncated = False
self._updateGraph()
return
nodes: dict[int | str, Node] = {
id(target): _nodeFor(
target,
depth=0,
max_depth=self._max_depth,
target_label=self._target_label,
)
}
edges: dict[str, Edge] = {}
seen_ids = {id(target)}
pending = deque([(target, 0)])
ignored_ids = {
id(self),
id(nodes),
id(edges),
id(seen_ids),
id(pending),
}
self.truncated = False
while pending:
current, depth = pending.popleft()
if depth >= self._max_depth:
continue
referrers = gc.get_referrers(current)
ignored_ids.add(id(referrers))
for owner in referrers:
owner_id = id(owner)
if _isInternalReferrer(owner, ignored_ids):
continue
if owner_id not in nodes and len(nodes) >= self._max_nodes:
self.truncated = True
continue
owner_depth = depth + 1
if owner_id not in nodes:
nodes[owner_id] = _nodeFor(
owner,
depth=owner_depth,
max_depth=self._max_depth,
)
edge_id = f"pyref-{owner_id}-{id(current)}"
edges[edge_id] = Edge(
from_=owner_id,
to=id(current),
id=edge_id,
title=_relationLabel(owner, current),
options=EdgeOptions(
label=_relationLabel(owner, current),
color=EdgeColorOptions(
color=_EDGE_COLOR,
highlight=_HIGHLIGHT_COLOR,
),
width=2,
arrows=ArrowOptions(
to=ArrowStyle(enabled=True, scaleFactor=1.1),
),
smooth=SmoothOptions(
type="cubicBezier",
roundness=0.25,
),
),
)
if owner_id not in seen_ids and not _isTerminalOwner(owner):
seen_ids.add(owner_id)
pending.append((owner, owner_depth))
del referrers
self.nodes = nodes
self.edges = edges
self._updateGraph()
def _updateGraph(self):
"""Update the stored NetworkGraph from the current node and edge maps."""
title = f"{self._title} (truncated)" if self.truncated else self._title
self.graph = NetworkGraph(
nodes=list(self.nodes.values()),
edges=list(self.edges.values()),
title=title,
options=self._network_options,
)
[docs]
def refresh(self):
"""Rebuild and broadcast the current Python referrer graph."""
self._buildGraph()
self.initial_graph_pending = False
self.updateClientGraphs()
[docs]
def maxDepth(self) -> int:
"""Return the current maximum traversal depth."""
return self._max_depth
[docs]
def maxNodes(self) -> int:
"""Return the current maximum node count."""
return self._max_nodes
[docs]
def setMaxDepth(self, max_depth: int):
"""Set the maximum traversal depth, rebuild, and broadcast the graph."""
if max_depth < 0:
raise ValueError("max_depth must be greater than or equal to zero")
self._max_depth = max_depth
self.refresh()
[docs]
def setMaxNodes(self, max_nodes: int):
"""Set the maximum node count, rebuild, and broadcast the graph."""
if max_nodes < 1:
raise ValueError("max_nodes must be greater than or equal to one")
self._max_nodes = max_nodes
self.refresh()
