Skip to content

Add annotation-aware visualizer #73

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
harsh-sikhwal merged 3 commits into from
Jul 15, 2026
Merged

Add annotation-aware visualizer #73

Show file tree
Hide file tree
Changes from 1 commit
Commits
Show all changes
3 commits
Select commit Hold shift + click to select a range
3366227
adding annotations in visualizer
harsh-sikhwal Jul 13, 2026
d853834
Merge branch 'main' into visualizer-annotation
harsh-sikhwal Jul 14, 2026
a68102a
changing visualizer to use service
harsh-sikhwal Jul 14, 2026
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
34 changes: 20 additions & 14 deletions src/infragraph/infragraph_service.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -10,6 +10,7 @@
import re
import json
import yaml
import warnings
import networkx
from networkx import Graph
from networkx.readwrite import json_graph
Expand Down Expand Up @@ -64,7 +65,6 @@ def __init__(self):
self._link_to_edges_map: Dict[str, List[Tuple[str, str]]] = {}
self._infrastructure: Infrastructure = Infrastructure()


@property
def infrastructure(self) -> Infrastructure:
"""Return the current backing store infrastructure"""
Expand All @@ -81,7 +81,6 @@ def get_networkx_graph(self) -> Graph:
raise ValueError("The networkx graph has not been created. Please call set_graph() first.")
return self._graph


def _expand_node_string(self, s: str) -> List[str]:
"""Expand a device/component string with slice notation into dot-notation paths.

Expand Down Expand Up @@ -670,11 +669,10 @@ def _build_partial_graph(self) -> Graph:
partial_graph.add_edge(ep1, ep2, **filtered)
return partial_graph


def get_graph(self, request: GraphRequest) -> str:
"""Returns the current networkx graph as a serialized json string."""
if self._graph is None:
raise ValueError("Graph is not set. Please call set_graph() first.")
raise ValueError("The networkx graph has not been created. Please call set_graph() first.")

if request.choice == request.INFRAGRAPH:
return self.populate_infragraph_dict(self._graph, request.infragraph.annotations.choice)
Expand Down Expand Up @@ -703,11 +701,17 @@ def edge_attrs_filter(attrs):
if k not in self._IMMUTABLE_ATTRIBUTES
)

def stringify(v):
# AnnotationAttribute.value is schema-typed as str; non-str attribute
# values (e.g. the immutable int "instance_idx") must be coerced so
# the resulting annotations can be deserialized back into an Annotation.
return v if isinstance(v, str) else str(v)

annotation_nodes = [
{
"name": node_name,
"attributes": [
{"attribute": k, "value": v}
{"attribute": k, "value": stringify(v)}
for k, v in node_attrs_filter(node_attrs)
]
}
Expand All @@ -724,18 +728,18 @@ def edge_attrs_filter(attrs):
annotation_edges.append({
"ep1": ep1,
"ep2": ep2,
"attributes": [{"attribute": k, "value": v} for k, v in filtered]
"attributes": [{"attribute": k, "value": stringify(v)} for k, v in filtered]
})
link_name = edge_attrs.get("link")
if link_name and link_name not in seen_links:
seen_links[link_name] = {
"name": link_name,
"attributes": [{"attribute": k, "value": v} for k, v in filtered]
"attributes": [{"attribute": k, "value": stringify(v)} for k, v in filtered]
}

graph_attrs = source_graph.graph
annotation_graph = [
{"attribute": k, "value": v}
{"attribute": k, "value": stringify(v)}
for k, v in graph_attrs.items()
if is_full or k not in self._IMMUTABLE_ATTRIBUTES
]
Expand Down Expand Up @@ -865,8 +869,8 @@ def annotate_graph(self, payload: Union[str, Annotation]):
if attribute_kvp.attribute not in self._IMMUTABLE_ATTRIBUTES:
networkx.set_node_attributes(self._graph, {n: {attribute_kvp.attribute: attribute_kvp.value} for n in matched})
else:
raise ValueError(f"cannot annotate pre-existing attribute {attribute_kvp.attribute} for {annotation_node.name}")

warnings.warn(f"Skipping immutable attribute {attribute_kvp.attribute} for {annotation_node.name}")
# edges
for annotation_node in annotate_request.edges:
# expand the nodes
Expand All @@ -884,23 +888,25 @@ def annotate_graph(self, payload: Union[str, Annotation]):
for attribute_kvp in annotation_node.attributes:
if attribute_kvp.attribute not in self._IMMUTABLE_ATTRIBUTES:
networkx.set_edge_attributes(self._graph, {(u, v): {attribute_kvp.attribute: attribute_kvp.value} for u, v in matched_edges})

else:
raise ValueError(f"Cannot annotate pre-existing attribute {attribute_kvp.attribute} for edge")
warnings.warn(f"Skipping immutable attribute {attribute_kvp.attribute} for edge")


# links
for annotation_link in annotate_request.links:
edges_for_link = self._link_to_edges_map.get(annotation_link.name, [])
for link_annotation in annotation_link.attributes:
if link_annotation.attribute in self._IMMUTABLE_ATTRIBUTES:
raise ValueError(f"Cannot annotate pre-existing attribute {link_annotation.attribute} for {annotation_link.name}")
warnings.warn(f"Skipping immutable attribute {link_annotation.attribute} for {annotation_link.name}")
continue
for ep1, ep2 in edges_for_link:
self._graph[ep1][ep2][link_annotation.attribute] = link_annotation.value

# graph
for attribute_kvp in annotate_request.graph:
if attribute_kvp.attribute in self._IMMUTABLE_ATTRIBUTES:
raise ValueError(f"Cannot annotate pre-existing attribute {attribute_kvp.attribute} for graph")
warnings.warn(f"Skipping immutable attribute {attribute_kvp.attribute} for graph")
continue
self._graph.graph[attribute_kvp.attribute] = attribute_kvp.value

def query_graph(self, payload: Union[str, QueryRequest]) -> QueryResponseContent:
Expand Down
Loading
Loading