6520232780

Committed 14 Oct 2023 10:12PM UTC coverage: 90.267% (-0.3%) from 90.603%

Build # 6520232780

Build Type

Pull #181

github

Committed by

rettigl

Commit Message

define jitter_amps as single amplitude in default config

Pull Request Pull Request #181: define jitter_amps as single amplitude in default config

Run Details

4229 of 4685 relevant lines covered (90.27%)

0.9 hits per line

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

86.11

/sed/core/dfops.py

"""This module contains dataframe operations functions for the sed package

"""
# Note: some of the functions presented here were
# inspired by https://github.com/mpes-kit/mpes
from typing import Callable
from typing import Sequence
from typing import Union

import dask.dataframe
import numpy as np
import pandas as pd


def apply_jitter(
    df: Union[pd.DataFrame, dask.dataframe.DataFrame],
    cols: Union[str, Sequence[str]],
    cols_jittered: Union[str, Sequence[str]] = None,
    amps: Union[float, Sequence[float]] = 0.5,
    jitter_type: str = "uniform",
) -> Union[pd.DataFrame, dask.dataframe.DataFrame]:
    """Add jittering to one or more dataframe columns.

    Args:
        df (Union[pd.DataFrame, dask.dataframe.DataFrame]): Dataframe to add
            noise/jittering to.
        cols (Union[str, Sequence[str]]): Names of the columns to add jittering to.
        cols_jittered (Union[str, Sequence[str]], optional): Names of the columns
            with added jitter. Defaults to None.
        amps (Union[float, Sequence[float]], optional): Amplitude scalings for the
            jittering noise. If one number is given, the same is used for all axes.
            For normal noise, the added noise will have sdev [-amp, +amp], for
            uniform noise it will cover the interval [-amp, +amp].
            Defaults to 0.5.
        jitter_type (str, optional): the type of jitter to add. 'uniform' or 'normal'
            distributed noise. Defaults to "uniform".

    Returns:
        Union[pd.DataFrame, dask.dataframe.DataFrame]: dataframe with added columns.
    """
    assert cols is not None, "cols needs to be provided!"
    assert jitter_type in (
        "uniform",
        "normal",
    ), "type needs to be one of 'normal', 'uniform'!"

    if isinstance(cols, str):
        cols = [cols]
    if isinstance(cols_jittered, str):
        cols_jittered = [cols_jittered]
    if cols_jittered is None:
        cols_jittered = [col + "_jittered" for col in cols]
    if isinstance(amps, float):
        amps = list(np.ones(len(cols)) * amps)

    colsize = df[cols[0]].size

    if jitter_type == "uniform":
        # Uniform Jitter distribution
        jitter = np.random.uniform(low=-1, high=1, size=colsize)
    elif jitter_type == "normal":
        # Normal Jitter distribution works better for non-linear transformations and
        # jitter sizes that don't match the original bin sizes
        jitter = np.random.standard_normal(size=colsize)

    for (col, col_jittered, amp) in zip(cols, cols_jittered, amps):
        df[col_jittered] = df[col] + amp * jitter

    return df


def drop_column(
    df: Union[pd.DataFrame, dask.dataframe.DataFrame],
    column_name: Union[str, Sequence[str]],
) -> Union[pd.DataFrame, dask.dataframe.DataFrame]:
    """Delete columns.

    Args:
        df (Union[pd.DataFrame, dask.dataframe.DataFrame]): Dataframe to use.
        column_name (Union[str, Sequence[str]])): List of column names to be dropped.

    Returns:
        Union[pd.DataFrame, dask.dataframe.DataFrame]: Dataframe with dropped columns.
    """
    out_df = df.drop(column_name, axis=1)

    return out_df


def apply_filter(
    df: Union[pd.DataFrame, dask.dataframe.DataFrame],
    col: str,
    lower_bound: float = -np.inf,
    upper_bound: float = np.inf,
) -> Union[pd.DataFrame, dask.dataframe.DataFrame]:
    """Application of bound filters to a specified column (can be used consecutively).

    Args:
        df (Union[pd.DataFrame, dask.dataframe.DataFrame]): Dataframe to use.
        col (str): Name of the column to filter.
        lower_bound (float, optional): The lower bound used in the filtering.
            Defaults to -np.inf.
        upper_bound (float, optional): The lower bound used in the filtering.
            Defaults to np.inf.

    Returns:
        Union[pd.DataFrame, dask.dataframe.DataFrame]: The filtered dataframe.
    """
    out_df = df[(df[col] > lower_bound) & (df[col] < upper_bound)]

    return out_df


def map_columns_2d(
    df: Union[pd.DataFrame, dask.dataframe.DataFrame],
    map_2d: Callable,
    x_column: np.ndarray,
    y_column: np.ndarray,
    **kwds,
) -> Union[pd.DataFrame, dask.dataframe.DataFrame]:
    """Apply a 2-dimensional mapping simultaneously to two dimensions.

    Args:
        df (Union[pd.DataFrame, dask.dataframe.DataFrame]): Dataframe to use.
        map_2d (Callable): 2D mapping function.
        x_column (np.ndarray): The X column of the dataframe to apply mapping to.
        y_column (np.ndarray): The Y column of the dataframe to apply mapping to.
        **kwds: Additional arguments for the 2D mapping function.

    Returns:
        Union[pd.DataFrame, dask.dataframe.DataFrame]: Dataframe with mapped columns.
    """
    new_x_column = kwds.pop("new_x_column", x_column)
    new_y_column = kwds.pop("new_y_column", y_column)

    (df[new_x_column], df[new_y_column]) = map_2d(
        df[x_column],
        df[y_column],
        **kwds,
    )

    return df

1	"""This module contains dataframe operations functions for the sed package
2
3	"""
4	# Note: some of the functions presented here were
5	# inspired by https://github.com/mpes-kit/mpes
6	from typing import Callable	1✔
7	from typing import Sequence	1✔
8	from typing import Union	1✔
9
10	import dask.dataframe	1✔
11	import numpy as np	1✔
12	import pandas as pd	1✔
13
14
15	def apply_jitter(	1✔
16	df: Union[pd.DataFrame, dask.dataframe.DataFrame],
17	cols: Union[str, Sequence[str]],
18	cols_jittered: Union[str, Sequence[str]] = None,
19	amps: Union[float, Sequence[float]] = 0.5,
20	jitter_type: str = "uniform",
21	) -> Union[pd.DataFrame, dask.dataframe.DataFrame]:
22	"""Add jittering to one or more dataframe columns.
23
24	Args:
25	df (Union[pd.DataFrame, dask.dataframe.DataFrame]): Dataframe to add
26	noise/jittering to.
27	cols (Union[str, Sequence[str]]): Names of the columns to add jittering to.
28	cols_jittered (Union[str, Sequence[str]], optional): Names of the columns
29	with added jitter. Defaults to None.
30	amps (Union[float, Sequence[float]], optional): Amplitude scalings for the
31	jittering noise. If one number is given, the same is used for all axes.
32	For normal noise, the added noise will have sdev [-amp, +amp], for
33	uniform noise it will cover the interval [-amp, +amp].
34	Defaults to 0.5.
35	jitter_type (str, optional): the type of jitter to add. 'uniform' or 'normal'
36	distributed noise. Defaults to "uniform".
37
38	Returns:
39	Union[pd.DataFrame, dask.dataframe.DataFrame]: dataframe with added columns.
40	"""
41	assert cols is not None, "cols needs to be provided!"	1✔
42	assert jitter_type in (	1✔
43	"uniform",
44	"normal",
45	), "type needs to be one of 'normal', 'uniform'!"
46
47	if isinstance(cols, str):	1✔
48	cols = [cols]	×
49	if isinstance(cols_jittered, str):	1✔
50	cols_jittered = [cols_jittered]	×
51	if cols_jittered is None:	1✔
52	cols_jittered = [col + "_jittered" for col in cols]	×
53	if isinstance(amps, float):	1✔
54	amps = list(np.ones(len(cols)) * amps)	1✔
55
56	colsize = df[cols[0]].size	1✔
57
58	if jitter_type == "uniform":	1✔
59	# Uniform Jitter distribution
60	jitter = np.random.uniform(low=-1, high=1, size=colsize)	1✔
61	elif jitter_type == "normal":	×
62	# Normal Jitter distribution works better for non-linear transformations and
63	# jitter sizes that don't match the original bin sizes
64	jitter = np.random.standard_normal(size=colsize)	×
65
66	for (col, col_jittered, amp) in zip(cols, cols_jittered, amps):	1✔
67	df[col_jittered] = df[col] + amp * jitter	1✔
68
69	return df	1✔
70
71
72	def drop_column(	1✔
73	df: Union[pd.DataFrame, dask.dataframe.DataFrame],
74	column_name: Union[str, Sequence[str]],
75	) -> Union[pd.DataFrame, dask.dataframe.DataFrame]:
76	"""Delete columns.
77
78	Args:
79	df (Union[pd.DataFrame, dask.dataframe.DataFrame]): Dataframe to use.
80	column_name (Union[str, Sequence[str]])): List of column names to be dropped.
81
82	Returns:
83	Union[pd.DataFrame, dask.dataframe.DataFrame]: Dataframe with dropped columns.
84	"""
85	out_df = df.drop(column_name, axis=1)	1✔
86
87	return out_df	1✔
88
89
90	def apply_filter(	1✔
91	df: Union[pd.DataFrame, dask.dataframe.DataFrame],
92	col: str,
93	lower_bound: float = -np.inf,
94	upper_bound: float = np.inf,
95	) -> Union[pd.DataFrame, dask.dataframe.DataFrame]:
96	"""Application of bound filters to a specified column (can be used consecutively).
97
98	Args:
99	df (Union[pd.DataFrame, dask.dataframe.DataFrame]): Dataframe to use.
100	col (str): Name of the column to filter.
101	lower_bound (float, optional): The lower bound used in the filtering.
102	Defaults to -np.inf.
103	upper_bound (float, optional): The lower bound used in the filtering.
104	Defaults to np.inf.
105
106	Returns:
107	Union[pd.DataFrame, dask.dataframe.DataFrame]: The filtered dataframe.
108	"""
109	out_df = df[(df[col] > lower_bound) & (df[col] < upper_bound)]	1✔
110
111	return out_df	1✔
112
113
114	def map_columns_2d(	1✔
115	df: Union[pd.DataFrame, dask.dataframe.DataFrame],
116	map_2d: Callable,
117	x_column: np.ndarray,
118	y_column: np.ndarray,
119	**kwds,
120	) -> Union[pd.DataFrame, dask.dataframe.DataFrame]:
121	"""Apply a 2-dimensional mapping simultaneously to two dimensions.
122
123	Args:
124	df (Union[pd.DataFrame, dask.dataframe.DataFrame]): Dataframe to use.
125	map_2d (Callable): 2D mapping function.
126	x_column (np.ndarray): The X column of the dataframe to apply mapping to.
127	y_column (np.ndarray): The Y column of the dataframe to apply mapping to.
128	**kwds: Additional arguments for the 2D mapping function.
129
130	Returns:
131	Union[pd.DataFrame, dask.dataframe.DataFrame]: Dataframe with mapped columns.
132	"""
133	new_x_column = kwds.pop("new_x_column", x_column)	1✔
134	new_y_column = kwds.pop("new_y_column", y_column)	1✔
135
136	(df[new_x_column], df[new_y_column]) = map_2d(	1✔
137	df[x_column],
138	df[y_column],
139	**kwds,
140	)
141
142	return df	1✔

OpenCOMPES / sed / 6520232780

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