Matching Context

MatchingContext(
    aris_dir: str,
    gantry_file: str,
    gopro_file: str,
    polar_img_format: str = 'png'
)

def get_aris_frametime(frame_idx: int) -> int

def get_aris_frametime_ext(frame_idx: int) -> int

def aristime_to_gopro_idx(aris_frametime: int) -> int

def get_gopro_frame(
    aris_frametime: int,
    exact: bool = True
) -> tuple[np.ndarray | None, int]

def get_usable_gopro_range() -> tuple[int, int]

def get_gantry_odom(
    aris_frametime: int
) -> tuple[tuple[float, float, float], int]

def folder_basename(s: str) -> str

def get_aris_metadata(aris_data_dir: str) -> tuple[dict, pd.DataFrame, dict | None]

def get_gopro_metadata(gopro_files_dir: str) -> pd.DataFrame

def get_gantry_metadata(gantry_files_dir: str) -> pd.DataFrame

import cv2
import numpy as np
from scripts.common.matching_context import MatchingContext

# Initialize context
ctx = MatchingContext(
    aris_dir='/data/recordings/rec_001_aris',
    gantry_file='/data/recordings/gantry/rec_001.csv',
    gopro_file='/data/recordings/gopro/rec_001.mp4'
)

print(f"Recording: {ctx.recording_label}")
print(f"ARIS frames: {ctx.aris_frames_total}")
print(f"Active region: frames {ctx.aris_start_frame}-{ctx.aris_end_frame}")
print(f"Duration: {ctx.aris_active_duration / 1e6:.2f} seconds")

# Process synchronized data
for aris_idx in range(ctx.aris_start_frame, ctx.aris_end_frame, ctx.aris_tick_step):
    # Load ARIS frame
    aris_frame = cv2.imread(ctx.aris_frames_raw[aris_idx], cv2.IMREAD_GRAYSCALE)
    
    # Get timestamp
    aris_time = ctx.get_aris_frametime(aris_idx)
    
    # Get corresponding GoPro frame
    if ctx.has_gopro:
        gopro_frame, gopro_idx = ctx.get_gopro_frame(aris_time)
        if gopro_frame is not None:
            print(f"ARIS frame {aris_idx} synced with GoPro frame {gopro_idx}")
    
    # Get gantry position
    (x, y, z), gantry_time = ctx.get_gantry_odom(aris_time)
    print(f"Position at frame {aris_idx}: ({x:.3f}, {y:.3f}, {z:.3f}) m")

# Adjust GoPro synchronization
ctx.gopro_offset = 30  # Delay GoPro by 30 frames

# Display both streams for manual alignment
aris_idx = ctx.aris_start_frame
while True:
    aris_frame = cv2.imread(ctx.aris_frames_raw[aris_idx], cv2.IMREAD_GRAYSCALE)
    aris_time = ctx.get_aris_frametime(aris_idx)
    
    gopro_frame, _ = ctx.get_gopro_frame(aris_time)
    
    # Display side-by-side
    cv2.imshow('ARIS', aris_frame)
    if gopro_frame is not None:
        cv2.imshow('GoPro', gopro_frame)
    
    key = cv2.waitKey(30)
    if key == ord('q'):
        break
    elif key == ord('a'):  # Advance GoPro
        ctx.gopro_offset += 1
    elif key == ord('d'):  # Delay GoPro
        ctx.gopro_offset -= 1
    
    aris_idx = (aris_idx + 1) % ctx.aris_frames_total

import matplotlib.pyplot as plt

# Track object position over time
positions = []
for aris_idx in range(ctx.aris_start_frame, ctx.aris_end_frame):
    aris_time = ctx.get_aris_frametime(aris_idx)
    (x, y, z), _ = ctx.get_gantry_odom(aris_time)
    positions.append((x, y, z))

positions = np.array(positions)

# Plot trajectory
plt.figure(figsize=(10, 5))
plt.subplot(1, 2, 1)
plt.plot(positions[:, 0], positions[:, 1])
plt.xlabel('X (m)')
plt.ylabel('Y (m)')
plt.title('XY Trajectory')

plt.subplot(1, 2, 2)
plt.plot(positions[:, 2])
plt.xlabel('Frame')
plt.ylabel('Z (m)')
plt.title('Depth Profile')
plt.show()