full pipeline
This commit is contained in:
90
scripts/preprocess/convert_env_to_ply.py
Normal file
90
scripts/preprocess/convert_env_to_ply.py
Normal file
@@ -0,0 +1,90 @@
|
||||
#!/usr/bin/env python3
|
||||
import argparse
|
||||
import struct
|
||||
import sys
|
||||
from pathlib import Path
|
||||
|
||||
def convert_env_to_ply(input_path, output_path, verbose=False):
|
||||
"""
|
||||
Parses Xgrids environment.bin using standard library only.
|
||||
Format: 44 bytes per splat (11 little-endian floats).
|
||||
"""
|
||||
input_path = Path(input_path)
|
||||
output_path = Path(output_path)
|
||||
|
||||
if not input_path.exists():
|
||||
print(f"✗ Error: File '{input_path}' not found.")
|
||||
sys.exit(1)
|
||||
|
||||
# 44 bytes per point (Position x,y,z | Scale x,y,z | Rotation q1,q2,q3,q4 | Opacity)
|
||||
POINT_SIZE = 44
|
||||
|
||||
try:
|
||||
file_size = input_path.stat().st_size
|
||||
num_points = file_size // POINT_SIZE
|
||||
|
||||
if verbose:
|
||||
print("-" * 50)
|
||||
print(f"Input: {input_path}")
|
||||
print(f"Output: {output_path}")
|
||||
print(f"Size: {file_size / (1024*1024):.2f} MB")
|
||||
print(f"Points: {num_points:,}")
|
||||
print("-" * 50)
|
||||
|
||||
with open(input_path, "rb") as f_in, open(output_path, "w") as f_out:
|
||||
# 1. Write PLY Header
|
||||
f_out.write("ply\n")
|
||||
f_out.write("format ascii 1.0\n")
|
||||
f_out.write(f"element vertex {num_points}\n")
|
||||
f_out.write("property float x\n")
|
||||
f_out.write("property float y\n")
|
||||
f_out.write("property float z\n")
|
||||
f_out.write("end_header\n")
|
||||
|
||||
# 2. Process Binary in Chunks (to keep RAM usage low)
|
||||
# 10,000 points per chunk is a good balance for standard Python
|
||||
chunk_size = 10000
|
||||
points_processed = 0
|
||||
|
||||
while points_processed < num_points:
|
||||
remaining = num_points - points_processed
|
||||
batch_size = min(chunk_size, remaining)
|
||||
|
||||
# Read binary chunk
|
||||
chunk_data = f_in.read(batch_size * POINT_SIZE)
|
||||
if not chunk_data:
|
||||
break
|
||||
|
||||
# Unpack and write
|
||||
# '<3f' grabs just the first 3 floats (XYZ) and ignores the rest of the 44 bytes
|
||||
for i in range(batch_size):
|
||||
offset = i * POINT_SIZE
|
||||
# We only unpack the first 12 bytes (3 floats) of the 44-byte block
|
||||
x, y, z = struct.unpack_from("<fff", chunk_data, offset)
|
||||
f_out.write(f"{x:.6f} {y:.6f} {z:.6f}\n")
|
||||
|
||||
points_processed += batch_size
|
||||
if verbose and points_processed % 50000 == 0:
|
||||
print(f"• Progress: {points_processed:,} / {num_points:,}")
|
||||
|
||||
print(f"✓ Success! Converted {num_points:,} points.")
|
||||
|
||||
except Exception as e:
|
||||
print(f"✗ Error during conversion: {e}")
|
||||
sys.exit(1)
|
||||
|
||||
if __name__ == "__main__":
|
||||
parser = argparse.ArgumentParser(
|
||||
description="Zero-dependency CLI tool to convert Xgrids environment.bin to PLY."
|
||||
)
|
||||
|
||||
parser.add_argument("input", help="Path to environment.bin")
|
||||
parser.add_argument("output", nargs="?", help="Output .ply path")
|
||||
parser.add_argument("-v", "--verbose", action="store_true", help="Enable logging")
|
||||
|
||||
args = parser.parse_args()
|
||||
|
||||
# Default output logic
|
||||
out_path = Path(args.output) if args.output else Path(args.input).with_suffix(".ply")
|
||||
|
||||
convert_env_to_ply(args.input, out_path, args.verbose)
|
||||
Reference in New Issue
Block a user