Created helper functions in radp library and mobility_model notebook

notebooks/coo_with_radp_digital_twin.ipynb

@@ -522,7 +522,7 @@
    "name": "python",
    "nbconvert_exporter": "python",
    "pygments_lexer": "ipython3",
-   "version": "3.9.6"
+   "version": "3.11.5"
   },
   "varInspector": {
    "cols": {

notebooks/radp_library.py

@@ -12,6 +12,7 @@
 import fastkml
 import matplotlib.animation as animation
 import matplotlib.pyplot as plt
+import matplotlib.cm as cm
 import numpy as np
 import pandas as pd
 import rasterio.features
@@ -25,6 +26,9 @@
     NormMethod,
 )
 from radp.digital_twin.utils.gis_tools import GISTools
+from radp.digital_twin.mobility.ue_tracks_params import UETracksGenerationParams
+from radp.digital_twin.mobility.ue_tracks import UETracksGenerator
+from radp.digital_twin.mobility.param_regression import ParameterRegression
 
 Boundary = Union[geometry.Polygon, geometry.MultiPolygon]
 KML_NS = "{http://www.opengis.net/kml/2.2}"
@@ -936,3 +940,239 @@ def rfco_to_best_server_shapes(
         key=lambda x: x[1],
     )
     return shapes
+
+
+# Mobility Model helper functions
+
+
+def get_ue_data(params: dict) -> pd.DataFrame:
+    """
+    Generates user equipment (UE) tracks data using specified simulation parameters.
+
+    This function initializes a UETracksGenerationParams object using the provided parameters
+    and then iterates over batches generated by the UETracksGenerator. Each batch of UE tracks
+    data is consolidated into a single DataFrame which captures mobility tracks across multiple
+    ticks and batches, as per the defined parameters.
+
+    Using the UETracksGenerator, the UE tracks are returned in form of a dataframe
+    The Dataframe is arranged as follows:
+
+        +------------+------------+-----------+------+
+        | mock_ue_id | lon        | lat       | tick |
+        +============+============+===========+======+
+        |   0        | 102.219377 | 33.674572 |   0  |
+        |   1        | 102.415954 | 33.855534 |   0  |
+        |   2        | 102.545935 | 33.878075 |   0  |
+        |   0        | 102.297766 | 33.575942 |   1  |
+        |   1        | 102.362725 | 33.916477 |   1  |
+        |   2        | 102.080675 | 33.832793 |   1  |
+        +------------+------------+-----------+------+
+    """
+
+    # Initialize the UE data
+    data = UETracksGenerationParams(params)
+
+    ue_tracks_generation = pd.DataFrame()  # Initialize an empty DataFrame
+    for ue_tracks_generation_batch in UETracksGenerator.generate_as_lon_lat_points(
+        rng_seed=data.rng_seed,
+        lon_x_dims=data.lon_x_dims,
+        lon_y_dims=data.lon_y_dims,
+        num_ticks=data.num_ticks,
+        num_UEs=data.num_UEs,
+        num_batches=data.num_batches,
+        alpha=data.alpha,
+        variance=data.variance,
+        min_lat=data.min_lat,
+        max_lat=data.max_lat,
+        min_lon=data.min_lon,
+        max_lon=data.max_lon,
+        mobility_class_distribution=data.mobility_class_distribution,
+        mobility_class_velocities=data.mobility_class_velocities,
+        mobility_class_velocity_variances=data.mobility_class_velocity_variances,
+    ):
+        # Append each batch to the main DataFrame
+        if ue_tracks_generation.empty:
+            ue_tracks_generation = ue_tracks_generation_batch
+        else:
+            ue_tracks_generation = pd.concat(
+                [ue_tracks_generation, ue_tracks_generation_batch], ignore_index=True
+            )
+
+    return ue_tracks_generation
+
+
+def plot_ue_tracks(df) -> None:
+    """
+    Plots the movement tracks of unique UE IDs on a grid of subplots.
+    """
+
+    # Initialize an empty list to store batch indices
+    batch_indices = []
+
+    # Identify where tick resets and mark the indices
+    for i in range(1, len(df)):
+        if df.loc[i, "tick"] == 0 and df.loc[i - 1, "tick"] != 0:
+            batch_indices.append(i)
+
+    # Add the final index to close the last batch
+    batch_indices.append(len(df))
+
+    # Now, iterate over the identified batches
+    start_idx = 0
+    for batch_num, end_idx in enumerate(batch_indices):
+        batch_data = df.iloc[start_idx:end_idx]
+
+        # Create a new figure
+        plt.figure(figsize=(10, 6))
+
+        # Generate a color map with different colors for each ue_id
+        color_map = cm.get_cmap("tab20", len(batch_data["mock_ue_id"].unique()))
+
+        # Plot each ue_id's movement over ticks in this batch
+        for idx, ue_id in enumerate(batch_data["mock_ue_id"].unique()):
+            ue_data = batch_data[batch_data["mock_ue_id"] == ue_id]
+            color = color_map(idx)  # Get a unique color for each ue_id
+
+            # Plot the path with arrows
+            for i in range(len(ue_data) - 1):
+                x_start = ue_data.iloc[i]["lon"]
+                y_start = ue_data.iloc[i]["lat"]
+                x_end = ue_data.iloc[i + 1]["lon"]
+                y_end = ue_data.iloc[i + 1]["lat"]
+
+                # Calculate the direction vector
+                dx = x_end - x_start
+                dy = y_end - y_start
+
+                # Plot the line with an arrow with reduced width and unique color
+                plt.quiver(
+                    x_start,
+                    y_start,
+                    dx,
+                    dy,
+                    angles="xy",
+                    scale_units="xy",
+                    scale=1,
+                    color=color,
+                    width=0.002,
+                    headwidth=3,
+                    headlength=5,
+                )
+
+            # Plot starting points as circles with the same color
+            plt.scatter(
+                ue_data["lon"].iloc[0],
+                ue_data["lat"].iloc[0],
+                color=color,
+                label=f"Start UE {ue_id}",
+            )
+
+        # Set plot title and labels
+        plt.title(f"UE Tracks with Direction for Batch {batch_num + 1}")
+        plt.xlabel("Longitude")
+        plt.ylabel("Latitude")
+        plt.legend(loc="upper right", bbox_to_anchor=(1.2, 1))
+
+        # Display the plot
+        plt.show()
+
+        # Update start_idx for the next batch
+        start_idx = end_idx
+
+def plot_ue_tracks_side_by_side(df1, df2):
+    """
+    Plots the movement tracks of unique UE IDs from two DataFrames side by side.
+    """
+    # Set up subplots with 2 columns for side by side plots
+    fig, axes = plt.subplots(1, 2, figsize=(25, 10))  # 2 rows, 2 columns (side by side)
+
+    # Plot the first DataFrame
+    plot_ue_tracks_on_axis(df1, axes[0], title='DataFrame 1')
+
+    # Plot the second DataFrame
+    plot_ue_tracks_on_axis(df2, axes[1], title='DataFrame 2')
+
+    # Adjust layout and show
+    plt.tight_layout()
+    plt.show()
+
+def plot_ue_tracks_on_axis(df, ax, title):
+    """
+    Helper function to plot UE tracks on a given axis.
+    """
+    data = df
+    unique_ids = data['mock_ue_id'].unique()
+    num_plots = len(unique_ids)
+
+    color_map = cm.get_cmap('tab20', num_plots)
+
+    for idx, ue_id in enumerate(unique_ids):
+        ue_data = data[data['mock_ue_id'] == ue_id]
+
+        for i in range(len(ue_data) - 1):
+            x_start = ue_data.iloc[i]['lon']
+            y_start = ue_data.iloc[i]['lat']
+            x_end = ue_data.iloc[i + 1]['lon']
+            y_end = ue_data.iloc[i + 1]['lat']
+
+            dx = x_end - x_start
+            dy = y_end - y_start
+            ax.quiver(x_start, y_start, dx, dy, angles='xy', scale_units='xy', scale=1, color=color_map(idx))
+
+        ax.scatter(ue_data['lon'], ue_data['lat'], color=color_map(idx), label=f'UE {ue_id}')
+
+    ax.set_title(title)
+    ax.legend()
+
+
+def calculate_distances_and_velocities(group):
+    """Calculating distances and velocities for each UE based on sorted data by ticks."""
+    group["prev_longitude"] = group["lon"].shift(1)
+    group["prev_latitude"] = group["lat"].shift(1)
+    group["distance"] = group.apply(
+        lambda row: GISTools.get_log_distance(
+            row["prev_latitude"], row["prev_longitude"], row["lat"], row["lon"]
+        )
+        if not pd.isna(row["prev_longitude"])
+        else 0,
+        axis=1,
+    )
+    # Assuming time interval between ticks is 1 unit, adjust below if different
+    group["velocity"] = (
+        group["distance"] / 1
+    )  # Convert to m/s by dividing by the seconds per tick, here assumed to be 1s
+    return group
+
+
+def preprocess_ue_data(data):
+    """Preprocessing data to calculate distances and velocities for each UE."""
+    # Ensure data is sorted by UE ID and tick to ensure accurate shift operations
+    data.sort_values(by=["mock_ue_id", "tick"], inplace=True)
+    data = data.groupby("mock_ue_id").apply(calculate_distances_and_velocities)
+
+    # Drop the temporary columns
+    data.drop(["prev_longitude", "prev_latitude", "distance"], axis=1, inplace=True)
+
+    return data
+
+
+def get_predicted_alpha(data, alpha0):
+    """
+    Estimate the alpha parameter for a Gauss-Markov mobility model using regression analysis
+    on the velocity data derived from user equipment (UE) tracks.
+
+    This function processes the provided UE track data to calculate velocities,
+    fits a regression model using polynomial regression and non linear least squares,
+    to estimate the alpha parameter that best describes the randomness or directionality in the mobility pattern,
+    and returns the optimized alpha.
+    """
+    # Preprocess data to calculate velocities
+    velocity_df = preprocess_ue_data(data)
+
+    # ParameterRegression is used to regress the data to predict alpha using velocity
+    regression = ParameterRegression(velocity_df)
+
+    # Optimize alpha using the initial guess alpha0
+    predicted_alpha, predicted_cov = regression.optimize_alpha(alpha0)
+
+    return float(predicted_alpha)