60 lines
2.0 KiB
Python
60 lines
2.0 KiB
Python
import csv
|
|
import os
|
|
import time
|
|
from typing import Iterable, Any
|
|
|
|
from target_simulator.config import DEBUG_CONFIG
|
|
|
|
|
|
def _ensure_temp_folder():
|
|
temp_folder = DEBUG_CONFIG.get("temp_folder_name", "Temp")
|
|
if not os.path.exists(temp_folder):
|
|
try:
|
|
os.makedirs(temp_folder, exist_ok=True)
|
|
except Exception:
|
|
# If we cannot create the folder, swallow the exception; callers
|
|
# should handle absence of files gracefully.
|
|
return None
|
|
return temp_folder
|
|
|
|
|
|
def append_row(filename: str, row: Iterable[Any], headers: Iterable[str] | None = None):
|
|
"""Append a row to a CSV file inside the Temp folder.
|
|
|
|
If the file doesn't exist and headers are provided, write headers first.
|
|
"""
|
|
if not DEBUG_CONFIG.get("enable_io_trace", False):
|
|
return False
|
|
|
|
temp_folder = _ensure_temp_folder()
|
|
if not temp_folder:
|
|
return False
|
|
|
|
file_path = os.path.join(temp_folder, filename)
|
|
write_headers = not os.path.exists(file_path) and headers is not None
|
|
|
|
try:
|
|
with open(file_path, "a", newline="", encoding="utf-8") as csvfile:
|
|
writer = csv.writer(csvfile)
|
|
if write_headers:
|
|
writer.writerow(list(headers))
|
|
writer.writerow(list(row))
|
|
except Exception:
|
|
return False
|
|
|
|
return True
|
|
|
|
|
|
def append_sent_position(timestamp: float, target_id: int, x: float, y: float, z: float, command: str):
|
|
filename = DEBUG_CONFIG.get("io_trace_sent_filename", "sent_positions.csv")
|
|
headers = ["timestamp", "target_id", "x_ft", "y_ft", "z_ft", "command"]
|
|
row = [timestamp, target_id, x, y, z, command]
|
|
return append_row(filename, row, headers=headers)
|
|
|
|
|
|
def append_received_position(timestamp: float, target_id: int, x: float, y: float, z: float):
|
|
filename = DEBUG_CONFIG.get("io_trace_received_filename", "received_positions.csv")
|
|
headers = ["timestamp", "target_id", "x_ft", "y_ft", "z_ft"]
|
|
row = [timestamp, target_id, x, y, z]
|
|
return append_row(filename, row, headers=headers)
|