"""
Mapper and InvertibleMapper
---------------------------
"""
from collections.abc import Mapping
from typing import Any, Callable, Union
import numpy as np
import pandas as pd
from darts import TimeSeries
from darts.dataprocessing.transformers.base_data_transformer import BaseDataTransformer
from darts.dataprocessing.transformers.invertible_data_transformer import (
InvertibleDataTransformer,
)
from darts.logging import get_logger
logger = get_logger(__name__)
MapperFn = Union[
Callable[[np.number], np.number], Callable[[pd.Timestamp, np.number], np.number]
]
[docs]
class Mapper(BaseDataTransformer):
def __init__(
self,
fn: Union[
Callable[[np.number], np.number],
Callable[[pd.Timestamp, np.number], np.number],
],
name: str = "Mapper",
n_jobs: int = 1,
verbose: bool = False,
):
"""
Data transformer to apply a custom function to a (sequence of) ``TimeSeries``
(similar to calling :func:`TimeSeries.map()` on each series).
The mapper takes care of parallelizing the operations on multiple series over
multiple processors.
Parameters
----------
fn
Either a function which takes a value and returns a value ie. `f(x) = y`
Or a function which takes a value and its timestamp and returns a value ie. `f(timestamp, x) = y`.
name
A specific name for the transformer.
n_jobs
The number of jobs to run in parallel. Parallel jobs are created only when a ``Sequence[TimeSeries]`` is
passed as input to a method, parallelising operations regarding different ``TimeSeries``. Defaults to `1`
(sequential). Setting the parameter to `-1` means using all the available processors.
Note: for a small amount of data, the parallelisation overhead could end up increasing the total
required amount of time.
verbose
Optionally, whether to print operations progress
Examples
--------
>>> import numpy as np
>>> from darts import TimeSeries
>>> from darts.dataprocessing.transformers import Mapper
>>> series = TimeSeries.from_values(np.array([1, 10, 100]))
>>> transformer = Mapper(np.log10)
>>> series_transformed = transformer.transform(series)
>>> print(series_transformed.values())
[[0.]
[1.]
[2.]]
"""
# Define fixed params (i.e. attributes defined before calling `super().__init__`):
self._fn = fn
super().__init__(name=name, n_jobs=n_jobs, verbose=verbose)
[docs]
class InvertibleMapper(InvertibleDataTransformer):
def __init__(
self,
fn: Union[
Callable[[np.number], np.number],
Callable[[pd.Timestamp, np.number], np.number],
],
inverse_fn: Union[
Callable[[np.number], np.number],
Callable[[pd.Timestamp, np.number], np.number],
],
name: str = "InvertibleMapper",
n_jobs: int = 1,
verbose: bool = False,
):
"""
Data transformer to apply a custom function and its inverse to a (sequence of) ``TimeSeries``
(similar to calling :func:`TimeSeries.map()` on each series).
Parameters
----------
fn
Either a function which takes a value and returns a value ie. `f(x) = y`
Or a function which takes a value and its timestamp and returns a value ie. `f(timestamp, x) = y`.
inverse_fn
Similarly to `fn`, either a function which takes a value and returns a value ie. `f(x) = y`
Or a function which takes a value and its timestamp and returns a value ie. `f(timestamp, x) = y`.
`inverse_fn` should be such that ``inverse_fn(fn(x)) == x``.
name
A specific name for the transformer.
n_jobs
The number of jobs to run in parallel. Parallel jobs are created only when a `Sequence[TimeSeries]` is
passed as input to a method, parallelising operations regarding different `TimeSeries`. Defaults to `1`
(sequential). Setting the parameter to `-1` means using all the available processors.
Note: for a small amount of data, the parallelisation overhead could end up increasing the total
required amount of time.
verbose
Optionally, whether to print operations progress
Examples
--------
>>> import numpy as np
>>> from darts import TimeSeries
>>> from darts.dataprocessing.transformers import InvertibleMapper
>>> series = TimeSeries.from_values(np.array([1, 10, 100]))
>>> transformer = InvertibleMapper(np.log10, lambda x: 10 ** x)
>>> series_transformed = transformer.transform(series)
>>> print(series_transformed.values())
[[0.]
[1.]
[2.]]
[3.]]
>>> series_restored = transformer.inverse_transform(series_transformed)
>>> print(series_restored.values())
[[ 1.]
[ 10.]
[100.]]
"""
self._fn = fn
self._inverse_fn = inverse_fn
super().__init__(name=name, n_jobs=n_jobs, verbose=verbose)
