# 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 a SubTree with vis-network.js."""
from itertools import count
from dataclasses import dataclass
from typing import cast
from pathlib import Path
from threading import RLock
import json
import math
import os
from tempfile import NamedTemporaryFile
import Karana.Core as kc
from Karana.Core import CppWeakRef, Base, warn
from ._datatypes import (
Node,
Edge,
NodeOptions,
NetworkGraph,
NodeColorOptions,
NodeFontOptions,
EdgeOptions,
EdgeColorOptions,
ArrowOptions,
ArrowStyle,
SmoothOptions,
Button,
)
from ._server import GraphServer
import Karana.Dynamics as kd
__all__ = [
"subTreeToGraph",
"multibodyConstraintEdges",
"MultibodyGraphServer",
"LayoutMultibodyGraphServer",
]
edge_colors = {
kd.HingeType.LOCKED: "#FFFF00", # yellow
kd.HingeType.LOCKED_OFFSET: "#FFFF88", # yellowish
kd.HingeType.SLIDER: "#00FF00", # green
kd.HingeType.REVOLUTE: "bisque", # bisque
kd.HingeType.BALL: "#A52A2A", # brown
# kd.HingeType.UJOINT: "#0FF0FF", # blue
kd.HingeType.UJOINT: "#0070FF", # bright blue
kd.HingeType.CYLINDRICAL: "#800080", # purple
kd.HingeType.TRANSLATIONAL3: "#00FFFF", # cyan
kd.HingeType.DISTANCE_3D: "#AAFF00", # cyan
kd.HingeType.DISTANCE_2D: "#00FFAA", # cyan
kd.HingeType.CUSTOM: "#0000FF", # blue
}
coupler_color = "#808080"
def _fillLabelMap(
subtree: kd.SubTree, label_map: dict[int | str, str] | None = None
) -> dict[int | str, str]:
"""Fill out the label_map with an entry for every body.
This is to ensure the same labels will be used consistently for different
SubTrees of the same multibody
"""
if label_map:
# Avoid mutating the user's map
label_map = label_map.copy()
else:
label_map = {}
# Generates 0, 1, 2, 3, ... for node ids
counter = count()
for body in [
subtree.virtualRoot()
] + subtree.sortedBodiesList(): # subtree.physicalBodiesList():
if label_map.get(body.id(), None):
continue
if label_map.get(body.name(), None):
continue
# Use the next sequential number from the counter
if body.isRootBody():
label_map[body.id()] = str(next(counter))
continue
label_map[body.id()] = body.name()
return label_map
def subTreeToGraph(
subtree: kd.SubTree,
title="Multibody System",
label_map: dict[int | str, str] | None = None,
constraints: bool = True,
) -> NetworkGraph:
"""Convert a SubTree to a visjs NetworkGraph.
Parameters
----------
subtree: kd.SubTree
The subtree object to show
title: str
Title of the graph, defaults to "Multibody System"
label_map: dict[int | str, str] | None
Optional map to look up string labels for bodies. Keys may be
either body ids or body names.
constraints: bool
Whether to created edges for constraints; True by default
Returns
-------
NetworkGraph
The NetworkGraph for the SubTree, ready for visualization
"""
label_map = _fillLabelMap(subtree, label_map)
nodes = []
edges = []
penwidth = 2.0
def toNode(body: kd.BodyBase) -> Node:
"""Create a node based on a body."""
if body.isRootBody():
color = "coral"
elif body.isCompoundBody():
color = "palegreen"
elif subtree.isBaseBody(body):
color = "#FFFF00"
else:
color = "#87CEEB"
# border color to use when a node is highlighted. Keep the
# original background color
color_highlight_border = "magenta"
# First try looking up label by body ID
label = label_map.get(body.id(), None)
# If that fails try looking up label by body name
if not label:
label = label_map.get(body.name(), None)
# Create node for a body
if body.isCompoundBody():
fontsz = 16
ttip = f"{body.name()}/{body.bodiesTree().numBodies()}"
else:
fontsz = 12
ttip = (body.name(),)
node = Node(
id=body.id(),
label=str(label),
title=ttip,
options=NodeOptions(
shape="box",
size=25,
color=NodeColorOptions(
background=color,
border=color,
highlight={"border": color_highlight_border, "background": color},
),
font=NodeFontOptions(
color="#000000", # White text for dark mode
size=fontsz,
face="Arial",
),
),
)
return node
def toEdge(parent: kd.BodyBase, child: kd.BodyBase) -> Edge:
"""Create an edge between adjacent bodies."""
hinge_type = child.parentHinge().hingeType()
style = "solid"
if hinge_type == kd.HingeType.FULL6DOF:
style = "dashed"
color = edge_colors.get(hinge_type, "#FFFFFF")
# border color to use when a edge is highlighted. Keep the
# original background color
color_highlight_border = "magenta"
if child.isCompoundBody():
shg = child.parentHinge().subhinge(0)
ttip = f"{hinge_type}/{child.bodiesTree().nQ()}/{shg.nQ()}/{shg.nU()}"
else:
ttip = str(hinge_type)
# Create edge between adjacent bodies
edge = Edge(
from_=parent.id(),
to=child.id(),
title=ttip, # str(hinge_type),
options=EdgeOptions(
color=EdgeColorOptions(
color=color,
highlight=color_highlight_border,
),
width=penwidth,
dashes=(style == "dashed"),
arrows=ArrowOptions(
to=ArrowStyle(
enabled=True,
scaleFactor=1.2,
),
),
smooth=SmoothOptions(
type="cubicBezier",
roundness=0.5,
),
),
)
parent_hinge = child.parentHinge()
if parent_hinge and isinstance(parent_hinge, Base):
# If the connecting hinge is actually derived from Base and
# thus has an id, use the id for the edge.
edge.id = parent_hinge.id()
return edge
def addChildren(parent):
"""Recursively add children of a given body."""
for child in subtree.childrenBodies(parent):
nodes.append(toNode(child))
hinge_type = child.parentHinge().hingeType()
if hinge_type != kd.HingeType.FULL6DOF:
edges.append(toEdge(parent, child))
addChildren(child) # Recursively process children
# Start with virtual root
vroot = subtree.virtualRoot()
nodes.append(toNode(vroot)) # Add root node
addChildren(vroot) # Process all children recursively
if constraints and isinstance(subtree, kd.SubGraph):
edges.extend(multibodyConstraintEdges(subtree))
return NetworkGraph(nodes=nodes, edges=edges, title=title)
def multibodyConstraintEdges(
subgraph: kd.SubGraph,
) -> list[Edge]:
"""Create edges for all constraints in the provided SubGraph.
Parameters
----------
subgraph : kd.SubGraph
The SubGraph whose constraints will be used to create edges.
Returns
-------
list[Edge]
A list of edges, where every edge is associated with a constraint in the provided SubGraph.
"""
edges = []
penwidth = 2.0
# border color to use when a edge is highlighted. Keep the
# original background color
color_highlight_border = "magenta"
for constraint in subgraph.enabledConstraints():
if isinstance(constraint, kd.LoopConstraintCutJoint) or isinstance(
constraint, kd.LoopConstraintConVel
):
loop_constraint = constraint
srcnd = loop_constraint.sourceNode()
if srcnd:
source_node = srcnd.parentBody()
if source_node.id() != subgraph.virtualRoot().id():
# handle case where the source body may be embedded
if not subgraph.hasBody(source_node):
subgraph.displayModel()
assert subgraph.containsBody(source_node)
source_node = subgraph.getContainingCompoundBody(source_node)
else:
source_node = subgraph.virtualRoot()
tgtnd = loop_constraint.targetNode()
if tgtnd:
target_node = tgtnd.parentBody()
if target_node.id() != subgraph.virtualRoot().id():
# handle case where the target body may be embedded
if not subgraph.hasBody(target_node):
assert subgraph.containsBody(target_node)
target_node = subgraph.getContainingCompoundBody(target_node)
else:
target_node = subgraph.virtualRoot()
if isinstance(loop_constraint, kd.LoopConstraintCutJoint):
hinge_type = loop_constraint.hinge().hingeType()
edge = Edge(
from_=source_node.id(),
to=target_node.id(),
id=loop_constraint.id(),
title=f"{loop_constraint.name()}/{hinge_type}",
options=EdgeOptions(
# color=edge_colors[hinge_type],
color=EdgeColorOptions(
color=edge_colors[hinge_type],
highlight=color_highlight_border,
),
width=penwidth,
dashes=True,
arrows=ArrowOptions(
to=ArrowStyle(
enabled=True,
scaleFactor=1.0,
),
),
smooth=SmoothOptions(
type="cubicBezier",
roundness=0.3,
),
),
)
edges.append(edge)
else:
# non cut-joint constraints
edge = Edge(
from_=source_node.id(),
to=target_node.id(),
id=loop_constraint.id(),
title=f"{loop_constraint.name()}/CONVEL",
options=EdgeOptions(
color="#FF7F24", # chocolate1
width=penwidth,
dashes=True,
arrows=ArrowOptions(
to=ArrowStyle(
enabled=True,
scaleFactor=1.0,
),
),
smooth=SmoothOptions(
type="cubicBezier",
roundness=0.3,
),
),
)
edges.append(edge)
else:
# Add coordinate constraints
# for coord_constraint in subgraph.enabledCoordinateConstraints():
coord_constraint = cast(kd.CoordinateConstraint, constraint)
obody = (
cast(kd.PhysicalHinge, coord_constraint.osubhinge().parentHinge())
.pnode()
.parentBody()
)
pbody = (
cast(kd.PhysicalHinge, coord_constraint.psubhinge().parentHinge())
.pnode()
.parentBody()
)
scale = coord_constraint.getScaleRatio()
edge = Edge(
from_=obody.id(),
to=pbody.id(),
id=coord_constraint.id(),
title=f"{coord_constraint.name()}/COUPLER/{scale}",
options=EdgeOptions(
color="palegreen",
width=penwidth,
dashes=True,
arrows=ArrowOptions(
to=ArrowStyle(
enabled=True,
scaleFactor=1.0,
),
),
smooth=SmoothOptions(
type="cubicBezier",
roundness=0.3,
),
),
)
edges.append(edge)
return edges
class MultibodyGraphServer(GraphServer):
"""Specialized GraphServer for SubTree viewing.
Given a SubTree, this automatically generates a set of graphs
with and without constraints and other sets of extra edges. Buttons
are automatically added to toggle different parts of the graph on
and off.
"""
@dataclass
class _GraphData:
label: str
graph: NetworkGraph
enabled: bool
def __init__(
self,
subtree: kd.SubTree,
*,
title="Multibody System",
port=8765,
buttons: list[Button] | None = None,
label_map: dict[int | str, str] | None = None,
extra_edges: dict[str, list[Edge]] | list[Edge] | None = None,
):
"""Create a MultibodyGraphServer instance."""
if isinstance(extra_edges, list):
self.extra_edges = {"extra": extra_edges}
elif isinstance(extra_edges, dict):
self.extra_edges = extra_edges
else:
self.extra_edges = {}
self._subtree = CppWeakRef(subtree)
self._title = title
self._label_map = label_map
self._recompileGraph()
buttons = self._defaultButtons() + (buttons or [])
super().__init__(graph=self.subgraphs["tree"].graph, port=port, buttons=buttons)
[docs]
def hasSubGraph(self, label: str):
"""Check for the SubGraph with the given label.
Parameters
----------
label : str
The SubGraph to check for.
Returns
-------
bool
True if the SubGraph exists, False otherwise.
"""
return label in self.subgraphs
[docs]
def enableSubGraph(self, label: str):
"""Enable the SubGraph with the given label.
Parameters
----------
label : str
The SubGraph to enable.
"""
self.subgraphs[label].enabled = True
self._rebuildGraph()
[docs]
def disableSubGraph(self, label: str):
"""Disable the SubGraph with the given label.
Parameters
----------
label : str
The SubGraph to disable.
"""
self.subgraphs[label].enabled = False
self._rebuildGraph()
def _recompileGraph(self):
"""Recompile the graph from scratch.
This should be done whenever the subgraph changes.
"""
subtree = self._subtree()
if subtree is None:
raise ValueError("_recompileGraph called after associated SubTree was destroyed.")
tree_graph = subTreeToGraph(
subtree, title=self._title, constraints=False, label_map=self._label_map
)
self.subgraphs = {
"tree": MultibodyGraphServer._GraphData(
label="tree",
graph=tree_graph,
enabled=True,
)
}
if isinstance(subtree, kd.SubGraph):
self.extra_edges["constraints"] = multibodyConstraintEdges(subtree)
for label, edges in self.extra_edges.items():
if not edges:
# Skip empty edge lists
continue
graph = tree_graph.clone()
# Replace the edges in the tree graph
graph.edges = edges
# Remove nodes not touched by the new edges
graph.removeIsolatedNodes()
self.subgraphs[label] = MultibodyGraphServer._GraphData(
label=label,
graph=graph,
enabled=False, # Don't show initially
)
def _rebuildGraph(self):
"""Rebuild the graph."""
graph = NetworkGraph()
for subgraph_data in self.subgraphs.values():
if subgraph_data.enabled:
graph |= subgraph_data.graph
self.graph = graph
self.updateClientGraphs()
def _defaultButtons(self) -> list[Button]:
"""Define UI buttons always created by this class."""
if len(self.subgraphs) == 1:
return []
buttons = []
def _addToggle(label):
buttons.append(
Button(
text=f"{label} on",
callback=lambda: self.enableSubGraph(label),
style={}, # Add CSS styling here
)
)
buttons.append(
Button(
text=f"{label} off",
callback=lambda: self.disableSubGraph(label),
style={}, # Add CSS styling here
)
)
for label in self.subgraphs:
_addToggle(label)
return buttons
class LayoutMultibodyGraphServer(MultibodyGraphServer):
"""Retain layout through graph rebuilds and save stable body paths to JSON."""
def __init__(
self, subtree: kd.SubTree, *, topology_layout_path: str | Path | None = None, **kwargs
):
"""Create a server; return None.
Parameters
----------
subtree : kd.SubTree
Topology to display. Sibling body names must be unique.
topology_layout_path : str | Path | None
Existing layout to restore and destination for Save layout. None
makes Save layout download JSON. Missing files start a fresh layout.
**kwargs
Remaining MultibodyGraphServer constructor arguments.
"""
self._layout_lock = RLock()
self._unmatched_layout_ids: set[int] = set()
self._positions: dict[str, dict] = {}
self._body_keys: dict[int, str] = {}
self.layout_path = Path(topology_layout_path) if topology_layout_path is not None else None
if self.layout_path is not None and self.layout_path.exists():
document = json.loads(self.layout_path.read_text(encoding="utf-8"))
if document.get("format") != "kdflex-topology-layout" or document.get("version") != 1:
raise ValueError("Unsupported topology layout format/version")
# Reuse coordinate validation, including duplicate-key detection.
NetworkGraph().applyLayout(document)
self._positions = {record["id"]: record for record in document["nodes"]}
if any(not isinstance(key, str) for key in self._positions):
raise ValueError("Topology layout keys must be body path strings")
super().__init__(subtree=subtree, **kwargs)
def _recompileGraph(self) -> None:
"""Rebuild topology and its runtime-ID to stable-path map; return None."""
super()._recompileGraph()
subtree = self._subtree()
if subtree is None:
raise ValueError("Associated subtree has been destroyed")
first_compile = not self._body_keys
root = subtree.virtualRoot()
paths = {root.id(): []}
pending = [root]
while pending:
parent = pending.pop()
names = set()
for child in subtree.childrenBodies(parent):
name = child.name()
if not name or name in names:
kc.warn(
f"Graph layout persistence requires nonempty, unique sibling names. Found duplicate {name} and {name} for {subtree.name()} subtree"
)
names.add(name)
paths[child.id()] = paths[parent.id()] + [name]
pending.append(child)
# JSON arrays escape separators in names and reserve [] for the root.
self._body_keys = {
node_id: json.dumps(path, ensure_ascii=False) for node_id, path in paths.items()
}
if first_compile and self._positions:
self._storeBodyPositions(self._positions)
for subgraph in self.subgraphs.values():
self._applyPositions(subgraph.graph)
[docs]
def onConnect(self, client_id: int) -> None:
"""Send topology and a restore summary to client_id; return None.
Parameters
----------
client_id : int
Newly connected view receiving the graph and matching-body counts.
"""
with self._layout_lock:
super().onConnect(client_id)
if self._positions:
current = set(self._body_keys.values())
saved = set(self._positions)
notice = (
f"Bodies {len(current & saved)}; "
f"{len(current - saved)} new; "
f"{len(saved - current)} absent"
)
else:
notice = "" # Arrange nodes, then save the layout"
# Announce support even when there is no layout file yet.
self.sendMessage(
json.dumps(
{"type": "topology_layout_status", "payload": {"message": notice}}
).encode("utf-8"),
client_id,
)
def _applyPositions(self, graph) -> None:
"""Apply the subtree's current coordinates to graph; return None.
Parameters
----------
graph : NetworkGraph
Graph receiving positions for currently matching body paths.
"""
subtree = self._subtree()
if subtree is None:
raise ValueError("Associated subtree has been destroyed")
layout = subtree.getLayoutBodyData()
current_ids = {body.id() for body in subtree.sortedBodiesList()} | {
subtree.virtualRoot().id()
}
unmatched = set(layout) - current_ids
# Graph updates run repeatedly and for partial subgraphs. Warn once for
# each stale ID while checking membership against the complete subtree.
for body_id in sorted(unmatched - self._unmatched_layout_ids):
warn(
f"Topology layout for subtree '{subtree.name()}': body id {body_id} "
"does not match any body in this subtree; skipping entry."
)
self._unmatched_layout_ids = unmatched
records = []
for node in graph.nodes:
entry = layout.get(node.id)
# Unset coordinates are NaNs; a position needs both finite components.
if entry is not None and math.isfinite(entry.x) and math.isfinite(entry.y):
records.append({"id": node.id, "x": entry.x, "y": entry.y})
if records:
graph.applyLayout({"version": 1, "nodes": records})
def _storeBodyPositions(self, positions: dict[str, dict]) -> None:
"""Write stable-key positions into the subtree map; return None.
Parameters
----------
positions : dict[str, dict]
Body-path keys mapped to validated x/y records. Existing labels,
colors, unrelated entries, and uncaptured coordinates are preserved.
Each current body gets an entry, even when its coordinates are unset.
"""
subtree = self._subtree()
if subtree is None:
raise ValueError("Associated subtree has been destroyed")
layout = subtree.getLayoutBodyData()
for key in sorted(set(positions) - set(self._body_keys.values())):
warn(
f"Topology layout JSON '{self.layout_path}': body path {key!r} "
f"does not match any body in subtree '{subtree.name()}'; skipping entry."
)
for node_id, key in self._body_keys.items():
entry = layout.get(node_id)
if entry is None:
entry = kd.SubTreeLayoutBodyData()
record = positions.get(key)
if record is not None:
entry.x = record["x"]
entry.y = record["y"]
layout[node_id] = entry
# Python receives copies of the map and structs, so mutation alone is insufficient.
subtree.setLayoutBodyData(layout)
def _rebuildGraph(self) -> None:
"""Reapply layout before broadcasting a subgraph toggle; return None."""
with self._layout_lock:
# The base method calls updateClientGraphs after assembling the graph.
super()._rebuildGraph()
[docs]
def updateClientGraphs(self) -> None:
"""Restore retained positions before sending current topology; return None."""
with self._layout_lock:
self._applyPositions(self.graph)
super().updateClientGraphs()
[docs]
def onMessage(self, message: str, client_id: int) -> None:
"""Save captured positions or delegate existing GUI events; return None.
Parameters
----------
message : str
Websocket JSON. topology_layout_save captures current positions;
topology_layout_reset restores the subtree's current map without saving;
topology_layout_raw requests automatic layout without saved coordinates.
client_id : int
Sender receiving the save result; other clients are left unchanged.
"""
parsed = json.loads(message)
if parsed.get("type") == "topology_layout_raw":
with self._layout_lock:
# A private copy leaves saved data and other browsers untouched.
# Restore the same physics/layout defaults used by a new graph.
raw = self.graph.clone()
defaults = NetworkGraph().options
raw.options.physics = defaults.physics
raw.options.layout = defaults.layout
for node in raw.nodes:
if node.options is not None:
node.options.x = None
node.options.y = None
response = {
"type": "topology_layout_raw_result",
"payload": {"ok": True, "message": "Raw layout", "graph": raw.toDict()},
}
self.sendMessage(json.dumps(response).encode("utf-8"), client_id)
return
if parsed.get("type") == "topology_layout_reset":
response = {"type": "topology_layout_reset_result"}
try:
with self._layout_lock:
subtree = self._subtree()
if subtree is None:
raise ValueError("Associated subtree has been destroyed")
layout = subtree.getLayoutBodyData()
count = sum(
node.id in layout
and math.isfinite(layout[node.id].x)
and math.isfinite(layout[node.id].y)
for node in self.graph.nodes
)
if count:
self._applyPositions(self.graph)
# Reset only the requesting browser; other users keep their view.
self.sendMessage(self._buildUpdateGraphMsg(), client_id)
response["payload"] = {
"ok": True,
"message": f"Reset {count} bodies"
if count
else "No saved positions in subtree",
}
except (ValueError, TypeError, KeyError) as error:
response["payload"] = {"ok": False, "message": str(error)}
self.sendMessage(json.dumps(response).encode("utf-8"), client_id)
return
if parsed.get("type") != "topology_layout_save":
super().onMessage(message, client_id)
return
response = {"type": "topology_layout_result"}
try:
with self._layout_lock:
payload = parsed["payload"]
NetworkGraph().applyLayout(payload)
expected = {node.id for node in self.graph.nodes}
if {record["id"] for record in payload["nodes"]} != expected:
raise ValueError("Topology changed during capture; please save again")
positions = dict(self._positions)
for record in payload["nodes"]:
key = self._body_keys[record["id"]]
positions[key] = {**record, "id": key}
document = {
"format": "kdflex-topology-layout",
"version": 1,
"nodes": sorted(positions.values(), key=lambda record: record["id"]),
}
encoded = json.dumps(document, indent=2, allow_nan=False) + "\n"
if self.layout_path is not None:
temporary = None
try:
# Replace only after a complete write, preserving any prior layout on failure.
with NamedTemporaryFile(
mode="w",
encoding="utf-8",
delete=False, # dir=self.layout_path.parent
) as stream:
temporary = Path(stream.name)
stream.write(encoded)
if self.layout_path.exists():
import datetime
# Create a timestamp string (e.g., _20261017_111300)
timestamp = datetime.datetime.now().strftime("_%Y%m%d_%H%M%S")
# Insert timestamp before the file extension
archive_name = (
f"{self.layout_path.stem}{timestamp}{self.layout_path.suffix}"
)
archive_path = self.layout_path.with_name(archive_name)
os.replace(self.layout_path, archive_path)
print(f"Existing layout file backed up to: {archive_path}")
print(f"New layout file: {self.layout_path}")
try:
os.replace(temporary, self.layout_path)
except OSError:
print("Error: unable to create layout file")
finally:
if temporary is not None:
temporary.unlink(missing_ok=True)
response["payload"] = {"ok": True, "message": "Saved"} # {self.layout_path}"}
else:
response["payload"] = {
"ok": True,
"message": "Layout downloaded",
"document": document,
}
self._storeBodyPositions(
{self._body_keys[record["id"]]: record for record in payload["nodes"]}
)
self._positions = positions
self._applyPositions(self.graph)
except (ValueError, TypeError, KeyError, OSError) as error:
response["payload"] = {"ok": False, "message": str(error)}
self.sendMessage(json.dumps(response).encode("utf-8"), client_id)
