Data Logging

Record trajectories, commands and real velocities, then save/load/plot them. Collisions (robot–robot, robot–obstacle, robot–grid) are recorded as timestamped events.

Guide

See Data Logging & Analysis for a practical walkthrough of saving/exporting a log and an example of the saved files.

data_logger

data_logger.py

DataLogger genérico para hivebr.engine.

Registra o estado de todos os robôs presentes na arena a cada frame, via draw_callback. Suporta janelas de plot interativas (Tk + Matplotlib) e exportação em pickle e csv.

Comportamento de gravação

• Gravação ativa enquanto sim.running_simulation for verdadeiro.
• Cada robô possui o seu próprio log e a sua própria sequência de segmentos.
• Um novo segmento é iniciado sempre que novos waypoints são adicionados àquele robô.
• Cada segmento é identificado pelo campo segment nos frames.
• Cada frame guarda uma flag collision (por robô). Além disso, toda colisão (robô-robô, obstáculo ou mapa) é registrada em MultiRobotLog.collision_events com o instante t e os participantes; no CSV, essas colisões vão para um arquivo *_collisions.csv.

Uso

from hivebr.engine.data_logger import DataLogger

logger = DataLogger()
sim    = Simulator(controller=ctrl)
sim.run()

logger.save("run_01.pkl")      # pickle (recommended)
logger.save_csv("run_01.csv")  # CSV (+ *_collisions.csv)
logger.export("run_01.csv")    # pick format by extension
logger.print_collisions()      # human-readable collision summary
logger.plot()                  # interactive plot window

# Carregar e plotar depois
log = DataLogger.load("run_01.pkl")
log.plot_log()

Colisões

Cada frame guarda uma flag collision por robô. Além disso, cada colisão (robô-robô, obstáculo ou mapa) é registrada uma única vez, na borda de subida, em :attr:MultiRobotLog.collision_events no formato

{"t": 0.03, "kind": "robot", "robots": [0, 1], "distance": 0.0}

com kind em {"robot", "obstacle", "grid"}, o instante t e os participantes (índices de robô, obstacle_index/obstacle_xy). No CSV, os eventos vão para um arquivo *_collisions.csv. Ver DataLogger.print_collisions e format_collision_event.

Frame dataclass

Frame(
    t,
    x,
    y,
    theta,
    vx=None,
    vy=None,
    v=None,
    omega=None,
    speed=None,
    steer=None,
    v_ref=None,
    omega_ref=None,
    vx_real=None,
    vy_real=None,
    v_real=None,
    omega_real=None,
    speed_real=None,
    steer_real=None,
    collision=False,
    goal=None,
    waypoint=None,
    wp_index=None,
    segment=0,
)

Uma amostra do simulador em um instante t.

t instance-attribute

t

x instance-attribute

x

y instance-attribute

y

theta instance-attribute

theta

vx class-attribute instance-attribute

vx = None

vy class-attribute instance-attribute

vy = None

v class-attribute instance-attribute

v = None

omega class-attribute instance-attribute

omega = None

speed class-attribute instance-attribute

speed = None

steer class-attribute instance-attribute

steer = None

v_ref class-attribute instance-attribute

v_ref = None

omega_ref class-attribute instance-attribute

omega_ref = None

vx_real class-attribute instance-attribute

vx_real = None

vy_real class-attribute instance-attribute

vy_real = None

v_real class-attribute instance-attribute

v_real = None

omega_real class-attribute instance-attribute

omega_real = None

speed_real class-attribute instance-attribute

speed_real = None

steer_real class-attribute instance-attribute

steer_real = None

collision class-attribute instance-attribute

collision = False

goal class-attribute instance-attribute

goal = None

waypoint class-attribute instance-attribute

waypoint = None

wp_index class-attribute instance-attribute

wp_index = None

segment class-attribute instance-attribute

segment = 0

SimLog dataclass

SimLog(
    robot_type,
    frames=list(),
    planned_path=None,
    planned_curve=None,
    robot_index=0,
    label="",
)

Container com todos os frames de um robô.

robot_type instance-attribute

robot_type

frames class-attribute instance-attribute

frames = field(default_factory=list)

planned_path class-attribute instance-attribute

planned_path = None

planned_curve class-attribute instance-attribute

planned_curve = None

robot_index class-attribute instance-attribute

robot_index = 0

label class-attribute instance-attribute

label = ''

to_numpy

to_numpy()

Converte frames para dicionário de arrays numpy.

Source code in src/hivebr/engine/data_logger.py

def to_numpy(self) -> dict[str, np.ndarray]:
    """Converte frames para dicionário de arrays numpy."""
    if not self.frames:
        return {}

    def _get(attr):
        return np.array([getattr(f, attr) for f in self.frames], dtype=float)

    def _get_nullable(attr):
        return np.array(
            [float(v) if (v := getattr(f, attr)) is not None else np.nan
             for f in self.frames], dtype=float,
        )

    result = {
        "t":       _get("t"),
        "x":       _get("x"),
        "y":       _get("y"),
        "theta":   _get("theta"),
        "segment": np.array([f.segment for f in self.frames], dtype=int),
        "collision": np.array([1 if f.collision else 0 for f in self.frames], dtype=int),
    }

    # ação de controle
    for cmd in _ROBOT_CMD_FIELDS.get(self.robot_type, []):
        result[cmd] = _get_nullable(cmd)

    # velocidade real
    for real_key in _ROBOT_REAL_MAP.get(self.robot_type, {}):
        result[real_key] = _get_nullable(real_key)

    # goal
    goals = [f.goal for f in self.frames]
    if any(g is not None for g in goals):
        result["goal_x"] = np.array([g[0] if g is not None else np.nan for g in goals])
        result["goal_y"] = np.array([g[1] if g is not None else np.nan for g in goals])
        if any(len(g) > 2 for g in goals if g is not None):
            result["goal_theta"] = np.array(
                [g[2] if (g is not None and len(g) > 2) else np.nan for g in goals]
            )

    # waypoint atual
    wps = [f.waypoint for f in self.frames]
    if any(w is not None for w in wps):
        result["wp_x"] = np.array([w[0] if w is not None else np.nan for w in wps])
        result["wp_y"] = np.array([w[1] if w is not None else np.nan for w in wps])

    # índice do waypoint
    wp_idx = [f.wp_index for f in self.frames]
    if any(i is not None for i in wp_idx):
        result["wp_index"] = np.array(
            [i if i is not None else -1 for i in wp_idx], dtype=int
        )

    return result

segments

segments()

Source code in src/hivebr/engine/data_logger.py

def segments(self) -> dict[int, list[Frame]]:
    segs: dict[int, list[Frame]] = {}
    for f in self.frames:
        segs.setdefault(f.segment, []).append(f)
    return segs

plot_log

plot_log()

Source code in src/hivebr/engine/data_logger.py

def plot_log(self):
    from .plot_window import show_interactive_plot
    index = getattr(self, "robot_index", 0)
    show_interactive_plot(MultiRobotLog(logs=OrderedDict([(index, self)])))

MultiRobotLog dataclass

MultiRobotLog(logs=dict(), collision_events=list())

Container com os logs de todos os robôs de uma simulação.

logs class-attribute instance-attribute

logs = field(default_factory=dict)

collision_events class-attribute instance-attribute

collision_events = field(default_factory=list)

robot_type property

robot_type

frames property

frames

planned_path property

planned_path

planned_curve property

planned_curve

to_numpy

to_numpy()

Source code in src/hivebr/engine/data_logger.py

def to_numpy(self) -> dict[int, dict[str, np.ndarray]]:
    return {i: log.to_numpy() for i, log in self.logs.items()}

segments

segments()

Source code in src/hivebr/engine/data_logger.py

def segments(self) -> dict[int, dict[int, list[Frame]]]:
    return {i: log.segments() for i, log in self.logs.items()}

plot_log

plot_log()

Source code in src/hivebr/engine/data_logger.py

def plot_log(self):
    from .plot_window import show_interactive_plot
    show_interactive_plot(self)

DataLogger

DataLogger(max_samples=None)

Registra estado + comando + velocidade real de cada robô a cada frame via draw_callback.

A gravação ocorre enquanto sim.running_simulation for verdadeiro. Cada robô tem o seu próprio :class:SimLog e a sua própria sequência de segmentos (novo segmento a cada waypoint adicionado).

Source code in src/hivebr/engine/data_logger.py

def __init__(self, max_samples: int | None = None):
    self.max_samples = max_samples
    self._t = 0.0
    self._logs: "OrderedDict[int, SimLog]" = OrderedDict()
    self._segment_idx: dict[int, int] = {}
    self._last_wp_count: dict[int, int] = {}
    self._initialized = False
    self._sim = None

max_samples instance-attribute

max_samples = max_samples

reset

reset()

Descarta frames e reseta estado de gravação.

Source code in src/hivebr/engine/data_logger.py

def reset(self):
    """Descarta frames e reseta estado de gravação."""
    self._t = 0.0
    self._logs = OrderedDict()
    self._segment_idx = {}
    self._last_wp_count = {}
    self._initialized = False
    self._sim = None
    print("[DataLogger] Logger reiniciado.")

draw_callback

draw_callback(sim, screen)

Source code in src/hivebr/engine/data_logger.py

def draw_callback(self, sim, screen):
    if not sim.running_simulation:
        return

    if not self._initialized:
        self._init(sim)

    self._t += sim.dt

    for index in range(len(sim.robots)):
        _, total_wp = self._waypoint_state(sim, index)

        # Novos waypoints -> novo segmento
        if total_wp > self._last_wp_count.get(index, 0):
            self._segment_idx[index] = self._segment_idx.get(index, -1) + 1
            print(f"[DataLogger] B-Bot{index} — segmento "
                  f"{self._segment_idx[index]} iniciado "
                  f"({len(self._logs[index].frames)} frames, "
                  f"{total_wp} waypoints).")

        self._last_wp_count[index] = total_wp
        self._record(sim, index, self._waypoint_manager(sim, index))

set_planned_path

set_planned_path(path, robot_index=0)

Registra caminho planejado de um robô como lista de (x,y[,theta]).

Source code in src/hivebr/engine/data_logger.py

def set_planned_path(self, path: list[tuple], robot_index: int = 0):
    """Registra caminho planejado de um robô como lista de (x,y[,theta])."""
    if robot_index not in self._logs:
        print(f"[DataLogger] Robô {robot_index} não inicializado.")
        return
    self._logs[robot_index].planned_path = [
        tuple(float(v) for v in p) for p in path
    ]
    print(f"[DataLogger] Caminho planejado (B-Bot{robot_index}): "
          f"{len(path)} pontos.")

set_planned_curve

set_planned_curve(t, x, y, theta=None, robot_index=0)

Registra caminho planejado de um robô como curva paramétrica.

Source code in src/hivebr/engine/data_logger.py

def set_planned_curve(self, t, x, y, theta=None, robot_index: int = 0):
    """Registra caminho planejado de um robô como curva paramétrica."""
    if robot_index not in self._logs:
        print(f"[DataLogger] Robô {robot_index} não inicializado.")
        return
    curve = {
        "t": np.array(t, dtype=float),
        "x": np.array(x, dtype=float),
        "y": np.array(y, dtype=float),
    }
    if theta is not None:
        curve["theta"] = np.array(theta, dtype=float)
    self._logs[robot_index].planned_curve = curve
    print(f"[DataLogger] Curva planejada (B-Bot{robot_index}): "
          f"{len(curve['x'])} pontos.")

result

result()

Source code in src/hivebr/engine/data_logger.py

def result(self) -> MultiRobotLog:
    events = list(getattr(self._sim, "collision_events", []) or [])
    return MultiRobotLog(logs=self._logs, collision_events=events)

collision_summary

collision_summary()

Linhas legíveis descrevendo cada colisão registrada.

Source code in src/hivebr/engine/data_logger.py

def collision_summary(self) -> list[str]:
    """Linhas legíveis descrevendo cada colisão registrada."""
    return [format_collision_event(e) for e in self.result().collision_events]

print_collisions

print_collisions()

Source code in src/hivebr/engine/data_logger.py

def print_collisions(self):
    lines = self.collision_summary()
    if not lines:
        print("[DataLogger] Nenhuma colisão registrada.")
        return
    print(f"[DataLogger] {len(lines)} colisão(ões) registrada(s):")
    for line in lines:
        print("  " + line)

save

save(path='log.pkl')

Salva o log completo (incluindo colisões) em pickle.

Source code in src/hivebr/engine/data_logger.py

def save(self, path: str = "log.pkl"):
    """Salva o log completo (incluindo colisões) em pickle."""
    if not self._has_data():
        print("[DataLogger] Nenhum dado para salvar.")
        return
    result = self.result()
    with open(path, "wb") as f:
        pickle.dump(result, f)
    total = sum(len(l.frames) for l in self._logs.values())
    print(f"[DataLogger] {total} frames / {len(self._logs)} robô(s) -> '{path}'")
    self._write_collisions(result.collision_events, path)

save_csv

save_csv(path='log.csv')

Salva o log completo em CSV (uma linha por robô por frame).

Source code in src/hivebr/engine/data_logger.py

def save_csv(self, path: str = "log.csv"):
    """Salva o log completo em CSV (uma linha por robô por frame)."""
    if not self._has_data():
        print("[DataLogger] Nenhum dado para salvar.")
        return

    columns = ["robot", "robot_type", "segment", "t", "x", "y", "theta"]
    arrays: "OrderedDict[int, dict]" = OrderedDict()
    for index in sorted(self._logs):
        arr = self._logs[index].to_numpy()
        arrays[index] = arr
        for key in arr:
            if key not in columns:
                columns.append(key)

    with open(path, "w", newline="", encoding="utf-8") as f:
        writer = csv.writer(f)
        writer.writerow(columns)
        for index, arr in arrays.items():
            log = self._logs[index]
            for row in range(len(log.frames)):
                writer.writerow([
                    index if col == "robot"
                    else log.robot_type if col == "robot_type"
                    else arr[col][row] if col in arr
                    else ""
                    for col in columns
                ])

    total = sum(len(l.frames) for l in self._logs.values())
    print(f"[DataLogger] {total} frames / {len(self._logs)} robô(s) -> '{path}'")
    self._write_collisions(self.result().collision_events, path)

export

export(path)

Exporta escolhendo o formato pela extensão (.csv ou .pkl).

Source code in src/hivebr/engine/data_logger.py

def export(self, path: str):
    """Exporta escolhendo o formato pela extensão (.csv ou .pkl)."""
    if str(path).lower().endswith(".csv"):
        self.save_csv(path)
    else:
        self.save(path)

load staticmethod

load(path)

Source code in src/hivebr/engine/data_logger.py

@staticmethod
def load(path: str):
    with open(path, "rb") as f:
        log = pickle.load(f)
    if isinstance(log, MultiRobotLog):
        total = sum(len(l.frames) for l in log.logs.values())
        print(f"[DataLogger] '{path}': {total} frames, "
              f"{len(log.logs)} robô(s)")
    else:
        print(f"[DataLogger] '{path}': {len(log.frames)} frames, "
              f"tipo={log.robot_type}")
    return log

plot

plot()

Source code in src/hivebr/engine/data_logger.py

def plot(self):
    if not self._has_data():
        print("[DataLogger] Nenhum dado para plotar.")
        return
    from .plot_window import show_interactive_plot
    show_interactive_plot(self.result())

format_collision_event

format_collision_event(event)

Formata um evento de colisão em uma linha legível.

Source code in src/hivebr/engine/data_logger.py

def format_collision_event(event: dict) -> str:
    """Formata um evento de colisão em uma linha legível."""
    t = float(event.get("t", 0.0))
    kind = event.get("kind", "?")
    robots = event.get("robots", []) or []
    if kind == "robot":
        pair = " <-> ".join(f"B-Bot{r}" for r in robots)
        dist = event.get("distance")
        dist_txt = f" (d={float(dist):.3f} m)" if dist is not None else ""
        return f"[{t:8.2f}s] COLISÃO robô-robô: {pair}{dist_txt}"
    if kind == "obstacle":
        xy = event.get("obstacle_xy")
        xy_txt = f" em ({float(xy[0]):.2f}, {float(xy[1]):.2f})" if xy else ""
        return (
            f"[{t:8.2f}s] COLISÃO B-Bot{robots[0]} com obstáculo "
            f"#{event.get('obstacle_index')}{xy_txt}"
        )
    if kind == "grid":
        return f"[{t:8.2f}s] COLISÃO B-Bot{robots[0]} com o mapa (inflado)"
    return f"[{t:8.2f}s] COLISÃO {event}"