Repository: modin-project/modin
Pull request: FEAT-#2342: Add axis partitions API
<!--
Thank you for your contribution!
Please review the contributing docs: https://modin.readthedocs.io/en/latest/CONTRIBUTING.html
if you have questions about contributing.
-->

## What do these changes do?

<!-- Please give a short brief about these changes. -->

- [x] commit message follows format outlined [here](https://modin.readthedocs.io/en/latest/CONTRIBUTING.html)
- [x] passes `flake8 modin`
- [x] passes `black --check modin`
- [x] signed commit with `git commit -s` <!-- you can amend your commit with a signature via `git commit -amend -s` -->
- [x] Resolves #2342  <!-- issue must be created for each patch -->
- [x] tests passing


Problem statement:
Row Partition API
**Is your feature request related to a problem? Please describe.**
A clear and concise description of what the problem is. What kind of performance improvements would you like to see with this new API?

I'd like to be able to obtain a list of row partitions from a distributed Modin dataframe.

This would allow me to easily do distributed data-parallel training by feeding in these dataframe shards. This would be useful for both XGBoost and Horovod.

cc @tgaddair

Hints from issue discussion:
Thanks @richardliaw for the request!

We should consider a developer API for cases like this that enable access to Modin internal partitions without needing to dig into them. I have been thinking about this some.

Design can look something like this:

```python
from modin.developer.partitioning import extract_row_partitions, extract_col_partitions
import ray

row_oid_list = extract_row_partitions(df, assert_type=ray.ObjectRef)
col_oid_list =  extract_col_partitions(df, assert_type=ray.ObjectRef)
```

Thoughts @richardliaw?

cc @aregm 
I ran into a similar issue.
In addition to being able to extract the partitions it would be good to know the locality of each partition. Knowing where a partition (currently) resides can be used to minimize data movement when passing partitions to ray actors. Minimal data movement requires pairing actors and partitions which are 'pinned' to the same node.
@devin-petersohn sounds good!
I have a prototype ready to try. Install via `pip install git+https://github.com/devin-petersohn/modin@feat/2342`

Usage is simple:
```python
from modin.developer import unwrap_row_partitions

unwrap_row_partitions(df)
```

You will get a list of `ray.ObjectRef` each contain a row partition.
Thanks, works great for me!
What about getting sufficient information to be able to map data to local actors?
How would this work with modin on dask?
> Thanks, works great for me!
> What about getting sufficient information to be able to map data to local actors?

Maybe it would be good to have some kind of `RowPartitionsActor` and `ColPartitionsActor` that will keep row partitions and column partitions, respectively, assigned to them. Also, these actors should keep some info on node ip address (it is supposed an actor per node).
> How would this work with modin on dask?

It will work the same way as for ray, but you will get a list of `Dask.Future` instead. If I am not correct, @devin-petersohn , please, correct me.
It will work the same, if you are expecting a certain engine, there is a keyword argument in the API for the engine. It will make sure that the objects are compatible with that engine or throw an error if they are not.
> Maybe it would be good to have some kind of RowPartitionsActor and ColPartitionsActor that will keep row partitions and column partitions, respectively, assigned to them. Also, these actors should keep some info on node ip address (it is supposed an actor per node).

Do mean as a modin-internal entity or something the user should use? How would a user benefit from this, can you provide a short made-up example?

Commit message: FEAT-#2342: Move to common function of partitions unwrapping

Signed-off-by: Igoshev, Yaroslav <yaroslav.igoshev@intel.com>

Patch to review:
diff --git a/modin/api/__init__.py b/modin/api/__init__.py
new file mode 100644
index 00000000000..ae69fb73221
--- /dev/null
+++ b/modin/api/__init__.py
@@ -0,0 +1,16 @@
+# Licensed to Modin Development Team under one or more contributor license agreements.
+# See the NOTICE file distributed with this work for additional information regarding
+# copyright ownership.  The Modin Development Team licenses this file to you under the
+# Apache License, Version 2.0 (the "License"); you may not use this file except in
+# compliance with the License.  You may obtain a copy of the License at
+#
+#     http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software distributed under
+# the License is distributed on an "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF
+# ANY KIND, either express or implied. See the License for the specific language
+# governing permissions and limitations under the License.
+
+from .partition_api import unwrap_partitions
+
+__all__ = ["unwrap_partitions"]
diff --git a/modin/api/partition_api.py b/modin/api/partition_api.py
new file mode 100644
index 00000000000..ca0dde69f3d
--- /dev/null
+++ b/modin/api/partition_api.py
@@ -0,0 +1,75 @@
+# Licensed to Modin Development Team under one or more contributor license agreements.
+# See the NOTICE file distributed with this work for additional information regarding
+# copyright ownership.  The Modin Development Team licenses this file to you under the
+# Apache License, Version 2.0 (the "License"); you may not use this file except in
+# compliance with the License.  You may obtain a copy of the License at
+#
+#     http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software distributed under
+# the License is distributed on an "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF
+# ANY KIND, either express or implied. See the License for the specific language
+# governing permissions and limitations under the License.
+
+
+def unwrap_partitions(api_layer_object, axis=None, bind_ip=False):
+    """
+    Unwrap partitions of the `api_layer_object`.
+
+    Parameters
+    ----------
+    api_layer_object : DataFrame or Series
+        The API layer object.
+    axis : None, 0 or 1. Default is None
+        The axis to unwrap partitions for (0 - row partitions, 1 - column partitions).
+        If axis is None, all the partitions of the API layer object are unwrapped.
+    bind_ip : boolean. Default is False
+        Whether to bind node ip address to each partition or not.
+
+    Returns
+    -------
+    list
+        A list of Ray.ObjectRef/Dask.Future to partitions of the `api_layer_object`
+        if Ray/Dask is used as an engine.
+
+    Notes
+    -----
+    In case bind_ip=True, a list containing tuples of Ray.ObjectRef/Dask.Future to node ip addresses
+    and partitions of the `api_layer_object`, respectively, is returned if Ray/Dask is used as an engine.
+    """
+    if not hasattr(api_layer_object, "_query_compiler"):
+        raise ValueError(
+            f"Only API Layer objects may be passed in here, got {type(api_layer_object)} instead."
+        )
+
+    if axis is None:
+
+        def _unwrap_partitions(oid):
+            return [
+                tuple(partition.ip, getattr(partition, oid))
+                if bind_ip
+                else getattr(partition, oid)
+                for row in api_layer_object._query_compiler._modin_frame._partitions
+                for partition in row
+            ]
+
+        actual_engine = type(
+            api_layer_object._query_compiler._modin_frame._partitions[0][0]
+        ).__name__
+        if any(actual_engine == engine for engine in ["PandasOnRayFramePartition"]):
+            return _unwrap_partitions("oid")
+        elif any(actual_engine == engine for engine in ["PandasOnDaskFramePartition"]):
+            return _unwrap_partitions("future")
+        raise ValueError(
+            f"Do not know how to unwrap '{actual_engine}' underlying partitions"
+        )
+    else:
+        partitions = (
+            api_layer_object._query_compiler._modin_frame._frame_mgr_cls.axis_partition(
+                api_layer_object._query_compiler._modin_frame._partitions, axis ^ 1
+            )
+        )
+        return [
+            part.coalesce(bind_ip=bind_ip).unwrap(squeeze=True, bind_ip=bind_ip)
+            for part in partitions
+        ]
diff --git a/modin/engines/base/frame/axis_partition.py b/modin/engines/base/frame/axis_partition.py
index aa7d565c6da..2c42ee9cc5a 100644
--- a/modin/engines/base/frame/axis_partition.py
+++ b/modin/engines/base/frame/axis_partition.py
@@ -95,6 +95,56 @@ def shuffle(self, func, lengths, **kwargs):
     def _wrap_partitions(self, partitions):
         return [self.partition_type(obj) for obj in partitions]
 
+    def coalesce(self, bind_ip=False):
+        """
+        Coalesce the axis partitions into a single partition.
+
+        Parameters
+        ----------
+        bind_ip : boolean, default False
+            Whether to bind node ip address to a single partition or not.
+
+        Returns
+        -------
+        BaseFrameAxisPartition
+            An axis partition containing only a single coalesced partition.
+        """
+        coalesced = self.apply(lambda x: x, num_splits=1, maintain_partitioning=False)
+        return type(self)(coalesced, bind_ip=bind_ip)
+
+    def unwrap(self, squeeze=False, bind_ip=False):
+        """
+        Unwrap partitions from axis partition.
+
+        Parameters
+        ----------
+        squeeze : boolean, default False
+            The flag used to unwrap only one partition.
+        bind_ip : boolean, default False
+            Whether to bind node ip address to each partition or not.
+
+        Returns
+        -------
+        list
+            List of partitions from axis partition.
+
+        Notes
+        -----
+        In case bind_ip=True, list containing tuples of Ray.ObjectRef/Dask.Future
+        to node ip addresses and unwrapped partitions, respectively, is returned
+        if Ray/Dask is used as an engine.
+        """
+        if squeeze and len(self.list_of_blocks) == 1:
+            if bind_ip:
+                return self.list_of_ips[0], self.list_of_blocks[0]
+            else:
+                return self.list_of_blocks[0]
+        else:
+            if bind_ip:
+                return list(zip(self.list_of_ips, self.list_of_blocks))
+            else:
+                return self.list_of_blocks
+
 
 class PandasFrameAxisPartition(BaseFrameAxisPartition):
     """An abstract class is created to simplify and consolidate the code for AxisPartitions that run pandas.
diff --git a/modin/engines/dask/pandas_on_dask/frame/axis_partition.py b/modin/engines/dask/pandas_on_dask/frame/axis_partition.py
index dbd5538aaaf..8ee82996cc8 100644
--- a/modin/engines/dask/pandas_on_dask/frame/axis_partition.py
+++ b/modin/engines/dask/pandas_on_dask/frame/axis_partition.py
@@ -16,14 +16,18 @@
 
 from distributed.client import get_client
 from distributed import Future
+from distributed.utils import get_ip
+import pandas
 
 
 class PandasOnDaskFrameAxisPartition(PandasFrameAxisPartition):
-    def __init__(self, list_of_blocks):
+    def __init__(self, list_of_blocks, bind_ip=False):
         # Unwrap from BaseFramePartition object for ease of use
         for obj in list_of_blocks:
             obj.drain_call_queue()
         self.list_of_blocks = [obj.future for obj in list_of_blocks]
+        if bind_ip:
+            self.list_of_ips = [obj.ip for obj in list_of_blocks]
 
     partition_type = PandasOnDaskFramePartition
     instance_type = Future
@@ -34,6 +38,7 @@ def deploy_axis_func(
     ):
         client = get_client()
         axis_result = client.submit(
+            deploy_dask_func,
             PandasFrameAxisPartition.deploy_axis_func,
             axis,
             func,
@@ -51,7 +56,7 @@ def deploy_axis_func(
         # get futures for each.
         return [
             client.submit(lambda l: l[i], axis_result, pure=False)
-            for i in range(result_num_splits)
+            for i in range(result_num_splits * 4)
         ]
 
     @classmethod
@@ -60,6 +65,7 @@ def deploy_func_between_two_axis_partitions(
     ):
         client = get_client()
         axis_result = client.submit(
+            deploy_dask_func,
             PandasFrameAxisPartition.deploy_func_between_two_axis_partitions,
             axis,
             func,
@@ -74,7 +80,13 @@ def deploy_func_between_two_axis_partitions(
         # get futures for each.
         return [
             client.submit(lambda l: l[i], axis_result, pure=False)
-            for i in range(num_splits)
+            for i in range(num_splits * 4)
+        ]
+
+    def _wrap_partitions(self, partitions):
+        return [
+            self.partition_type(future, length, width, ip)
+            for (future, length, width, ip) in zip(*[iter(partitions)] * 4)
         ]
 
 
@@ -94,3 +106,26 @@ class PandasOnDaskFrameRowPartition(PandasOnDaskFrameAxisPartition):
     """
 
     axis = 1
+
+
+def deploy_dask_func(func, *args):
+    """
+    Run a function on a remote partition.
+
+    Parameters
+    ----------
+    func : callable
+        The function to run.
+
+    Returns
+    -------
+        The result of the function `func`.
+    """
+    result = func(*args)
+    ip = get_ip()
+    if isinstance(result, pandas.DataFrame):
+        return result, len(result), len(result.columns), ip
+    elif all(isinstance(r, pandas.DataFrame) for r in result):
+        return [i for r in result for i in [r, len(r), len(r.columns), ip]]
+    else:
+        return [i for r in result for i in [r, None, None, ip]]
diff --git a/modin/engines/dask/pandas_on_dask/frame/partition.py b/modin/engines/dask/pandas_on_dask/frame/partition.py
index 5c592e810bd..bbc16a27e19 100644
--- a/modin/engines/dask/pandas_on_dask/frame/partition.py
+++ b/modin/engines/dask/pandas_on_dask/frame/partition.py
@@ -17,6 +17,7 @@
 from modin.data_management.utils import length_fn_pandas, width_fn_pandas
 
 from distributed.client import get_client
+from distributed.utils import get_ip
 import cloudpickle as pkl
 
 
@@ -25,7 +26,7 @@ def apply_list_of_funcs(funcs, df):
         if isinstance(func, bytes):
             func = pkl.loads(func)
         df = func(df, **kwargs)
-    return df
+    return df, get_ip()
 
 
 class PandasOnDaskFramePartition(BaseFramePartition):
@@ -40,13 +41,14 @@ class PandasOnDaskFramePartition(BaseFramePartition):
     subclasses. There is no logic for updating inplace.
     """
 
-    def __init__(self, future, length=None, width=None, call_queue=None):
+    def __init__(self, future, length=None, width=None, ip=None, call_queue=None):
         self.future = future
         if call_queue is None:
             call_queue = []
         self.call_queue = call_queue
         self._length_cache = length
         self._width_cache = width
+        self.ip = ip
 
     def get(self):
         """Flushes the call_queue and returns the data.
@@ -81,7 +83,8 @@ def apply(self, func, **kwargs):
         future = get_client().submit(
             apply_list_of_funcs, call_queue, self.future, pure=False
         )
-        return PandasOnDaskFramePartition(future)
+        futures = [get_client().submit(lambda l: l[i], future) for i in range(2)]
+        return PandasOnDaskFramePartition(futures[0], ip=futures[1])
 
     def add_to_apply_calls(self, func, **kwargs):
         return PandasOnDaskFramePartition(
@@ -91,7 +94,9 @@ def add_to_apply_calls(self, func, **kwargs):
     def drain_call_queue(self):
         if len(self.call_queue) == 0:
             return
-        self.future = self.apply(lambda x: x).future
+        new_partition = self.apply(lambda x: x)
+        self.future = new_partition.future
+        self.ip = new_partition.ip
         self.call_queue = []
 
     def mask(self, row_indices, col_indices):
diff --git a/modin/engines/ray/pandas_on_ray/frame/axis_partition.py b/modin/engines/ray/pandas_on_ray/frame/axis_partition.py
index 9ec6620eae9..621802387f3 100644
--- a/modin/engines/ray/pandas_on_ray/frame/axis_partition.py
+++ b/modin/engines/ray/pandas_on_ray/frame/axis_partition.py
@@ -17,14 +17,17 @@
 from .partition import PandasOnRayFramePartition
 
 import ray
+from ray.services import get_node_ip_address
 
 
 class PandasOnRayFrameAxisPartition(PandasFrameAxisPartition):
-    def __init__(self, list_of_blocks):
+    def __init__(self, list_of_blocks, bind_ip=False):
         # Unwrap from BaseFramePartition object for ease of use
         for obj in list_of_blocks:
             obj.drain_call_queue()
         self.list_of_blocks = [obj.oid for obj in list_of_blocks]
+        if bind_ip:
+            self.list_of_ips = [obj.ip for obj in list_of_blocks]
 
     partition_type = PandasOnRayFramePartition
     instance_type = ray.ObjectID
@@ -44,7 +47,7 @@ def deploy_axis_func(
                 maintain_partitioning,
             )
             + tuple(partitions),
-            num_returns=num_splits * 3 if lengths is None else len(lengths) * 3,
+            num_returns=num_splits * 4 if lengths is None else len(lengths) * 4,
         )
 
     @classmethod
@@ -62,13 +65,13 @@ def deploy_func_between_two_axis_partitions(
                 kwargs,
             )
             + tuple(partitions),
-            num_returns=num_splits * 3,
+            num_returns=num_splits * 4,
         )
 
     def _wrap_partitions(self, partitions):
         return [
-            self.partition_type(partitions[i], partitions[i + 1], partitions[i + 2])
-            for i in range(0, len(partitions), 3)
+            self.partition_type(object_id, length, width, ip)
+            for (object_id, length, width, ip) in zip(*[iter(partitions)] * 4)
         ]
 
 
@@ -92,20 +95,27 @@ class PandasOnRayFrameRowPartition(PandasOnRayFrameAxisPartition):
 
 @ray.remote
 def deploy_ray_func(func, *args):  # pragma: no cover
-    """Run a function on a remote partition.
-
-    Note: Ray functions are not detected by codecov (thus pragma: no cover)
+    """
+    Run a function on a remote partition.
 
-    Args:
-        func: The function to run.
+    Parameters
+    ----------
+    func : callable
+        The function to run.
 
-    Returns:
+    Returns
+    -------
         The result of the function `func`.
+
+    Notes
+    -----
+    Ray functions are not detected by codecov (thus pragma: no cover)
     """
     result = func(*args)
+    ip = get_node_ip_address()
     if isinstance(result, pandas.DataFrame):
-        return result, len(result), len(result.columns)
+        return result, len(result), len(result.columns), ip
     elif all(isinstance(r, pandas.DataFrame) for r in result):
-        return [i for r in result for i in [r, len(r), len(r.columns)]]
+        return [i for r in result for i in [r, len(r), len(r.columns), ip]]
     else:
-        return [i for r in result for i in [r, None, None]]
+        return [i for r in result for i in [r, None, None, ip]]
diff --git a/modin/engines/ray/pandas_on_ray/frame/partition.py b/modin/engines/ray/pandas_on_ray/frame/partition.py
index b309fd7094c..140b42f60a2 100644
--- a/modin/engines/ray/pandas_on_ray/frame/partition.py
+++ b/modin/engines/ray/pandas_on_ray/frame/partition.py
@@ -18,11 +18,12 @@
 from modin.engines.ray.utils import handle_ray_task_error
 
 import ray
+from ray.services import get_node_ip_address
 from ray.worker import RayTaskError
 
 
 class PandasOnRayFramePartition(BaseFramePartition):
-    def __init__(self, object_id, length=None, width=None, call_queue=None):
+    def __init__(self, object_id, length=None, width=None, ip=None, call_queue=None):
         assert type(object_id) is ray.ObjectID
 
         self.oid = object_id
@@ -31,6 +32,7 @@ def __init__(self, object_id, length=None, width=None, call_queue=None):
         self.call_queue = call_queue
         self._length_cache = length
         self._width_cache = width
+        self.ip = ip
 
     def get(self):
         """Gets the object out of the plasma store.
@@ -60,8 +62,8 @@ def apply(self, func, **kwargs):
         """
         oid = self.oid
         call_queue = self.call_queue + [(func, kwargs)]
-        result, length, width = deploy_ray_func.remote(call_queue, oid)
-        return PandasOnRayFramePartition(result, length, width)
+        result, length, width, ip = deploy_ray_func.remote(call_queue, oid)
+        return PandasOnRayFramePartition(result, length, width, ip)
 
     def add_to_apply_calls(self, func, **kwargs):
         return PandasOnRayFramePartition(
@@ -73,14 +75,17 @@ def drain_call_queue(self):
             return
         oid = self.oid
         call_queue = self.call_queue
-        self.oid, self._length_cache, self._width_cache = deploy_ray_func.remote(
-            call_queue, oid
-        )
+        (
+            self.oid,
+            self._length_cache,
+            self._width_cache,
+            self.ip,
+        ) = deploy_ray_func.remote(call_queue, oid)
         self.call_queue = []
 
     def __copy__(self):
         return PandasOnRayFramePartition(
-            self.oid, self._length_cache, self._width_cache, self.call_queue
+            self.oid, self._length_cache, self._width_cache, call_queue=self.call_queue
         )
 
     def to_pandas(self):
@@ -202,7 +207,7 @@ def get_index_and_columns(df):
     return len(df.index), len(df.columns)
 
 
-@ray.remote(num_returns=3)
+@ray.remote(num_returns=4)
 def deploy_ray_func(call_queue, partition):  # pragma: no cover
     def deserialize(obj):
         if isinstance(obj, ray.ObjectID):
@@ -231,4 +236,5 @@ def deserialize(obj):
         result,
         len(result) if hasattr(result, "__len__") else 0,
         len(result.columns) if hasattr(result, "columns") else 0,
+        get_node_ip_address(),
     )
diff --git a/modin/pandas/utils.py b/modin/pandas/utils.py
index 1c631514c15..072520bae1c 100644
--- a/modin/pandas/utils.py
+++ b/modin/pandas/utils.py
@@ -24,7 +24,7 @@ def from_non_pandas(df, index, columns, dtype):
     ----------
     What arguments does this function have.
     [
-    PARAMETER_NAME: PARAMETERS TYPES
+    PARAMETER_NAME : PARAMETERS TYPES
         Description.
     ]
 
@@ -48,7 +48,7 @@ def from_pandas(df):
 
     Parameters
     ----------
-    df: pandas.DataFrame
+    df : pandas.DataFrame
         The pandas DataFrame to convert.
 
     Returns
@@ -67,7 +67,7 @@ def from_arrow(at):
 
     Parameters
     ----------
-    at: Arrow Table
+    at : Arrow Table
         The Arrow Table to convert from.
 
     Returns
@@ -93,7 +93,7 @@ def is_scalar(obj):
 
     Parameters
     ----------
-    val: object
+    val : object
         Object to check.
 
     Returns


Code review comments:
diff_hunk: @@ -0,0 +1,75 @@
+# Licensed to Modin Development Team under one or more contributor license agreements.
+# See the NOTICE file distributed with this work for additional information regarding
+# copyright ownership.  The Modin Development Team licenses this file to you under the
+# Apache License, Version 2.0 (the "License"); you may not use this file except in
+# compliance with the License.  You may obtain a copy of the License at
+#
+#     http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software distributed under
+# the License is distributed on an "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF
+# ANY KIND, either express or implied. See the License for the specific language
+# governing permissions and limitations under the License.
+
+
+def unwrap_partitions(api_layer_object, axis=None, bind_ip=False):
+    """
+    Unwrap partitions of the `api_layer_object`.
+
+    Parameters
+    ----------
+    api_layer_object : DataFrame or Series
+        The API layer object.
+    axis : None, 0 or 1. Default is None
+        The axis to unwrap partitions for (0 - row partitions, 1 - column partitions).
+        If axis is None, all the partitions of the API layer object are unwrapped.
+    bind_ip : boolean. Default is False
+        Whether to bind node ip address to each partition or not.
+
+    Returns
+    -------
+    list
+        A list of Ray.ObjectRef/Dask.Future to partitions of the `api_layer_object`
+        if Ray/Dask is used as an engine.
+
+    Notes
+    -----
+    In case bind_ip=True, a list containing tuples of Ray.ObjectRef/Dask.Future to node ip addresses
+    and partitions of the `api_layer_object`, respectively, is returned if Ray/Dask is used as an engine.
+    """
+    if not hasattr(api_layer_object, "_query_compiler"):
+        raise ValueError(
+            f"Only API Layer objects may be passed in here, got {type(api_layer_object)} instead."
+        )
+
+    if axis is None:
+
+        def _unwrap_partitions(oid):
+            return [
+                tuple(partition.ip, getattr(partition, oid))
+                if bind_ip
+                else getattr(partition, oid)
+                for row in api_layer_object._query_compiler._modin_frame._partitions
+                for partition in row
+            ]
path: modin/api/partition_api.py
text: @user1:
My suggestion is this:
```suggestion
        def _unwrap_partitions(oid):
            if bind_ip:
                return [
                    (partition.ip, getattr(partition, oid))
                    for row in api_layer_object._query_compiler._modin_frame._partitions
                    for partition in row
                ]
            else:
                return [
                    getattr(partition, oid)
                    for row in api_layer_object._query_compiler._modin_frame._partitions
                    for partition in row
                ]
```

Basically I'm suggesting to move the check _outside_ of the loop completely.

@author:
Applied.