# 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 LockingBase dependencies with vis-network.js."""
from collections import defaultdict
from collections.abc import Mapping
from pathlib import Path
import re
from tempfile import TemporaryDirectory
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__ = ["LockingBaseGraphServer"]
_DOT_EDGE_PATTERN = re.compile(r"^\s*(\d+)\s*->\s*(\d+)\s*;", re.MULTILINE)
_DEFAULT_EDGE_COLOR = "#848484"
_TRANSITIVE_EDGE_COLOR = "#FF0000"
_TRANSITIVE_PATH_EDGE_COLOR = "#0072B2"
_HIGHLIGHT_COLOR = "#FF00FF"
def _lockingBases(base_container: kc.BaseContainer) -> list[kc.LockingBase]:
"""Return a stable snapshot of the container's LockingBase objects."""
container_snapshot = list(base_container)
locking_bases = []
for object_id, _ in container_snapshot:
locking_base = base_container.atLockingBase(object_id)
if locking_base is not None:
locking_bases.append(locking_base)
return sorted(locking_bases, key=lambda locking_base: locking_base.id())
def _graphvizDownstreamEdges(locking_base: kc.LockingBase) -> set[tuple[int, int]]:
"""Read direct downstream edges using kdFlex's ID-based debug output."""
with TemporaryDirectory(prefix="kdflex-lockingbase-") as temporary_directory:
dot_path = Path(temporary_directory) / "dependencies.dot"
kc.DebugManager.dumpDependencyGraphviz(
locking_base,
dot_path,
downstream_depth=1,
upstream_depth=0,
)
dot_text = dot_path.read_text(encoding="utf-8")
return {
(int(source_id), int(target_id))
for source_id, target_id in _DOT_EDGE_PATTERN.findall(dot_text)
if int(source_id) == locking_base.id()
}
def _dependencyEdges(locking_bases: list[kc.LockingBase]) -> set[tuple[int, int]]:
"""Find direct dependency-to-dependent edges for the supplied objects."""
edges: set[tuple[int, int]] = set()
for lb in locking_bases:
edges |= set((lb.id(), x.id()) for x in kc.DebugManager.getDownstreamDeps(lb))
known_ids = {locking_base.id() for locking_base in locking_bases}
return {
(source_id, target_id)
for source_id, target_id in edges
if source_id in known_ids and target_id in known_ids
}
def _transitiveEdges(edges: set[tuple[int, int]]) -> set[tuple[int, int]]:
"""Return edges whose endpoints are also connected by another path."""
downstream: dict[int, set[int]] = defaultdict(set)
for source_id, target_id in edges:
downstream[source_id].add(target_id)
redundant_edges = set()
for source_id, target_id in edges:
pending = list(downstream[source_id] - {target_id})
visited = {source_id}
while pending:
candidate_id = pending.pop()
if candidate_id == target_id:
redundant_edges.add((source_id, target_id))
break
if candidate_id in visited:
continue
visited.add(candidate_id)
pending.extend(downstream[candidate_id] - visited)
return redundant_edges
def _reachableNodes(
start_id: int,
adjacency: Mapping[int, set[int]],
excluded_edge: tuple[int, int],
*,
reverse: bool = False,
) -> set[int]:
"""Return nodes reachable without traversing the excluded edge."""
visited = {start_id}
pending = [start_id]
while pending:
current_id = pending.pop()
for candidate_id in adjacency.get(current_id, set()):
traversed_edge = (candidate_id, current_id) if reverse else (current_id, candidate_id)
if traversed_edge == excluded_edge or candidate_id in visited:
continue
visited.add(candidate_id)
pending.append(candidate_id)
return visited
def _transitivePathEdges(
edges: set[tuple[int, int]],
transitive_edges: set[tuple[int, int]],
) -> set[tuple[int, int]]:
"""Return edges on alternate paths parallel to transitive edges."""
downstream: dict[int, set[int]] = defaultdict(set)
upstream: dict[int, set[int]] = defaultdict(set)
for source_id, target_id in edges:
downstream[source_id].add(target_id)
upstream[target_id].add(source_id)
path_edges = set()
for transitive_edge in transitive_edges:
source_id, target_id = transitive_edge
reachable_from_source = _reachableNodes(
source_id,
downstream,
transitive_edge,
)
can_reach_target = _reachableNodes(
target_id,
upstream,
transitive_edge,
reverse=True,
)
path_edges.update(
edge
for edge in edges - {transitive_edge}
if edge[0] in reachable_from_source and edge[1] in can_reach_target
)
return path_edges
def _toNode(
locking_base: kc.LockingBase,
label_map: Mapping[int | str, str],
) -> Node:
"""Create a visjs node for a LockingBase."""
healthy = locking_base.isHealthy()
color = "#B7E4C7" if healthy else "#FFB4A2"
label = label_map.get(
locking_base.id(),
label_map.get(locking_base.name(), locking_base.name()),
)
return Node(
id=locking_base.id(),
label=str(label),
title=(
f"{locking_base.typeString()}\n"
f"{locking_base.name()} (id={locking_base.id()})\n"
f"healthy={healthy}"
),
options=NodeOptions(
shape="box",
size=25,
borderWidth=2,
color=NodeColorOptions(
background=color,
border=color,
highlight={"border": _HIGHLIGHT_COLOR, "background": color},
),
font=NodeFontOptions(color="#000000", size=12, face="Arial"),
),
)
def _toEdge(
source_id: int,
target_id: int,
transitive_edges: set[tuple[int, int]],
transitive_path_edges: set[tuple[int, int]],
) -> Edge:
"""Create a visjs edge for a LockingBase dependency."""
is_transitive = (source_id, target_id) in transitive_edges
is_transitive_path = (source_id, target_id) in transitive_path_edges
if is_transitive:
color = _TRANSITIVE_EDGE_COLOR
title = "Transitive dependency (removable)"
elif is_transitive_path:
color = _TRANSITIVE_PATH_EDGE_COLOR
title = "Alternate path for a transitive dependency"
else:
color = _DEFAULT_EDGE_COLOR
title = "Dependency"
return Edge(
from_=source_id,
to=target_id,
id=f"lockingbase-{source_id}-{target_id}",
title=title,
options=EdgeOptions(
color=EdgeColorOptions(color=color, highlight=_HIGHLIGHT_COLOR),
width=3 if is_transitive or is_transitive_path else 2,
dashes=False,
arrows=ArrowOptions(
to=ArrowStyle(enabled=True, scaleFactor=1.2),
),
smooth=SmoothOptions(type="cubicBezier", roundness=0.3),
),
)
# The stored graph configuration mirrors FSMGraphServer, and the constructor
# deliberately exposes the corresponding graph/server options in one place.
# pylint: disable=too-many-instance-attributes,too-few-public-methods
class LockingBaseGraphServer(GraphServer):
"""Specialized GraphServer for viewing all LockingBase dependencies.
Edges point from a dependency to its dependent, matching the direction
established by ``dependency.addDependent(dependent)``.
"""
def __init__(
self,
base_container: kc.BaseContainer | None = None,
*,
port: int = 0,
title: str = "LockingBase dependencies",
label_map: Mapping[int | str, str] | None = None,
highlight_transitiveEdges: bool = True,
network_options: NetworkOptions | None = None,
): # pylint: disable=too-many-arguments
"""Create and bind a LockingBaseGraphServer.
Parameters
----------
base_container : kc.BaseContainer | None
Container whose LockingBases should be displayed. The singleton
BaseContainer is used by default.
port : int
Port to bind. Port 0 selects an arbitrary free port.
title : str
Graph title.
label_map : Mapping[int | str, str] | None
Optional labels keyed by a LockingBase ID or name. ID labels take
precedence over name labels.
highlight_transitiveEdges : bool
If true, color a direct edge red when the same source and target
are connected by another path, meaning the edge is removable by a
transitive reduction. Edges on the alternate path are colored
blue.
network_options : NetworkOptions | None
visjs network options. Default options are used when omitted.
"""
self._base_container = (
kc.BaseContainer.singleton() if base_container is None else base_container
)
self._title = title
self._label_map = dict(label_map or {})
self._highlight_transitive_edges = highlight_transitiveEdges
self._network_options = NetworkOptions() if network_options is None else network_options
self.nodes: dict[int, Node] = {}
self.edges: dict[str, Edge] = {}
self._buildGraph()
super().__init__(graph=self.graph, port=port, buttons=[])
def _buildGraph(self):
"""Build the graph from the current BaseContainer state."""
locking_bases = _lockingBases(self._base_container)
dependencies = _dependencyEdges(locking_bases)
transitive_edges = (
_transitiveEdges(dependencies) if self._highlight_transitive_edges else set()
)
transitive_path_edges = _transitivePathEdges(
dependencies,
transitive_edges,
)
self.nodes = {
locking_base.id(): _toNode(locking_base, self._label_map)
for locking_base in locking_bases
}
graph_edges = [
_toEdge(
source_id,
target_id,
transitive_edges,
transitive_path_edges,
)
for source_id, target_id in sorted(dependencies)
]
self.edges = {str(edge.id): edge for edge in graph_edges}
self.graph = NetworkGraph(
nodes=list(self.nodes.values()),
edges=list(self.edges.values()),
title=self._title,
options=self._network_options,
)
[docs]
def refresh(self, *, highlight_transitiveEdges: bool | None = None):
"""Rebuild the dependency graph and broadcast it to connected clients.
Parameters
----------
highlight_transitiveEdges : bool | None
Optionally change transitive-edge highlighting before rebuilding.
If omitted, the current setting is retained.
"""
if highlight_transitiveEdges is not None:
self._highlight_transitive_edges = highlight_transitiveEdges
self._buildGraph()
self.updateClientGraphs()
