ipython

Transforms all executions in IPython to execute with lineapy, by adding to input_transformers_post. You can find more documentations below: https://ipython.readthedocs.io/en/stable/config/inputtransforms.html

CellsExecutedState dataclass

Source code in lineapy/editors/ipython.py

@dataclass
class CellsExecutedState:
    # Keep the ipython for reset the session
    ipython: InteractiveShell
    tracer: Tracer
    # The code for this cell's execution
    code: str
    # If set, we should update this display on every cell execution.
    visualize_display_handle: Optional[DisplayHandle] = field(default=None)

    def create_visualize_display_object(self) -> DisplayObject:
        """
        Returns a jupyter display object for the visualization.
        """
        from lineapy.visualizer import Visualizer

        return Visualizer.for_public(self.tracer).ipython_display_object()

create_visualize_display_object()

Returns a jupyter display object for the visualization.

Source code in lineapy/editors/ipython.py

def create_visualize_display_object(self) -> DisplayObject:
    """
    Returns a jupyter display object for the visualization.
    """
    from lineapy.visualizer import Visualizer

    return Visualizer.for_public(self.tracer).ipython_display_object()

custom_get_exc_info(*args, **kwargs)

A custom get_exc_info which will transform exceptions raised from the users code to remove our frames that we have added.

Add an extra frame on top (in the AddFrame call), since ipython will strip out the first one (might change in future versions), which is probably also for similar reasons as us.

Source code in lineapy/editors/ipython.py

def custom_get_exc_info(*args, **kwargs):
    """
    A custom get_exc_info which will transform exceptions raised from the users
    code to remove our frames that we have added.

    Add an extra frame on top (in the `AddFrame` call), since ipython will
    strip out the first one (might change in future versions), which is
    probably also for similar reasons as us.
    """
    return transform_except_hook_args(
        original_get_exc_info(*args, **kwargs),
        AddFrame("", 0),
    )

input_transformer_post(lines, session_name=None)

Translate the lines of code for the cell provided by ipython.

Source code in lineapy/editors/ipython.py

def input_transformer_post(
    lines: List[str],
    session_name: Optional[str] = None,
) -> List[str]:
    """
    Translate the lines of code for the cell provided by ipython.
    """
    global STATE
    if not STATE:
        raise RuntimeError(
            "input_transformer_post shouldn't be called when we don't have an active tracer"
        )
    code = "".join(lines)

    # If we have just started, first start everything up
    if isinstance(STATE, StartedState):
        configure_logging()

        db = RelationalLineaDB.from_config(options)
        # pass in globals from ipython so that `get_ipython()` works
        # and things like `!cat df.csv` work in the notebook
        ipython_globals = STATE.ipython.user_global_ns
        tracer = Tracer(
            db, SessionType.JUPYTER, STATE.session_name, ipython_globals
        )

        # add statement so it is
        # included in artifact.get_code()
        if not "import lineapy\n" in lines:
            code = "import lineapy\n" + code

        STATE = CellsExecutedState(STATE.ipython, tracer, code=code)
    else:
        if STATE.tracer.db.url != options.database_url:
            STATE.ipython.reset()  # TODO: double check we are not over deleting variables
            configure_logging()
            db = RelationalLineaDB.from_config(options)
            tracer = Tracer(
                db,
                SessionType.JUPYTER,
                session_name,
                STATE.ipython.user_global_ns,
            )
            STATE = CellsExecutedState(STATE.ipython, tracer, code=code)
        STATE.code = code

    return RETURNED_LINES

start(session_name=None, db_url=None, ipython=None)

Initializing the runtime so that the cells are traced with lineapy.

Source code in lineapy/editors/ipython.py

def start(
    session_name: Optional[str] = None,
    db_url: Optional[str] = None,
    ipython: Optional[InteractiveShell] = None,
) -> None:
    """
    Initializing the runtime so that the cells are traced with lineapy.
    """
    global STATE
    ipython = ipython or get_ipython()  # type: ignore
    # IPython does not use exceptionhook, so instead we monkeypatch
    # how it processes the exceptions, in order to add our handler
    # that removes the outer frames.
    if REWRITE_EXCEPTIONS:
        InteractiveShell._get_exc_info = custom_get_exc_info

    ipython.input_transformers_post.append(input_transformer_post)
    STATE = StartedState(ipython, session_name=session_name, db_url=db_url)

stop()

Stop tracing if the stop() was called in the cell and should_stop was set. Also track the lib information (would be the most complete).

Source code in lineapy/editors/ipython.py

def stop() -> None:
    """
    Stop tracing if the `stop()` was called in the cell and should_stop was set.
    Also track the lib information (would be the most complete).
    """
    if isinstance(STATE, CellsExecutedState):
        send_lib_info_from_db(STATE.tracer.db, STATE.tracer.get_session_id())
        STATE.tracer.db.close()
    cleanup_cells()

visualize(*, live=False)

Display a visualization of the Linea graph from this session using Graphviz.

If live=True, then this visualization will live update after cell execution. Note that this comes with a substantial performance penalty, so it is False by default.

Note

If the visualization is not live, it will print out the visualization as of the previous cell execution, not the one where visualize is executed.

Source code in lineapy/editors/ipython.py

def visualize(*, live=False) -> None:
    """
    Display a visualization of the Linea graph from this session using Graphviz.

    If `live=True`, then this visualization will live update after cell execution.
    Note that this comes with a substantial performance penalty, so it is False
    by default.

    ??? note

        If the visualization is not live, it will print out the visualization
        as of the previous cell execution, not the one where `visualize` is executed.
    """
    if not isinstance(STATE, CellsExecutedState):
        raise RuntimeError(
            "Cannot visualize before we have started executing cells"
        )
    display_object = STATE.create_visualize_display_object()
    if live:
        # If we have an existing display handle, display a new version of it.
        if STATE.visualize_display_handle:
            STATE.visualize_display_handle.display(display_object)
        # Otherwise, create a new one
        else:
            STATE.visualize_display_handle = display(
                display_object, display_id=True
            )
    else:
        # Otherwise, just display the visualization
        display(display_object)

---

Was this helpful?

Help us improve docs with your feedback!

Send