Skip to content

Merging

merging

merge_trajectories 

merge_trajectories(
    trajectories: list[Trajectory],
) -> Trajectory

Merges a list of trajectories into one trajectory.

This function ignores EPSG codes and merges the trajectories based on their timestamps. Therefore, all trajectories should be in the same coordinate system before merging.

Parameters:

  • trajectories (List[Trajectory]) –

    List of trajectories to merge.

Returns:

  • Trajectory ( Trajectory ) –

    Merged trajectory.

Raises:

  • ValueError –

    If no trajectories are provided.

Source code in trajectopy\processing\merging.py 
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
def merge_trajectories(trajectories: list[Trajectory]) -> Trajectory:
    """Merges a list of trajectories into one trajectory.

    This function ignores EPSG codes and merges the trajectories based on their timestamps.
    Therefore, all trajectories should be in the same coordinate system before merging.

    Args:
        trajectories (List[Trajectory]): List of trajectories to merge.

    Returns:
        Trajectory: Merged trajectory.

    Raises:
        ValueError: If no trajectories are provided.
    """
    if not trajectories:
        raise ValueError("Cannot merge empty list of trajectories")

    epsg_set = {t.positions.epsg for t in trajectories}

    if len(epsg_set) > 1:
        logger.warning(
            "Merging trajectories with different EPSG codes. "
            "This may lead to unexpected results. "
            "Consider reprojecting the trajectories to the same EPSG code."
        )

    epsg = trajectories[0].positions.epsg  # the first trajectory's, not an arbitrary one of the set

    merged_xyz = np.concatenate([t.positions.xyz for t in trajectories], axis=0)
    # orientations only if all trajectories have them: any substitute would be made-up data
    with_orientation = all(t.rotations is not None for t in trajectories)
    if not with_orientation and any(t.rotations is not None for t in trajectories):
        logger.warning("Not all merged trajectories have orientations. The merged trajectory has none.")
    merged_rotations = (
        Rotations.from_quat(np.concatenate([t.quat for t in trajectories], axis=0)) if with_orientation else None
    )
    merged_timestamps = np.concatenate([t.timestamps for t in trajectories], axis=0)
    merged_velocities = np.concatenate([t.velocity_xyz for t in trajectories], axis=0)

    # the trajectory orders the poses chronologically before it derives the path lengths
    return Trajectory(
        name="Merged",
        timestamps=merged_timestamps,
        positions=Positions(xyz=merged_xyz, epsg=epsg),
        rotations=merged_rotations,
        velocity_xyz=merged_velocities,
    )

average_trajectories 

average_trajectories(
    trajectories: list[Trajectory],
) -> Trajectory

Averages a list of trajectories into one trajectory.

This function ignores EPSG codes and averages trajectories.

Parameters:

  • trajectories (List[Trajectory]) –

    List of trajectories to average.

Returns:

  • Trajectory ( Trajectory ) –

    Averaged trajectory.

Raises:

  • ValueError –

    If trajectories have different lengths or no trajectories provided.

Source code in trajectopy\processing\merging.py 
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
def average_trajectories(trajectories: list[Trajectory]) -> Trajectory:
    """Averages a list of trajectories into one trajectory.

    This function ignores EPSG codes and averages trajectories.

    Args:
        trajectories (List[Trajectory]): List of trajectories to average.

    Returns:
        Trajectory: Averaged trajectory.

    Raises:
        ValueError: If trajectories have different lengths or no trajectories provided.
    """
    if not trajectories:
        raise ValueError("Cannot average empty list of trajectories")

    lengths = {len(t) for t in trajectories}

    if len(lengths) != 1:
        raise ValueError(
            "Trajectories must have the same length to be averaged. Ensure this using trajectory matching."
        )

    xyz_stack = np.stack([t.positions.xyz for t in trajectories], axis=0)

    rots = [t.rotations.as_quat() for t in trajectories if t.rotations is not None]
    mean_rots = []
    for i in range(len(trajectories[0])):
        rotations = [rot[i] for rot in rots]
        if not rotations:
            continue

        mean_rots.append(Rotations.from_quat(np.array(rotations)).mean().as_quat())

    avg_xyz = np.mean(xyz_stack, axis=0)

    return Trajectory(
        name="Averaged",
        timestamps=trajectories[0].timestamps,
        positions=Positions(xyz=avg_xyz, epsg=trajectories[0].positions.epsg),
        rotations=Rotations.from_quat(np.array(mean_rots)) if mean_rots else None,
    )