janus package

Submodules

janus.base module

class janus.base.JanusBase(mode, max_history=50000)

Bases: object

Base class for Janus state tracking, providing history, undo/redo, and persistence.

JanusBase intercepts attribute assignments and container mutations to build a directed acyclic graph (DAG) of state transitions. This enables features like multiverse branching, state restoration, and timeline squashing.

create_moment_label(label)

Assign a human-readable label to the current state node.

Return type:

None

jump_to(label)

Restore the application state to a previously labeled moment.

Return type:

None

get_labeled_moments()

Retrieve a list of all labels assigned in the current history.

Return type:

list[str]

undo()

Revert the state to the previous node in the current timeline.

Return type:

None

redo()

Advance the state to the next node in the current timeline.

Return type:

None

apply_plugin_op(path, adapter_name, delta, forward)

Called by the engine to apply a plugin operation to a specific object.

Parameters:

• path (str) – The relative path to the object within this Janus instance.

• adapter_name (str) – The name of the adapter to use.

• delta (Any) – The delta blob to apply.

• forward (bool) – True if applying forward, False for backward (undo).

Return type:

None

tag_moment(**kwargs)

Attach arbitrary metadata tags to the current state node.

Return type:

None

get_all_tag_keys(label=None)

Get all metadata keys associated with a specific moment.

Return type:

tuple[str, ...]

get_all_tag_values(label=None)

Get all metadata values associated with a specific moment.

Return type:

tuple[Any, ...]

get_all_tags(label=None)

Get all metadata key-value pairs associated with a specific moment.

Return type:

dict[str, Any]

get_moment_tag(key, label=None)

Retrieve a specific metadata value by key from a moment.

Return type:

Any

label_node(label)

Assign a human-readable label to the current state node.

Return type:

None

squash(start_label=None, end_label=None)

Collapse state nodes into a single node.

Return type:

None

flatten(label=None)

Alias for squash().

Return type:

None

diff(start_label, end_label)

Compare the state between two moments (labels).

Return type:

dict[str, Any]

save(path)

Persist the entire multiverse/timeline history to a .jns file.

Return type:

None

load(path)

Restore history and state from a .jns file.

Return type:

None

plot(backend=None, **kwargs)

Visualize the multiverse DAG using a specialized backend.

Return type:

Any

visualize()

Compatibility shortcut for Mermaid-based visualization.

Return type:

Any

class janus.base.TimelineBase(max_history=50000)

Bases: JanusBase

A linear state tracking implementation.

class janus.base.MultiverseBase(max_history=50000)

Bases: JanusBase

A multiversal state tracking implementation supporting branching and merging.

property current_branch: str

The name of the currently active branch.

branch(label)

Create a new branch from the current state.

Return type:

None

create_branch(label)

Alias for branch() for API convenience.

Return type:

None

switch_branch(label)

Alias for jump_to() for API convenience.

Return type:

None

list_branches()

List all existing branch names.

Return type:

list[str]

list_nodes()

Return type:

list[str]

create_moment_label(label)

Alias for branch() to stay compatible with brainstorming terminology.

Return type:

None

merge(label, strategy='overshadow')

Merge changes from another branch into the current one.

Return type:

None

extract_timeline(label=None, filter_attr=None)

Extract history of operations from root to a specific node or label.

Return type:

list[dict[str, Any]]

find_moments(**criteria)

Search the entire multiverse for nodes matching criteria.

Return type:

list[str | int]

delete_branch(label)

Permanently delete a branch and its head reference.

Return type:

None

janus.containers module

class janus.containers.TrackedList(items, engine, name, owner=None)

Bases: list[Any]

A list subclass that automatically logs mutations to the Janus engine.

TrackedList intercepts methods like append, extend, insert, pop, clear, and __setitem__ to ensure the Janus engine can track and revert changes to the collection.

append(value)

Append object to the end of the list.

Return type:

None

extend(values)

Extend list by appending elements from the iterable.

Return type:

None

insert(index, value)

Insert object before index.

Return type:

None

pop(index=-1)

Remove and return item at index (default last).

Raises IndexError if list is empty or index is out of range.

Return type:

Any

clear()

Remove all items from list.

Return type:

None

remove(value)

Remove first occurrence of value.

Raises ValueError if the value is not present.

Return type:

None

class janus.containers.TrackedDict(items, engine, name, owner=None)

Bases: dict[Any, Any]

A dict subclass that automatically logs mutations to the Janus engine.

TrackedDict intercepts key assignments, deletions, and updates to ensure the Janus engine can track and revert changes to the collection.

update(other=(), /, **kwargs)

If E is present and has a .keys() method, then does: for k in E.keys(): D[k] = E[k] If E is present and lacks a .keys() method, then does: for k, v in E: D[k] = v In either case, this is followed by: for k in F: D[k] = F[k]

Return type:

None

pop(key, default=None)

If the key is not found, return the default if given; otherwise, raise a KeyError.

Return type:

Any

popitem()

Remove and return a (key, value) pair as a 2-tuple.

Pairs are returned in LIFO (last-in, first-out) order. Raises KeyError if the dict is empty.

Return type:

tuple[str, Any]

setdefault(key, default=None)

Insert key with a value of default if key is not in the dictionary.

Return the value for key if key is in the dictionary, else default.

Return type:

Any

clear()

Return type:

None

janus.options module

class janus.options.PlottingOptions(backend='mermaid', show_labels=True, color_scheme='default', engine_kwargs=<factory>)

Bases: object

Configuration for Multiverse visualization.

backend

The visualization backend to use (“mermaid” or “matplotlib”).

show_labels

Whether to show human-readable labels on nodes.

color_scheme

The color palette for the rendered graph.

engine_kwargs

Additional arguments passed to the underlying engine.

backend: str = 'mermaid'

show_labels: bool = True

color_scheme: str = 'default'

engine_kwargs: dict[str, Any]

class janus.options.Options(plotting=<factory>, max_history=50000, default_mode='multiversal')

Bases: object

Global configuration for Janus objects.

plotting

Options for graph visualization.

max_history

The maximum number of nodes to keep in the DAG.

default_mode

The default engine mode (“linear” or “multiversal”).

plotting: PlottingOptions

max_history: int = 50000

default_mode: str = 'multiversal'

janus.persistence module

janus.persistence.janus_encoder(obj)

Custom encoder for Janus-tracked objects to ensure msgpack compatibility.

Return type:

Any

janus.persistence.janus_decoder(obj)

Custom decoder for Janus-tracked objects to re-hydrate during load.

Return type:

Any

class janus.persistence.JanusPersistence

Bases: object

Handles serialization and persistence of Janus state histories.

static save(obj, path)

Persist the entire multiverse/timeline history to a .jns file.

Return type:

None

static load(obj, path)

Restore history and state from a .jns file.

Return type:

None

janus.registry module

class janus.registry.JanusAdapter(*args, **kwargs)

Bases: Protocol

Protocol for defining custom object tracking logic.

Adapters allow Janus to track third-party objects (like NumPy arrays or Pandas DataFrames) by defining how to calculate deltas and apply them forward or backward in time.

get_delta(old_state, new_state)

Calculate the difference (delta) between two states of the object.

Return type:

Any

apply_backward(target, delta_blob)

Apply a delta in reverse to restore a previous state.

Return type:

None

apply_forward(target, delta_blob)

Apply a delta forward to reach a newer state.

Return type:

None

get_snapshot(value)

Create a serializable snapshot of the object’s current state.

Return type:

Any

janus.registry.register_adapter(target_class)

Decorator to register a JanusAdapter for a specific class.

Return type:

Callable[[TypeVar(T, bound= type)], TypeVar(T, bound= type)]

janus.registry.register_wrapper(target_class)

Decorator to register a wrapping function for a specific class.

Return type:

Callable[[TypeVar(F, bound= Callable[..., Any])], TypeVar(F, bound= Callable[..., Any])]

janus.registry.wrap_value(value, engine, path, owner=None)

Recursively wrap value in Janus tracking proxies using the registry.

Return type:

Any

janus.tachyon_rs module

class janus.tachyon_rs.TachyonEngine(owner, mode, max_nodes=50000)

Bases: object

create_branch(label)

current_branch

current_node

delete_branch(label)

extract_timeline(label=None, filter_attr=None)

find_nodes_by_metadata(key, value)

get_diff(start_label, end_label)

get_graph_data()

get_graph_state()

get_metadata(key, node_id=None)

get_metadata_items(node_id=None)

get_metadata_keys(node_id=None)

get_metadata_values(node_id=None)

get_node_id(label)

label_node(node_label)

list_branches()

list_nodes()

log_dict_clear(path, keys, old_values)

log_dict_delete(path, key, old_value)

log_dict_pop(path, key, old_value)

log_dict_popitem(path, key, old_value)

log_dict_setdefault(path, key, value)

log_dict_update(path, keys, old_values, new_values)

log_list_clear(path, old_values)

log_list_extend(path, new_values)

log_list_insert(path, index, value)

log_list_pop(path, index, popped_value)

log_list_remove(path, value)

log_list_replace(path, index, old_val, new_val)

log_plugin_op(path, adapter_name, delta_blob)

log_update_attr(name, old_value, new_value)

merge_branch(source_label, strategy=None)

move_to(label)

move_to_creation()

move_to_node_id(node_id)

owner

redo()

set_graph_state(state)

set_metadata(key, value)

squash(start_label, end_label)

squash_branch(label=None)

sync_from_root()

undo()

class janus.tachyon_rs.TrackedListCore(engine, name)

Bases: object

log_clear(old_values)

log_extend(new_values)

log_insert(index, value)

log_pop(index, value)

log_replace(index, old_value, new_value)

class janus.tachyon_rs.TrackedDictCore(engine, name)

Bases: object

log_clear(keys, old_values)

log_delete(key, old_value)

log_pop(key, old_value)

log_update(keys, old_values, new_values)

janus.utils module

janus.utils.get_engine(obj)

Find the Janus engine by traversing the view/proxy hierarchy.

Parameters:

obj (Any) – The tracked object (or a view of one).

Return type:

Any | None

Returns:

The TachyonEngine if found, otherwise None.

janus.utils.resolve_path(owner, path)

Resolve a nested path like ‘data[0].key’ relative to an owner object.

Parameters:

• owner (Any) – The object to start resolution from (usually a JanusBase instance).

• path (str) – The dot-separated and bracketed path string.

Return type:

Any

Returns:

The object reached by the path.

janus.viz module

class janus.viz.VizBackend(*args, **kwargs)

Bases: Protocol

Protocol for Janus visualization backends.

plot(obj, **kwargs)

Render the multiverse DAG.

Return type:

Any

class janus.viz.MermaidBackend

Bases: object

Renders the multiverse DAG as a Mermaid diagram string.

plot(obj, **kwargs)

Generate a Mermaid diagram for the given Janus object.

Parameters:

• obj (JanusBase) – The JanusBase instance to visualize.

• **kwargs (Any) – Additional plotting options.

Return type:

str

Returns:

A string containing the Mermaid diagram definition.

janus.viz.get_backend(name)

Retrieve a visualization backend by name.

Return type:

VizBackend

janus.viz.register_backend(name, backend)

Register a new visualization backend.

Return type:

None

janus.viz_mpl module

class janus.viz_mpl.MatplotlibBackend

Bases: object

Renders the multiverse DAG using Matplotlib and NetworkX.

This backend provides a more traditional node-link diagram with customizable colors and layout.

plot(obj, **kwargs)

Renders the multiverse graph using Matplotlib.

Parameters:

• obj (JanusBase) – The Multiverse instance to visualize.

• show – Whether to call plt.show() immediately.

• figsize – Tuple of (width, height) for the figure.

• node_size – Size of the nodes in the plot.

• font_size – Size of the text labels.

• title – Custom title for the plot.

Return type:

Any

Returns:

A Matplotlib Figure object.