History is implemented directly by Network; there is no separate
NetworkHistory class. Mutating decorated methods save deep node/edge
snapshots as NetworkState objects. Undo, redo, checkout, and subsequent
mutations form a branching tree.
For structured topology explanations, provenance, pruning gaps, and JSON
export/recovery, see Interaction curation.
history_status() returns pruned_state_count and
working_state_matches_current. It compares the working tables with the current
snapshot, including nested provenance, without traversing older snapshots.
It costs work proportional to the current network, not the retained history.
state_count is the number of retained states.
list_states(*, offset=0, limit=None, detail="full") slices IDs before copying
metadata. The default result is unchanged. Summary entries retain id,
parent_ids, children_ids and current; their metadata contains method,
label, ancestry_gaps and transition. The summary transition includes
explanation_status and counts, with nodes/edges counts for added,
removed and changed. Evidence, parameters and inclusion explanations are
available through get_state. Summary entries omit the repeated
pruned_state_ids list; use history_status for the count and history_dict
for exact IDs. All returned mutable values are independent copies.
For legacy snapshots without a recorded table delta, that table's counts are
None, rather than invented zeroes.
Offsets and limits must be non-negative integers (booleans are rejected);
limit=None selects the remaining states, and zero returns an empty list.
Unknown detail modes and invalid pagination values raise ValueError.
Pages use creation order; they are not frozen cursors across concurrent edits.
MCP summaries can use the page's parent/child IDs directly, together with
state_count and history_status. Calling history_graph() or history_dict()
as well would reintroduce full-history work.
The `Network` object is the central organizing component of the `neko`
module. It is the subject of all operations implemented here, including
topological algorithms, graph analysis, network visualization and
integration of database knowledge.
Args:
initial_nodes: A list of initial nodes to be added to the network.
sif_file: A SIF (Simple Interaction Format) file to load the network from.
resources: A pandas DataFrame containing the resources database.
def__init__(self,initial_nodes:list[str]=None,sif_file=None,resources=None,*,database_name=None,):self._init_args=locals()delself._init_args['self']self.nodes=pd.DataFrame(columns=["Genesymbol","Uniprot","Type"])self.edges=pd.DataFrame(columns=["source","target","Type","Effect","References"])# Internal object-based storageself._node_objs=set()# Set of Node objectsself._edge_objs=set()# Set of Edge objectsself.initial_nodes=initial_nodesself._ontology=Ontology()# --- NetworkState history tracking ---self._states:dict[int,NetworkState]={}self._state_metadata:dict[int,dict]={}self._state_log:list[int]=[]self._state_counter:int=0self._current_state_id:Optional[int]=Noneself._root_state_id:Optional[int]=Noneself._auto_state_depth:int=0self._history_enabled:bool=Trueself._max_history:Optional[int]=Noneself._pruned_state_ids=[]self._transition_context=Noneself._resources_available=Trueself._is_initializing=Trueself._populate()self._is_initializing=False
defsave_state(self,metadata:Optional[dict]=None)->int:"""Persist the current network snapshot and return its state id."""parent_id=self._current_state_idstate=self._build_state(metadata,parent_id)state_id=state.state_idself._states[state_id]=stateself._state_metadata[state_id]=state.metadataself._state_log.append(state_id)ifparent_idisNone:self._root_state_id=state_idelse:parent_state=self._states[parent_id]parent_state.add_child(state_id)self._current_state_id=state_idself._state_counter+=1self._enforce_max_history()returnstate_id
defset_max_history(self,max_states:Optional[int])->None:"""Set the maximum number of stored history states (None disables pruning)."""ifmax_statesisNone:self._max_history=Noneelse:max_states=int(max_states)ifmax_states<2:max_states=2self._max_history=max_statesself._enforce_max_history()
@contextmanagerdefsuspend_history(self):"""Temporarily suspend automatic state capture within the context."""previous=self._history_enabledself._history_enabled=Falsetry:yieldfinally:self._history_enabled=previous
deflist_states(self,*,offset=0,limit=None,detail="full")->list:"""Return a creation-ordered page; defaults preserve the full listing. Summary metadata includes labels, explanation status and change counts, without copying evidence, parameters or inclusion explanations. """iftype(offset)isnotintoroffset<0:raiseValueError("offset must be a non-negative integer.")iflimitisnotNoneand(type(limit)isnotintorlimit<0):raiseValueError("limit must be None or a non-negative integer.")ifdetailnotin("full","summary"):raiseValueError("detail must be 'full' or 'summary'.")stop=NoneiflimitisNoneelseoffset+limitresult=[]forstate_idinself._state_log[offset:stop]:metadata=self._state_metadata.get(state_id,{})ifdetail=="summary":transition=metadata.get("transition",{})metadata={"method":metadata.get("method"),"label":metadata.get("label"),"ancestry_gaps":list(metadata.get("ancestry_gaps",[])),"transition":{"explanation_status":transition.get("explanation_status","unavailable"),"counts":{table:{kind:len(transition.get(table,{}).get(kind,[]))forkindin("added","removed","changed")}iftableintransitionelseNonefortablein("nodes","edges")},},}entry={"id":state_id,"metadata":copy.deepcopy(metadata),"parent_ids":list(self._states[state_id].parent_ids),"children_ids":list(self._states[state_id].children_ids),"current":state_id==self.current_state_id,}ifdetail=="full":entry["pruned_state_ids"]=list(self._pruned_state_ids)result.append(entry)returnresult
defcheckout(self,state_id:int)->None:"""Restore the network to a previously saved state."""state=self._get_state(state_id)self.nodes=_history.copy_frame(state.nodes)self.edges=_history.copy_frame(state.edges)self.initial_nodes=copy.deepcopy(state.metadata.get("initial_nodes",self.initial_nodes))self._database_name=state.metadata.get("database_name",self.database_name)self._current_state_id=state_idself.sync_nodes_from_df()self.sync_edges_from_df()
defrestore_state(self,state_id:Optional[int]=None)->None:"""Backward compatible wrapper around :meth:`checkout`."""target=state_idifstate_idisnotNoneelseself._current_state_idiftargetisNone:raiseValueError("No saved states to restore.")self.checkout(self._resolve_state_id(target))
defundo(self)->None:"""Move to the parent state if available."""ifself._current_state_idisNone:returnparents=self._get_state(self._current_state_id).parent_idsifnotparents:returnself.checkout(parents[-1])
defredo(self,state_id:Optional[int]=None)->None:"""Move to a child state. If more than one child exists, a specific state id is required."""ifself._current_state_idisNone:returnchildren=self._get_state(self._current_state_id).children_idsifnotchildren:returntarget=state_idiftargetisNone:iflen(children)!=1:raiseValueError("Multiple branches available; specify a target state id.")target=children[0]eliftargetnotinchildren:raiseValueError(f"State {target} is not a child of {self._current_state_id}.")self.checkout(target)
defcompare_states(self,state_a:int,state_b:int)->dict:"""Compare two states identified by their ids."""resolved_a=self._resolve_state_id(state_a)resolved_b=self._resolve_state_id(state_b)state1=self._get_state(resolved_a)state2=self._get_state(resolved_b)nodes1=self._state_node_labels(state1)nodes2=self._state_node_labels(state2)edges1=self._state_edge_signatures(state1)edges2=self._state_edge_signatures(state2)diff={"added_nodes":list(nodes2-nodes1),"removed_nodes":list(nodes1-nodes2),"added_edges":list(edges2-edges1),"removed_edges":list(edges1-edges2),"node_changes":_history.delta(state1.nodes,state2.nodes,["Uniprot","Genesymbol"]),"edge_changes":_history.delta(state1.edges,state2.edges,["source","target"]),}returndiff
defdescribe_history(self)->None:"""Pretty-print the branching history tree."""ifnotself._states:print("<no states recorded>")returndef_label(state:NetworkState)->str:meta=state.metadataor{}label=meta.get("label")ormeta.get("description")iflabel:returnstr(label)meta={key:valueforkey,valueinmeta.items()ifkeynotin_history.INTERNAL_METADATA_KEYS}ifmeta:returnstr(meta)return""def_walk(state_id:int,depth:int)->None:state=self._get_state(state_id)indent=" "*depthlabel=_label(state)suffix=f" - {label}"iflabelelse""print(f"{indent}State {state_id}{suffix}")forchild_idinstate.children_ids:_walk(child_id,depth+1)_walk(self._root_state_id,0)
defhistory_graph(self):"""Return a networkx DiGraph representing the state transitions."""from.._visual.historyimportbuild_history_graphreturnbuild_history_graph(self)
defhistory_digraph(self,include_metadata:bool=True):"""Return a Graphviz digraph for the history."""from.._visual.historyimporthistory_digraphreturnhistory_digraph(self,include_metadata=include_metadata)
defhistory_html(self,include_metadata:bool=True,div_class:str="neko-history-graph")->str:"""Return an HTML snippet embedding the history graph."""from.._visual.historyimporthistory_htmlreturnhistory_html(self,include_metadata=include_metadata,div_class=div_class)
defexport_history(self,path,*,pretty=False):"""Write all retained states, working state, provenance and ancestry."""return_history.write_json(path,self.history_dict(),pretty=pretty,_normalized=True,)
@classmethoddeffrom_history(cls,path,resources=None):"""Restore stored history offline; resources are needed for growth."""return_history.load_history(cls,path,resources)
defhistory_status(self)->dict:"""Inspect pruning and working changes without serializing past states."""return{"pruned_state_count":len(self._pruned_state_ids),"working_state_matches_current":_history.working_state_matches_current(self),}
defget_state(self,state_id:Optional[int]=None)->dict:"""Read one exact saved state without checkout or mutable internal references."""state_id=self.current_state_idifstate_idisNoneelsestate_idiftype(state_id)isnotint:raiseValueError("state_id must be an integer.")state=self._get_state(state_id)return{"id":state_id,"current":state_id==self.current_state_id,"parent_ids":list(state.parent_ids),"children_ids":list(state.children_ids),"nodes":_history.frame_payload(state.nodes),"edges":_history.frame_payload(state.edges),"metadata":_history.json_value(state.metadata),}
defbuild_history_graph(network:"Network")->nx.DiGraph:"""Return a directed graph of the network state history."""graph=nx.DiGraph()graph.graph.update(database_name=network.database_name,current_state_id=network.current_state_id,pruned_state_ids=list(network._pruned_state_ids))forstate_idinnetwork.list_states():# list_states returns dictionaries with id/metadatagraph.add_node(state_id["id"],**state_id.get("metadata",{}))forstate_id,stateinnetwork._states.items():# pylint: disable=protected-accessforchildinstate.children_ids:graph.add_edge(state_id,child)returngraph
defhistory_digraph(network:"Network",include_metadata:bool=True)->Digraph:"""Render history as a Graphviz Digraph."""digraph=Digraph()forstate_id,stateinnetwork._states.items():# pylint: disable=protected-accessmetadata=state.metadataor{}label_lines=[f"State {state_id}"]ifmetadata.get("label"):label_lines.append(str(metadata["label"]))ifstate_id==network.current_state_id:label_lines.append("current")ifinclude_metadataandmetadata.get("method"):args=metadata.get("args",[])kwargs=metadata.get("kwargs",{})params=", ".join(args)ifkwargs:kwargs_repr=", ".join(f"{k}={v}"fork,vinkwargs.items())params=", ".join(filter(None,[params,kwargs_repr]))label_lines.append(f"{metadata['method']}({params[:160]})")elifinclude_metadata:extras={key:valueforkey,valueinmetadata.items()ifkeynotin{"label","method","args","kwargs"}and(key=="database_name"orkeynotinINTERNAL_METADATA_KEYS)}forkey,valueinextras.items():label_lines.append(f"{key}: {value}")ifmetadata.get('ancestry_gaps'):label_lines.append('ancestry gap (pruned states)')digraph.node(str(state_id),"\n".join(label_lines),URL=f"#neko-state-{state_id}"ifinclude_metadataelse"",tooltip=f"State {state_id}: structured explanation")forchildinstate.children_ids:digraph.edge(str(state_id),str(child))returndigraph
defhistory_html(network:"Network",include_metadata:bool=True,div_class:str="neko-history-graph")->str:"""Return an HTML snippet embedding the history graph as SVG."""svg=history_digraph(network,include_metadata=include_metadata).pipe(format="svg").decode("utf-8")details=''ifinclude_metadata:details=''.join(f'<details id="neko-state-{sid}"><summary>State {sid} explanation</summary>'f'<pre>{html.escape(json.dumps(json_value(state.metadata),indent=2))}</pre></details>'forsid,stateinnetwork._states.items())returnf'<div class="{html.escape(div_class,quote=True)}">{svg}{details}</div>'