Source code for pathomx.kernel_helpers

import os
import sys
import numpy as np
import pandas as pd
import re
import io

from matplotlib.figure import Figure, AxesStack
from matplotlib.axes import Subplot

from mplstyler import StylesManager

import warnings
from . import displayobjects
from .utils import scriptdir, basedir
from IPython.core import display
from copy import deepcopy

        # Numpy
        np.array, np.ndarray,
        # Pandas
        pd.Series, pd.DataFrame,
        Figure, Subplot,
        # View types
        displayobjects.Svg, displayobjects.Html, displayobjects.Markdown,

[docs]class PathomxTool(object): ''' Simple wrapper class that holds the output data for a given tool; This is for user-friendliness not for use ''' def __str__(self): return self._name def __repr__(self): return self._name def __init__(self, name, *args, **kwargs): self.__dict__.update(kwargs) self._name = name
[docs]def pathomx_notebook_start(vars): #for k, v in varsi.items(): # vars[k] = v # _keep_input_vars = ['styles'] # vars['_pathomx_exclude_input_vars'] = [x for x in varsi.keys() if x not in _keep_input_vars] # Handle IO magic if '_io' in vars: for k, v in vars['_io']['input'].items(): if v in vars: vars[k] = deepcopy(vars[v]) else: vars[k] = None if '_rcParams' in vars: global rcParams from matplotlib import rcParams # Block warnings from deprecated rcParams here with warnings.catch_warnings(): warnings.simplefilter("ignore") for k, v in vars['_rcParams'].items(): rcParams[k] = v # Legacy shim if '_styles' in vars: vars['styles'] = vars['_styles']
[docs]def pathomx_notebook_stop(vars): varso = {} if '_io' in vars: # Handle IO magic for k, v in vars['_io']['output'].items(): if k in vars: vars[v] = vars[k] else: vars[v] = None for k, v in vars.items(): # Check it's an accepted type for passing; and not private (starts with _) if not k.startswith('_') and \ not k in vars['_io']['input'].keys(): if type(v) in MAGIC_TYPES or k in vars['_pathomx_expected_output_vars']: varso[k] = v elif hasattr(v, '_repr_html_'): try: # Check if it is a bound method (not a class definition) v._repr_html_() except: pass else: varso[k] = displayobjects.Html(v) vars['varso'] = varso
[docs]def progress(progress): ''' Output the current progress to stdout on the remote core this will be read from stdout and displayed in the UI ''' print("____pathomx_execute_progress_%.2f____" % progress)
[docs]class open_with_progress(io.IOBase): def __init__(self, f, *args, **kwargs): super(open_with_progress, self).__init__(f, *args, **kwargs) self._fsize = os.path.getsize(f) self._progress = None
[docs] def read(self, *args, **kwargs): super(open_with_progress, self).read(*args, **kwargs) self.check_and_emit_progress()
[docs] def check_and_emit_progress(self): # We only output at 2dp so only emit when that changes prg = round(self.tell() / self._fsize, 2) if prg != self._progress: self._progress = prg progress(prg)