9644630967

Committed 24 Jun 2024 11:35AM UTC coverage: 88.695% (-0.7%) from 89.382%

Build # 9644630967

Build Type

push

github

Committed by

GuyTeichman

Commit Message

* When clicking on a legend node in an automatic analysis report, all nodes of that type will be highlighted.

Run Details

15914 of 18653 branches covered (85.32%)

Branch coverage included in aggregate %.

5 of 5 new or added lines in 1 file covered. (100.0%)

646 existing lines in 15 files now uncovered.

13829 of 14881 relevant lines covered (92.93%)

8.36 hits per line

Source File
Press 'n' to go to next uncovered line, 'b' for previous

95.06

/rnalysis/utils/generic.py

import abc
import inspect
import itertools
import math
import types
import typing
import warnings
from datetime import datetime
from functools import lru_cache
from pathlib import Path
from typing import Union, Callable, Tuple

import joblib
import matplotlib.collections
import matplotlib.pyplot as plt
import numpy as np
import polars as pl
import polars.selectors as cs
import yaml
from matplotlib import figure
from matplotlib.widgets import MultiCursor
from scipy.special import comb
from sklearn.preprocessing import PowerTransformer, StandardScaler
from tqdm.auto import tqdm

from rnalysis import __version__

try:
    import numba
except ImportError:  # pragma: no cover
    warnings.warn("RNAlysis can perform faster when package 'numba' is installed. \n"
                  "If you want to improve the performance of slow operations on RNAlysis, "
                  "please install package 'numba'. ")


    class numba:  # pragma: no cover
        @staticmethod
        def jit(*args, **kwargs):
            return lambda f: f

        @staticmethod
        def njit(*args, **kwargs):
            return lambda f: f


def readable_name(name: str):
    def decorator(item):
        item.readable_name = name
        return item

    return decorator


class TemporaryMatplotlibBackend:
    def __init__(self, backend):
        self.backend = backend
        self.original_backend = matplotlib.get_backend()

    def __enter__(self):
        matplotlib.use(self.backend, force=True)

    def __exit__(self, exc_type, exc_val, exc_tb):
        matplotlib.use(self.original_backend, force=True)


class ProgressParallel(joblib.Parallel):
    # tqdm progress bar for parallel tasks based on:
    # https://stackoverflow.com/questions/37804279/how-can-we-use-tqdm-in-a-parallel-execution-with-joblib/50925708
    # answer by 'user394430'
    def __init__(self, use_tqdm=True, total=None, desc: str = '', unit: str = 'it', backend='loky', *args, **kwargs):
        self._use_tqdm = use_tqdm
        self._total = total
        self._desc = desc
        self._unit = unit
        kwargs['n_jobs'] = -2
        super().__init__(*args, backend=backend, **kwargs, verbose=100)

    def __call__(self, *args, **kwargs):
        fmt = '{desc}: {percentage:3.0f}%|{bar}| [{elapsed}<{remaining}, {rate_fmt}{postfix}]'
        with tqdm(disable=not self._use_tqdm, total=self._total, desc=self._desc, unit=self._unit,
                  bar_format=fmt) as self._pbar:
            return joblib.Parallel.__call__(self, *args, **kwargs)

    def print_progress(self):
        if self._total is None:
            self._pbar.total = self.n_dispatched_tasks
        self._pbar.n = self.n_completed_tasks
        self._pbar.refresh()


def standard_box_cox(data: Union[np.ndarray, pl.DataFrame]) -> Union[np.ndarray, pl.DataFrame]:
    """

    :param data:
    :type data:
    :return:
    :rtype:
    """
    if isinstance(data, pl.DataFrame):
        array = data.select(cs.numeric()).to_numpy()
    else:
        array = data
    res_array = StandardScaler().fit_transform(PowerTransformer(method='box-cox').fit_transform(array + 1))
    if isinstance(data, pl.DataFrame):
        return data.select(~cs.numeric()).with_columns(
            pl.DataFrame(res_array, schema=data.select(cs.numeric()).columns))
    return res_array


def shift_to_baseline(data: Union[np.ndarray, pl.DataFrame], baseline: float = 0) -> Union[np.ndarray, pl.DataFrame]:
    """

    :param data:
    :type data:
    :param baseline:
    :type baseline:
    :return:
    :rtype:
    """
    if isinstance(data, pl.DataFrame):
        array = data.select(cs.numeric()).to_numpy()
    elif isinstance(data,pl.Series):
        array = data.to_numpy()
    else:
        array = data
    min_val = array.min()

    while isinstance(min_val, np.ndarray):
        min_val = min_val.min()
    diff = baseline - min_val
    return data + diff


def standardize(data: Union[np.ndarray, pl.DataFrame]) -> Union[np.ndarray, pl.DataFrame]:
    """

    :param data:
    :type data:
    :return:
    :rtype:
    """
    if isinstance(data, pl.DataFrame):
        array = data.select(cs.numeric()).to_numpy()
    else:
        array = data
    res_array = StandardScaler().fit_transform(array)
    if isinstance(data, pl.DataFrame):
        return data.select(~cs.numeric()).with_columns(pl.DataFrame(res_array, schema=data.columns))
    return res_array


def intersection_nonempty(*objs: Union[list, set, tuple]):
    return set.intersection(*[set(item) for item in objs if len(item) > 0])


def color_generator():
    """
    A generator that yields distinct colors up to a certain limit, and then yields randomized RGB values.

    :return: a color name string (like 'black', \
    or a numpy.ndarray of size (3,) containing three random values each between 0 and 1.

    """
    preset_colors = ['tab:blue', 'tab:red', 'tab:green', 'tab:orange', 'tab:purple', 'tab:brown', 'tab:pink',
                     'tab:gray', 'tab:olive', 'tab:cyan', 'gold', 'maroon', 'mediumslateblue', 'fuchsia',
                     'lawngreen', 'moccasin', 'thistle']
    for color in preset_colors:
        yield color
    while True:
        yield np.random.random(3)


@lru_cache(maxsize=2 ** 16)
def combination(a: int, b: int) -> int:
    return int(comb(a, b))


class SetWithMajorityVote(set):
    def majority_vote_intersection(self, *others, majority_threshold: float = 0.5):
        counts = {}
        n_sets = len(others) + 1
        for set_obj in itertools.chain([self], others):
            for obj in set_obj:
                counts[obj] = counts.get(obj, 0) + (1 / n_sets)
        result = {key for key, val in counts.items() if val >= majority_threshold}
        return result


def get_method_readable_name(method: Union[str, Callable], obj: object = None):
    try:
        if isinstance(method, str):
            func = getattr(obj, method)
        else:
            func = method
        if hasattr(func, 'readable_name'):
            name = func.readable_name
        else:
            name = func.__name__
        return name
    except AttributeError:
        return 'NAME NOT FOUND'


def get_method_signature(method: Union[str, Callable], obj: object = None):
    try:
        if isinstance(method, str):
            func = getattr(obj, method)
        else:
            func = method
        signature = inspect.signature(func)
        return signature.parameters
    except AttributeError:
        return {}


def despine(ax):
    for side in ['top', 'right']:
        ax.spines[side].set_visible(False)


def mix_colors(*colors: Tuple[float, float, float]):
    n_colors = len(colors)
    if n_colors == 1:
        return colors[0]

    colors = np.array(colors)
    multiplier = (1 / n_colors) + (1.6 / (2 ** n_colors)) / n_colors

    mix_color = multiplier * np.sum(colors, axis=0)
    mix_color = np.min([mix_color, [1.0, 1.0, 1.0]], 0)
    return mix_color


def sum_intervals_inclusive(intervals: typing.List[typing.Tuple[int, int]]) -> int:
    sorted_intervals = sorted(intervals)
    total = 0
    prev_interval = (-np.inf, -np.inf)
    for i in range(len(sorted_intervals)):
        this_interval = sorted_intervals[i]
        if this_interval[0] <= prev_interval[1]:
            this_interval = (prev_interval[1] + 1, this_interval[1])
        total += this_interval[1] - this_interval[0] + 1
        prev_interval = this_interval

    return total


def bic_score(X: np.ndarray, labels: np.ndarray):
    """
      BIC score for the goodness of fit of clusters.
      This Python function is translated from the Golang implementation by the author of the paper.
      The original code is available here:
      https://github.com/bobhancock/goxmeans/blob/a78e909e374c6f97ddd04a239658c7c5b7365e5c/km.go#L778
      """

    n_points = len(labels)
    n_clusters = len(set(labels))
    n_dimensions = X.shape[1]

    n_parameters = (n_clusters - 1) + (n_dimensions * n_clusters) + 1

    loglikelihood = 0
    for label_name in set(labels):
        X_cluster = X[labels == label_name]
        n_points_cluster = len(X_cluster)
        centroid = np.mean(X_cluster, axis=0)
        variance = np.sum((X_cluster - centroid) ** 2) / (len(X_cluster) - 1)
        loglikelihood += \
            n_points_cluster * (np.log(n_points_cluster) - np.log(n_points) - (n_dimensions / 2) * np.log(
                2 * math.pi * variance) - (n_points_cluster - 1) / 2)

    bic = loglikelihood - (n_parameters / 2) * np.log(n_points)

    return bic


def format_time(seconds: float):
    seconds = int(seconds)
    n_minutes = seconds // 60
    n_seconds = seconds % 60
    return f'{n_minutes:02d}:{n_seconds:02d}'


def sanitize_variable_name(name: str) -> str:
    """
    Sanitize a string to turn it into a legal variable name in R/Python.
    :param name: name to sanitize
    :type name: str
    :return: sanitizeed name
    :rtype: str
    """
    new_name = name.rstrip().replace(' ', '_')
    if new_name[0].isdigit():
        new_name = 'var_' + new_name

    if not new_name.isalnum():
        new_name = ''.join([char if char.isalnum() else '_' for char in new_name])

    return new_name


class InteractiveScatterFigure(figure.Figure):
    def __init__(self, labels: typing.List[str], annotation_fontsize: float = 10, show_cursor: bool = True, *args,
                 **kwargs):
        super().__init__(*args, **kwargs)
        self.ax = self.add_subplot()
        self.labels = labels
        self.annotation_fontsize = annotation_fontsize
        self.is_labeled: typing.Dict[int, plt.Line2D] = {}
        self.canvas.mpl_connect('pick_event', self.on_pick)
        if show_cursor:
            self.cursor = MultiCursor(self.canvas, self.axes, color='k', lw=0.5, horizOn=True, vertOn=True)
            self.canvas.mpl_connect('axes_leave_event', self.on_exit)
            self.canvas.mpl_connect('motion_notify_event', self.on_move)

    def on_exit(self, event):
        self.cursor.clear(event)
        self.canvas.draw()

    def on_move(self, event):
        self.canvas.draw()

    def on_pick(self, event):
        for this_ind in event.ind:
            if isinstance(event.artist, matplotlib.collections.PathCollection):
                data = event.artist.get_offsets().data
                xdata, ydata = data[:, 0], data[:, 1]
            elif isinstance(event.artist, plt.Line2D):
                thisline = event.artist
                xdata = thisline.get_xdata()
                ydata = thisline.get_ydata()
            else:
                return

            if this_ind in self.is_labeled:
                ann = self.is_labeled.pop(this_ind)
                ann.remove()
            else:
                self.is_labeled[this_ind] = plt.annotate(self.labels[this_ind],
                                                         (np.take(xdata, this_ind), np.take(ydata, this_ind)),
                                                         xytext=(3, 3), textcoords='offset points',
                                                         fontsize=self.annotation_fontsize)

            self.canvas.draw()


class GenericPipeline(abc.ABC):
    __slots__ = {'functions': 'list of functions to perform', 'params': 'list of function parameters'}

    def __init__(self):
        self.functions = []
        self.params = []

    def __len__(self):
        return len(self.functions)

    def __eq__(self, other):
        if self.functions == other.functions and self.params == other.params:
            return True
        return False

    @staticmethod
    def _param_string(args: tuple, kwargs: dict):

        """
        Returns a formatted string of the given arguments and keyworded arguments.

        :param args: arguments to format as string
        :type args: tuple
        :param kwargs: keyworded arguments to format as string
        :type kwargs: dict
        :return: a formatted string of arguments and keyworded argumentss
        :rtype: str

        """
        args_str = ', '.join([f"'{arg}'" if isinstance(arg, str) else f"{arg}" for arg in args])
        kwargs_str = ', '.join(
            [f"{key}='{arg}'" if isinstance(arg, str) else f"{key}={arg}" for key, arg in kwargs.items()])
        if len(args_str) == 0:
            return kwargs_str
        elif len(kwargs_str) == 0:
            return args_str
        else:
            return f"{args_str}, {kwargs_str}"

    def remove_last_function(self):

        """
        Removes from the Pipeline the last function that was added to it. Removal is in-place.

        :Examples:
            >>> from rnalysis import filtering
            >>> pipe = filtering.Pipeline()
            >>> pipe.add_function(filtering.Filter.filter_missing_values)
            Added function 'Filter.filter_missing_values()' to the pipeline.
            >>> pipe.remove_last_function()
            Removed function filter_missing_values with parameters [] from the pipeline.

        """
        assert len(self.functions) > 0 and len(self.params) > 0, "Pipeline is empty, no functions to remove!"
        func = self.functions.pop(-1)
        args, kwargs = self.params.pop(-1)
        print(
            f"Removed function {func.__name__} with parameters [{self._param_string(args, kwargs)}] from the pipeline.")

    def _readable_func_signature(self, func: types.FunctionType, args: tuple, kwargs: dict):
        """
        Returns a human-readable string functions signature for the given function and arguments.

        :param func: the function or method to generate signature for
        :type func: function
        :param args: arguments given for the function
        :type args: tuple
        :param kwargs: keyworded arguments given for the function
        :type kwargs: dict
        :return: function signature string
        :rtype: str
        """
        return f"{get_method_readable_name(func)}: ({self._param_string(args, kwargs)})"

    def export_pipeline(self, filename: Union[str, Path, None]) -> Union[None, str]:
        """
        Export a Pipeline to a Pipeline YAML file or YAML-like string.

        :param filename: filename to save the Pipeline YAML to, or None to return a YAML-like string instead.
        :type filename: str, pathlib.Path, or None
        :return: if filename is None, returns the Pipeline YAML-like string.
        """
        pipeline_dict = self._get_pipeline_dict()
        for func, params in zip(self.functions, self.params):
            pipeline_dict['functions'].append(func.__name__)
            pipeline_dict['params'].append(params)
        if filename is None:
            return yaml.safe_dump(pipeline_dict)
        else:
            with open(filename, 'w') as f:
                yaml.safe_dump(pipeline_dict, f)

    def _get_pipeline_dict(self):
        d = dict(functions=[], params=[], metadata={'rnalysis_version': f'{__version__}',
                                                    'export_time': datetime.today().strftime('%Y/%m/%d, %H:%M:%S')})
        return d

    @classmethod
    def import_pipeline(cls, filename: Union[str, Path]) -> 'GenericPipeline':
        """
        Import a Pipeline from a Pipeline YAML file or YAML-like string.

        :param filename: name of the YAML file containing the Pipeline, or a YAML-like string.
        :type filename: str or pathlib.Path
        :return: the imported Pipeline
        :rtype: Pipeline
        """
        try:
            with open(filename) as f:
                pipeline_dict = yaml.safe_load(f)
        except OSError:
            pipeline_dict = yaml.safe_load(filename)
        pipeline = cls.__new__(cls)
        pipeline._init_from_dict(pipeline_dict)
        return pipeline

    def _init_from_dict(self, pipeline_dict: dict):
        raise NotImplementedError

    def _func_signature(self, func: types.FunctionType, args: tuple, kwargs: dict):
        """
        Returns a string functions signature for the given function and arguments.

        :param func: the function or method to generate signature for
        :type func: function
        :param args: arguments given for the function
        :type args: tuple
        :param kwargs: keyworded arguments given for the function
        :type kwargs: dict
        :return: function signature string
        :rtype: str
        """
        return f"{func.__name__}({self._param_string(args, kwargs)})"

    def add_function(self, func: types.FunctionType, *args, **kwargs):
        assert isinstance(func, types.FunctionType), f"'func' must be a function, is {type(func)} instead."

        self.functions.append(func)
        self.params.append((args, kwargs))
        print(f"Added function '{self._func_signature(func, args, kwargs)}' to the pipeline.")

    def _validate_pipeline(self):
        assert len(self.functions) > 0 and len(self.params) > 0, "Cannot apply an empty pipeline!"
        assert len(self.functions) == len(self.params), f"Cannot apply Pipeline: " \
                                                        f"length of 'functions' different from length of 'params'!"

    @abc.abstractmethod
    def apply_to(self, *args, **kwargs):
        raise NotImplementedError


def jitter(n: int, jitter_range: float = 0.2) -> np.ndarray:
    if n == 0:
        return np.array([])
    if jitter_range == 0:
        return np.zeros(n)

    jittered = np.random.uniform(-jitter_range, jitter_range, n)
    jittered -= np.mean(jittered)  # center the jittered points
    scaling_factor = jitter_range / max(np.max(np.abs(jittered)), 1e-6)
    jittered *= scaling_factor  # scale the jitter points to the jitter range
    return jittered

1	import abc	9✔
2	import inspect	9✔
3	import itertools	9✔
4	import math	9✔
5	import types	9✔
6	import typing	9✔
7	import warnings	9✔
8	from datetime import datetime	9✔
9	from functools import lru_cache	9✔
10	from pathlib import Path	9✔
11	from typing import Union, Callable, Tuple	9✔
12
13	import joblib	9✔
14	import matplotlib.collections	9✔
15	import matplotlib.pyplot as plt	9✔
16	import numpy as np	9✔
17	import polars as pl	9✔
18	import polars.selectors as cs	9✔
19	import yaml	9✔
20	from matplotlib import figure	9✔
21	from matplotlib.widgets import MultiCursor	9✔
22	from scipy.special import comb	9✔
23	from sklearn.preprocessing import PowerTransformer, StandardScaler	9✔
24	from tqdm.auto import tqdm	9✔
25
26	from rnalysis import __version__	9✔
27
28	try:	9✔
29	import numba	9✔
30	except ImportError: # pragma: no cover
31	warnings.warn("RNAlysis can perform faster when package 'numba' is installed. \n"
32	"If you want to improve the performance of slow operations on RNAlysis, "
33	"please install package 'numba'. ")
34
35
36	class numba: # pragma: no cover
37	@staticmethod
38	def jit(args, *kwargs):
39	return lambda f: f
40
41	@staticmethod
42	def njit(args, *kwargs):
43	return lambda f: f
44
45
46	def readable_name(name: str):	9✔
47	def decorator(item):	9✔
48	item.readable_name = name	9✔
49	return item	9✔
50
51	return decorator	9✔
52
53
54	class TemporaryMatplotlibBackend:	9✔
55	def __init__(self, backend):	9✔
56	self.backend = backend	9✔
57	self.original_backend = matplotlib.get_backend()	9✔
58
59	def __enter__(self):	9✔
60	matplotlib.use(self.backend, force=True)	9✔
61
62	def __exit__(self, exc_type, exc_val, exc_tb):	9✔
63	matplotlib.use(self.original_backend, force=True)	9✔
64
65
66	class ProgressParallel(joblib.Parallel):	9✔
67	# tqdm progress bar for parallel tasks based on:
68	# https://stackoverflow.com/questions/37804279/how-can-we-use-tqdm-in-a-parallel-execution-with-joblib/50925708
69	# answer by 'user394430'
70	def __init__(self, use_tqdm=True, total=None, desc: str = '', unit: str = 'it', backend='loky', args, *kwargs):	9✔
71	self._use_tqdm = use_tqdm	9✔
72	self._total = total	9✔
73	self._desc = desc	9✔
74	self._unit = unit	9✔
75	kwargs['n_jobs'] = -2	9✔
76	super().__init__(args, backend=backend, *kwargs, verbose=100)	9✔
77
78	def __call__(self, args, *kwargs):	9✔
79	fmt = '{desc}: {percentage:3.0f}%\|{bar}\| [{elapsed}<{remaining}, {rate_fmt}{postfix}]'	9✔
80	with tqdm(disable=not self._use_tqdm, total=self._total, desc=self._desc, unit=self._unit,	9✔
81	bar_format=fmt) as self._pbar:
82	return joblib.Parallel.__call__(self, args, *kwargs)	9✔
83
84	def print_progress(self):	9✔
85	if self._total is None:	9✔
86	self._pbar.total = self.n_dispatched_tasks	9✔
87	self._pbar.n = self.n_completed_tasks	9!
88	self._pbar.refresh()	9✔
89
90
91	def standard_box_cox(data: Union[np.ndarray, pl.DataFrame]) -> Union[np.ndarray, pl.DataFrame]:	9✔
92	"""
93
94	:param data:
95	:type data:
96	:return:
97	:rtype:
98	"""
99	if isinstance(data, pl.DataFrame):	9✔
100	array = data.select(cs.numeric()).to_numpy()	9✔
101	else:
102	array = data	9✔
103	res_array = StandardScaler().fit_transform(PowerTransformer(method='box-cox').fit_transform(array + 1))	9✔
104	if isinstance(data, pl.DataFrame):	9✔
105	return data.select(~cs.numeric()).with_columns(	9✔
106	pl.DataFrame(res_array, schema=data.select(cs.numeric()).columns))
107	return res_array	9✔
108
109
110	def shift_to_baseline(data: Union[np.ndarray, pl.DataFrame], baseline: float = 0) -> Union[np.ndarray, pl.DataFrame]:	9✔
111	"""
112
113	:param data:
114	:type data:
115	:param baseline:
116	:type baseline:
117	:return:
118	:rtype:
119	"""
120	if isinstance(data, pl.DataFrame):	9✔
121	array = data.select(cs.numeric()).to_numpy()	9✔
122	elif isinstance(data,pl.Series):	9✔
UNCOV 123	array = data.to_numpy()	×
124	else:
125	array = data	9✔
126	min_val = array.min()	9✔
127
128	while isinstance(min_val, np.ndarray):	9✔
UNCOV 129	min_val = min_val.min()	×
130	diff = baseline - min_val	9✔
131	return data + diff	9✔
132
133
134	def standardize(data: Union[np.ndarray, pl.DataFrame]) -> Union[np.ndarray, pl.DataFrame]:	9✔
135	"""
136
137	:param data:
138	:type data:
139	:return:
140	:rtype:
141	"""
142	if isinstance(data, pl.DataFrame):	9✔
143	array = data.select(cs.numeric()).to_numpy()	9✔
144	else:
145	array = data	9✔
146	res_array = StandardScaler().fit_transform(array)	9✔
147	if isinstance(data, pl.DataFrame):	9✔
148	return data.select(~cs.numeric()).with_columns(pl.DataFrame(res_array, schema=data.columns))	9✔
149	return res_array	9✔
150
151
152	def intersection_nonempty(*objs: Union[list, set, tuple]):	9✔
153	return set.intersection(*[set(item) for item in objs if len(item) > 0])	9✔
154
155
156	def color_generator():	9✔
157	"""
158	A generator that yields distinct colors up to a certain limit, and then yields randomized RGB values.
159
160	:return: a color name string (like 'black', \
161	or a numpy.ndarray of size (3,) containing three random values each between 0 and 1.
162
163	"""
164	preset_colors = ['tab:blue', 'tab:red', 'tab:green', 'tab:orange', 'tab:purple', 'tab:brown', 'tab:pink',	9✔
165	'tab:gray', 'tab:olive', 'tab:cyan', 'gold', 'maroon', 'mediumslateblue', 'fuchsia',
166	'lawngreen', 'moccasin', 'thistle']
167	for color in preset_colors:	9✔
168	yield color	9✔
169	while True:	7✔
170	yield np.random.random(3)	9✔
171
172
173	@lru_cache(maxsize=2 ** 16)	9✔
174	def combination(a: int, b: int) -> int:	9✔
175	return int(comb(a, b))	9✔
176
177
178	class SetWithMajorityVote(set):	9✔
179	def majority_vote_intersection(self, *others, majority_threshold: float = 0.5):	9✔
180	counts = {}	9✔
181	n_sets = len(others) + 1	9✔
182	for set_obj in itertools.chain([self], others):	9✔
183	for obj in set_obj:	9✔
184	counts[obj] = counts.get(obj, 0) + (1 / n_sets)	9✔
185	result = {key for key, val in counts.items() if val >= majority_threshold}	9✔
186	return result	9✔
187
188
189	def get_method_readable_name(method: Union[str, Callable], obj: object = None):	9✔
190	try:	9✔
191	if isinstance(method, str):	9✔
192	func = getattr(obj, method)	9✔
193	else:
194	func = method	9✔
195	if hasattr(func, 'readable_name'):	9✔
196	name = func.readable_name	9✔
197	else:
198	name = func.__name__	9✔
199	return name	9✔
UNCOV 200	except AttributeError:	×
UNCOV 201	return 'NAME NOT FOUND'	×
202
203
204	def get_method_signature(method: Union[str, Callable], obj: object = None):	9✔
205	try:	9✔
206	if isinstance(method, str):	9✔
207	func = getattr(obj, method)	9✔
208	else:
209	func = method	9✔
210	signature = inspect.signature(func)	9✔
211	return signature.parameters	9✔
212	except AttributeError:	9✔
213	return {}	9✔
214
215
216	def despine(ax):	9✔
217	for side in ['top', 'right']:	9✔
218	ax.spines[side].set_visible(False)	9✔
219
220
221	def mix_colors(*colors: Tuple[float, float, float]):	9✔
222	n_colors = len(colors)	9✔
223	if n_colors == 1:	9✔
224	return colors[0]	9✔
225
226	colors = np.array(colors)	9✔
227	multiplier = (1 / n_colors) + (1.6 / (2 ** n_colors)) / n_colors	9✔
228
229	mix_color = multiplier * np.sum(colors, axis=0)	9✔
230	mix_color = np.min([mix_color, [1.0, 1.0, 1.0]], 0)	9✔
231	return mix_color	9✔
232
233
234	def sum_intervals_inclusive(intervals: typing.List[typing.Tuple[int, int]]) -> int:	9✔
235	sorted_intervals = sorted(intervals)	9✔
236	total = 0	9✔
237	prev_interval = (-np.inf, -np.inf)	9✔
238	for i in range(len(sorted_intervals)):	9✔
239	this_interval = sorted_intervals[i]	9✔
240	if this_interval[0] <= prev_interval[1]:	9✔
241	this_interval = (prev_interval[1] + 1, this_interval[1])	9✔
242	total += this_interval[1] - this_interval[0] + 1	9✔
243	prev_interval = this_interval	9✔
244
245	return total	9✔
246
247
248	def bic_score(X: np.ndarray, labels: np.ndarray):	9✔
249	"""
250	BIC score for the goodness of fit of clusters.
251	This Python function is translated from the Golang implementation by the author of the paper.
252	The original code is available here:
253	https://github.com/bobhancock/goxmeans/blob/a78e909e374c6f97ddd04a239658c7c5b7365e5c/km.go#L778
254	"""
255
256	n_points = len(labels)	9✔
257	n_clusters = len(set(labels))	9✔
258	n_dimensions = X.shape[1]	9✔
259
260	n_parameters = (n_clusters - 1) + (n_dimensions * n_clusters) + 1	9✔
261
262	loglikelihood = 0	9✔
263	for label_name in set(labels):	9✔
264	X_cluster = X[labels == label_name]	9✔
265	n_points_cluster = len(X_cluster)	9✔
266	centroid = np.mean(X_cluster, axis=0)	9✔
267	variance = np.sum((X_cluster - centroid) ** 2) / (len(X_cluster) - 1)	9✔
268	loglikelihood += \	9✔
269	n_points_cluster * (np.log(n_points_cluster) - np.log(n_points) - (n_dimensions / 2) * np.log(
270	2 * math.pi * variance) - (n_points_cluster - 1) / 2)
271
272	bic = loglikelihood - (n_parameters / 2) * np.log(n_points)	9✔
273
274	return bic	9✔
275
276
277	def format_time(seconds: float):	9✔
278	seconds = int(seconds)	9✔
279	n_minutes = seconds // 60	9✔
280	n_seconds = seconds % 60	9✔
281	return f'{n_minutes:02d}:{n_seconds:02d}'	9✔
282
283
284	def sanitize_variable_name(name: str) -> str:	9✔
285	"""
286	Sanitize a string to turn it into a legal variable name in R/Python.
287	:param name: name to sanitize
288	:type name: str
289	:return: sanitizeed name
290	:rtype: str
291	"""
292	new_name = name.rstrip().replace(' ', '_')	9✔
293	if new_name[0].isdigit():	9✔
294	new_name = 'var_' + new_name	9✔
295
296	if not new_name.isalnum():	9✔
297	new_name = ''.join([char if char.isalnum() else '_' for char in new_name])	9✔
298
299	return new_name	9✔
300
301
302	class InteractiveScatterFigure(figure.Figure):	9✔
303	def __init__(self, labels: typing.List[str], annotation_fontsize: float = 10, show_cursor: bool = True, *args,	9✔
304	**kwargs):
305	super().__init__(args, *kwargs)	9✔
306	self.ax = self.add_subplot()	9✔
307	self.labels = labels	9✔
308	self.annotation_fontsize = annotation_fontsize	9✔
309	self.is_labeled: typing.Dict[int, plt.Line2D] = {}	9✔
310	self.canvas.mpl_connect('pick_event', self.on_pick)	9✔
311	if show_cursor:	9✔
312	self.cursor = MultiCursor(self.canvas, self.axes, color='k', lw=0.5, horizOn=True, vertOn=True)	9✔
313	self.canvas.mpl_connect('axes_leave_event', self.on_exit)	9✔
314	self.canvas.mpl_connect('motion_notify_event', self.on_move)	9✔
315
316	def on_exit(self, event):	9✔
317	self.cursor.clear(event)	9✔
318	self.canvas.draw()	9✔
319
320	def on_move(self, event):	9✔
321	self.canvas.draw()	9✔
322
323	def on_pick(self, event):	9✔
324	for this_ind in event.ind:	9✔
325	if isinstance(event.artist, matplotlib.collections.PathCollection):	9✔
326	data = event.artist.get_offsets().data	9✔
327	xdata, ydata = data[:, 0], data[:, 1]	9✔
328	elif isinstance(event.artist, plt.Line2D):	9✔
UNCOV 329	thisline = event.artist	×
UNCOV 330	xdata = thisline.get_xdata()	×
UNCOV 331	ydata = thisline.get_ydata()	×
332	else:
333	return	9✔
334
335	if this_ind in self.is_labeled:	9✔
336	ann = self.is_labeled.pop(this_ind)	9✔
337	ann.remove()	9✔
338	else:
339	self.is_labeled[this_ind] = plt.annotate(self.labels[this_ind],	9✔
340	(np.take(xdata, this_ind), np.take(ydata, this_ind)),
341	xytext=(3, 3), textcoords='offset points',
342	fontsize=self.annotation_fontsize)
343
344	self.canvas.draw()	9✔
345
346
347	class GenericPipeline(abc.ABC):	9✔
348	__slots__ = {'functions': 'list of functions to perform', 'params': 'list of function parameters'}	9✔
349
350	def __init__(self):	9✔
351	self.functions = []	9✔
352	self.params = []	9✔
353
354	def __len__(self):	9✔
355	return len(self.functions)	9✔
356
357	def __eq__(self, other):	9✔
358	if self.functions == other.functions and self.params == other.params:	9✔
359	return True	9✔
UNCOV 360	return False	×
361
362	@staticmethod	9✔
363	def _param_string(args: tuple, kwargs: dict):	9✔
364
365	"""
366	Returns a formatted string of the given arguments and keyworded arguments.
367
368	:param args: arguments to format as string
369	:type args: tuple
370	:param kwargs: keyworded arguments to format as string
371	:type kwargs: dict
372	:return: a formatted string of arguments and keyworded argumentss
373	:rtype: str
374
375	"""
376	args_str = ', '.join([f"'{arg}'" if isinstance(arg, str) else f"{arg}" for arg in args])	9✔
377	kwargs_str = ', '.join(	9✔
378	[f"{key}='{arg}'" if isinstance(arg, str) else f"{key}={arg}" for key, arg in kwargs.items()])
379	if len(args_str) == 0:	9✔
380	return kwargs_str	9✔
381	elif len(kwargs_str) == 0:	9✔
382	return args_str	9✔
383	else:
384	return f"{args_str}, {kwargs_str}"	9✔
385
386	def remove_last_function(self):	9✔
387
388	"""
389	Removes from the Pipeline the last function that was added to it. Removal is in-place.
390
391	:Examples:
392	>>> from rnalysis import filtering
393	>>> pipe = filtering.Pipeline()
394	>>> pipe.add_function(filtering.Filter.filter_missing_values)
395	Added function 'Filter.filter_missing_values()' to the pipeline.
396	>>> pipe.remove_last_function()
397	Removed function filter_missing_values with parameters [] from the pipeline.
398
399	"""
400	assert len(self.functions) > 0 and len(self.params) > 0, "Pipeline is empty, no functions to remove!"	9✔
401	func = self.functions.pop(-1)	9✔
402	args, kwargs = self.params.pop(-1)	9✔
403	print(	9✔
404	f"Removed function {func.__name__} with parameters [{self._param_string(args, kwargs)}] from the pipeline.")
405
406	def _readable_func_signature(self, func: types.FunctionType, args: tuple, kwargs: dict):	9✔
407	"""
408	Returns a human-readable string functions signature for the given function and arguments.
409
410	:param func: the function or method to generate signature for
411	:type func: function
412	:param args: arguments given for the function
413	:type args: tuple
414	:param kwargs: keyworded arguments given for the function
415	:type kwargs: dict
416	:return: function signature string
417	:rtype: str
418	"""
419	return f"{get_method_readable_name(func)}: ({self._param_string(args, kwargs)})"	9✔
420
421	def export_pipeline(self, filename: Union[str, Path, None]) -> Union[None, str]:	9✔
422	"""
423	Export a Pipeline to a Pipeline YAML file or YAML-like string.
424
425	:param filename: filename to save the Pipeline YAML to, or None to return a YAML-like string instead.
426	:type filename: str, pathlib.Path, or None
427	:return: if filename is None, returns the Pipeline YAML-like string.
428	"""
429	pipeline_dict = self._get_pipeline_dict()	9✔
430	for func, params in zip(self.functions, self.params):	9✔
431	pipeline_dict['functions'].append(func.__name__)	9✔
432	pipeline_dict['params'].append(params)	9✔
433	if filename is None:	9✔
434	return yaml.safe_dump(pipeline_dict)	9✔
435	else:
436	with open(filename, 'w') as f:	9✔
437	yaml.safe_dump(pipeline_dict, f)	9✔
438
439	def _get_pipeline_dict(self):	9✔
440	d = dict(functions=[], params=[], metadata={'rnalysis_version': f'{__version__}',	9✔
441	'export_time': datetime.today().strftime('%Y/%m/%d, %H:%M:%S')})
442	return d	9✔
443
444	@classmethod	9✔
445	def import_pipeline(cls, filename: Union[str, Path]) -> 'GenericPipeline':	9✔
446	"""
447	Import a Pipeline from a Pipeline YAML file or YAML-like string.
448
449	:param filename: name of the YAML file containing the Pipeline, or a YAML-like string.
450	:type filename: str or pathlib.Path
451	:return: the imported Pipeline
452	:rtype: Pipeline
453	"""
454	try:	9✔
455	with open(filename) as f:	9✔
456	pipeline_dict = yaml.safe_load(f)	9✔
457	except OSError:	9✔
458	pipeline_dict = yaml.safe_load(filename)	9✔
459	pipeline = cls.__new__(cls)	9✔
460	pipeline._init_from_dict(pipeline_dict)	9✔
461	return pipeline	9✔
462
463	def _init_from_dict(self, pipeline_dict: dict):	9✔
464	raise NotImplementedError
465
466	def _func_signature(self, func: types.FunctionType, args: tuple, kwargs: dict):	9✔
467	"""
468	Returns a string functions signature for the given function and arguments.
469
470	:param func: the function or method to generate signature for
471	:type func: function
472	:param args: arguments given for the function
473	:type args: tuple
474	:param kwargs: keyworded arguments given for the function
475	:type kwargs: dict
476	:return: function signature string
477	:rtype: str
478	"""
479	return f"{func.__name__}({self._param_string(args, kwargs)})"	9✔
480
481	def add_function(self, func: types.FunctionType, args, *kwargs):	9✔
482	assert isinstance(func, types.FunctionType), f"'func' must be a function, is {type(func)} instead."	9✔
483
484	self.functions.append(func)	9✔
485	self.params.append((args, kwargs))	9✔
486	print(f"Added function '{self._func_signature(func, args, kwargs)}' to the pipeline.")	9✔
487
488	def _validate_pipeline(self):	9✔
489	assert len(self.functions) > 0 and len(self.params) > 0, "Cannot apply an empty pipeline!"	9✔
490	assert len(self.functions) == len(self.params), f"Cannot apply Pipeline: " \	9✔
491	f"length of 'functions' different from length of 'params'!"
492
493	@abc.abstractmethod	9✔
494	def apply_to(self, args, *kwargs):	9✔
495	raise NotImplementedError
496
497
498	def jitter(n: int, jitter_range: float = 0.2) -> np.ndarray:	9✔
499	if n == 0:	9✔
UNCOV 500	return np.array([])	×
501	if jitter_range == 0:	9✔
502	return np.zeros(n)	9✔
503
504	jittered = np.random.uniform(-jitter_range, jitter_range, n)	9✔
505	jittered -= np.mean(jittered) # center the jittered points	9✔
506	scaling_factor = jitter_range / max(np.max(np.abs(jittered)), 1e-6)	9✔
507	jittered *= scaling_factor # scale the jitter points to the jitter range	9✔
508	return jittered	9✔

GuyTeichman / RNAlysis / 9644630967

Source File Press 'n' to go to next uncovered line, 'b' for previous

Source File
Press 'n' to go to next uncovered line, 'b' for previous