How to use func_kwargs method in assertpy

Best Python code snippet using assertpy_python

apply.py

Source:apply.py

1import pandas as pd234def apply_by(5    df: pd.core.frame.DataFrame,6    func,7    func_kwargs: dict = None,8    spiketimes_col: str = "spiketimes",9    spiketrain_col: str = "spiketrain",10    returned_colname: str = "apply_result",11):12    """13    Apply an arbitrary function to each spiketrain in a DataFrame.1415    The passed function should have a single return value for each spiketrain.1617    Args:18        df: A pandas DataFrame containing spiketimes indexed by spiketrain19        func: The function to apply to the data20        func_kwargs: dictionary of key-word arguments to be passed to the function21        spiketimes_col: The label of the column containing spiketimes22        spiketrain_col: The label of the column containing spiketrain identifiers23        return_colname: The label of the column in the returned DataFrame containing the function result24    Returns:25        A pandas DataFrame with columns {spiketrian_col and returned_colname}26    """27    if not func_kwargs:28        func_kwargs = {}29    res = (30        df.groupby(spiketrain_col)31        .apply(lambda x: func(x[spiketimes_col].values, **func_kwargs))32        .reset_index()33        .rename(columns={0: returned_colname})34    )35    if "level_1" in res.columns:36        res = res.rename(columns={"level_1": f"{returned_colname}_idx"})37    return res383940def apply_by_rolling(41    df: pd.core.frame.DataFrame,42    func,43    num_periods: int = 10,44    func_kwargs: dict = None,45    spiketimes_col: str = "spiketimes",46    spiketrain_col: str = "spiketrain",47    returned_colname: str = "rolling_result",48    copy: bool = True,49):50    """51    Apply a function in a roling window along each neuron in a dataframe5253    Args:54        df: A pandas DataFrame containing spiketimes indexed by spiketrain55        func: funtion to apply along the datafrmae56        num_period: The number of rows in the rolling window57        spiketimes_col: The label of the column containing spiketimes58        spiketrain_col: The label of the column containing spiketrain identifiers59        returned_colname: The label of the column in the returned DataFrame containing the function result60        copy: Whether make a copy of the passed to DataFrame before applying the function61    Returns:62        A copy of the passed DataFrame with returned_colname appended63    """64    original_index_name = df.index.name65    if not func_kwargs:66        func_kwargs = {}67    if copy:68        df = df.copy()69    tmp_res = (70        df.groupby(spiketrain_col)[spiketimes_col]71        .rolling(num_periods)72        .apply(lambda x: func(x.values, **func_kwargs), raw=True)73        .reset_index()74        .rename(columns={spiketimes_col: returned_colname})75        .set_index("level_1")76    )77    tmp_res.index.name = "index"78    tmp_res = pd.merge(df.reset_index(), tmp_res.reset_index()).set_index("index")79    tmp_res.index.name = original_index_name
...

bootstrap.py

Source:bootstrap.py

1import numpy as np2def bootstrap(data, n, axis=0, func=np.var, func_kwargs={"ddof": 1}):3    """Produce n bootstrap samples of data of the statistic given by func.4    Arguments5    ---------6    data : numpy.ndarray7        Data to resample.8    n : int9        Number of bootstrap trails.10    axis : int, optional11        Axis along which to resample. (Default ``0``).12    func : callable, optional13        Statistic to calculate. (Default ``numpy.var``).14    func_kwargs : dict, optional15        Dictionary with extra arguments for func. (Default ``{"ddof" : 1}``).16    Returns17    -------18    samples : numpy.ndarray19        Bootstrap samples of statistic func on the data.20    """21    if axis != 0:22        raise NotImplementedError("Only axis == 0 supported.")23    fiducial_output = func(data, axis=axis, **func_kwargs)24    if isinstance(data, list):25        assert all([d.shape[1:] == data[0].shape[1:] for d in data])26    samples = np.zeros((n, *fiducial_output.shape),27                       dtype=fiducial_output.dtype)28    for i in range(n):29        if isinstance(data, list):30            idx = [np.random.choice(d.shape[0], size=d.shape[0], replace=True)31                   for d in data]32            samples[i] = func([d[i] for d, i in zip(data, idx)],33                              axis=axis, **func_kwargs)34        else:35            idx = np.random.choice(data.shape[axis], size=data.shape[axis],36                                   replace=True)37            samples[i] = func(data[idx], axis=axis, **func_kwargs)38    return samples39def bootstrap_var(data, n, axis=0, func=np.var, func_kwargs={"ddof": 1}):40    """Calculate the variance of the statistic given by func.41    Arguments42    ---------43    data : numpy.ndarray44        Data to resample.45    n : int46        Number of bootstrap trails.47    axis : int, optional48        Axis along which to resample. (Default ``0``).49    func : callable, optional50        Statistic to calculate. (Default ``numpy.var``).51    func_kwargs : dict, optional52        Dictionary with extra arguments for func. (Default ``{"ddof" : 1}``).53    Returns54    -------55    var : numpy.ndarray56        Bootstrap variance of statistic func on the data.57    """58    samples = bootstrap(data, n, axis, func, func_kwargs)...

common_task.py

Source:common_task.py

1# -*- coding:utf-8 -*-2import sys3from .celery_worker import app4def normal_task(func, func_args=None, func_kwargs=None, **kwargs):5    """6    :param func: å½æ°å¯¹è±¡7    :param func_args: å½æ°çåæ° list8    :param func_kwargs: å½æ°çåæ° dict9    :param args: celery send_taskæ¹æ³çåæ°10    :param kwargs: celery send_taskæ¹æ³çåæ°11    :return:12    """13    # åæ°æ£æ¥14    if func_args is not None and not isinstance(func_args, list):15        raise Exception('invalid func_args type')16    if func_kwargs is not None and not isinstance(func_kwargs, dict):17        raise Exception('invalid func_kwargs type')18    args_list = [sys.path, func.__module__, func.__name__]19    if func_args is None:20        func_args = []21    func_args = args_list + func_args22    app.send_task('common_async_task', args=func_args, kwargs=func_kwargs, **kwargs)23def async_http_request(method, url, func_kwargs=None, **kwargs):24    """25    å¼æ¥httpè¯·æ±26    :param method: è¯·æ±ç±»å: "GET","POST","PATCH",...27    :param url: è¯·æ±çå°å28    :param func_kwargs: requestsåæ°29    :param kwargs: celeryåæ°30    :return:31    """32    if func_kwargs is not None and not isinstance(func_kwargs, dict):33        raise Exception('invalid func_kwargs type')...

Automation Testing Tutorials

Learn to execute automation testing from scratch with LambdaTest Learning Hub. Right from setting up the prerequisites to run your first automation test, to following best practices and diving deeper into advanced test scenarios. LambdaTest Learning Hubs compile a list of step-by-step guides to help you be proficient with different test automation frameworks i.e. Selenium, Cypress, TestNG etc.