Buckets:
| import json | |
| from pathlib import Path | |
| import numpy as np | |
| ACTORS = ("left", "right", "object") | |
| class MaskErrorTrace: | |
| def __init__( | |
| self, | |
| target_masks, | |
| sim_masks, | |
| confidence, | |
| observed, | |
| timestamps, | |
| frame_indices=None, | |
| names=ACTORS, | |
| geom_ids=None, | |
| geom_mapping=None, | |
| ): | |
| self.target_masks = np.asarray(target_masks, dtype=bool) | |
| self.sim_masks = np.asarray(sim_masks, dtype=bool) | |
| self.names = tuple(str(name) for name in names) | |
| self.confidence = np.asarray(confidence, dtype=float) | |
| self.observed = np.asarray(observed, dtype=bool) | |
| self.timestamps = np.asarray(timestamps, dtype=float) | |
| if ( | |
| self.target_masks.ndim != 4 | |
| or self.sim_masks.shape != self.target_masks.shape | |
| ): | |
| raise ValueError( | |
| "Target and simulation masks must share (actors, frames, height, width)" | |
| ) | |
| actors, frames, height, width = self.target_masks.shape | |
| if actors != len(self.names) or not frames or not height or not width: | |
| raise ValueError("Trace requires named actors and nonempty masks") | |
| if len(set(self.names)) != actors: | |
| raise ValueError("Actor names must be unique") | |
| if self.confidence.shape != (actors, frames) or self.observed.shape != ( | |
| actors, | |
| frames, | |
| ): | |
| raise ValueError("Confidence and observed must have shape (actors, frames)") | |
| if not np.isfinite(self.confidence).all() or np.any(self.confidence < 0): | |
| raise ValueError("Confidence must be finite and nonnegative") | |
| if self.timestamps.shape != (frames,) or not np.isfinite(self.timestamps).all(): | |
| raise ValueError("Trace requires one finite timestamp per frame") | |
| if frames > 1 and np.any(np.diff(self.timestamps) <= 0): | |
| raise ValueError("Trace timestamps must increase") | |
| self.frame_indices = np.asarray( | |
| np.arange(frames) if frame_indices is None else frame_indices, dtype=int | |
| ) | |
| if ( | |
| self.frame_indices.shape != (frames,) | |
| or len(np.unique(self.frame_indices)) != frames | |
| ): | |
| raise ValueError("Trace requires unique source frame indices") | |
| self.geom_ids = ( | |
| None if geom_ids is None else np.asarray(geom_ids, dtype=np.int32) | |
| ) | |
| if self.geom_ids is not None and self.geom_ids.shape != (frames, height, width): | |
| raise ValueError("Geom IDs must have shape (frames, height, width)") | |
| self.geom_mapping = { | |
| str(key): value for key, value in (geom_mapping or {}).items() | |
| } | |
| self.target_actor_ids = self._actor_ids(self.target_masks, self.confidence) | |
| self.sim_actor_ids = self._actor_ids( | |
| self.sim_masks, np.ones_like(self.confidence) | |
| ) | |
| self.target_overlap = self.target_masks.sum(axis=0) > 1 | |
| self.sim_overlap = self.sim_masks.sum(axis=0) > 1 | |
| self.false_positive = self.sim_masks & ~self.target_masks | |
| self.false_negative = self.target_masks & ~self.sim_masks | |
| self.disagreement = self.target_actor_ids != self.sim_actor_ids | |
| self.weights = self.confidence * self.observed | |
| self.trusted_disagreement = np.any( | |
| (self.false_positive | self.false_negative) | |
| & (self.weights[..., None, None] > 0), | |
| axis=0, | |
| ) | |
| def _actor_ids(masks, confidence): | |
| priority = np.where(masks, confidence[..., None, None], -1.0) | |
| actor_ids = priority.argmax(axis=0).astype(np.int16) + 1 | |
| actor_ids[~masks.any(axis=0)] = 0 | |
| return actor_ids | |
| def error_scores(self): | |
| union = np.count_nonzero(self.target_masks | self.sim_masks, axis=(-2, -1)) | |
| differences = np.count_nonzero( | |
| self.false_positive | self.false_negative, axis=(-2, -1) | |
| ) | |
| per_frame = differences / np.maximum(union, 1) | |
| totals = self.weights.sum(axis=1) | |
| per_actor = (per_frame * self.weights).sum(axis=1) / np.maximum(totals, 1e-12) | |
| return per_actor, per_frame | |
| def heatmap(self): | |
| weighted = ( | |
| (self.false_positive | self.false_negative) * self.weights[..., None, None] | |
| ).sum(axis=(0, 1)) | |
| return (weighted / max(float(self.weights.sum()), 1e-12)).astype(np.float32) | |
| def summary(self, top_frames=10): | |
| per_actor, per_frame = self.error_scores() | |
| frame_weights = self.weights.sum(axis=0) | |
| frame_scores = (per_frame * self.weights).sum(axis=0) / np.maximum( | |
| frame_weights, 1e-12 | |
| ) | |
| active = self.weights.sum(axis=1) > 0 | |
| worst_actor = ( | |
| int(np.argmax(np.where(active, per_actor, -1))) if active.any() else None | |
| ) | |
| ranked_frames = np.argsort(-np.where(frame_weights > 0, frame_scores, -1)) | |
| actors = {} | |
| for index, name in enumerate(self.names): | |
| valid = self.observed[index, :, None, None] | |
| actors[name] = { | |
| "actor_id": index + 1, | |
| "weighted_error": float(per_actor[index]) if active[index] else None, | |
| "observed_frames": int(self.observed[index].sum()), | |
| "false_positive_pixels": int(self.false_positive[index].sum()), | |
| "false_negative_pixels": int(self.false_negative[index].sum()), | |
| "observed_false_positive_pixels": int( | |
| (self.false_positive[index] & valid).sum() | |
| ), | |
| "observed_false_negative_pixels": int( | |
| (self.false_negative[index] & valid).sum() | |
| ), | |
| } | |
| return { | |
| "actors": actors, | |
| "actor_id_mapping": { | |
| "0": "background", | |
| **{str(i + 1): name for i, name in enumerate(self.names)}, | |
| }, | |
| "frames": len(self.timestamps), | |
| "image_size": [self.target_masks.shape[-1], self.target_masks.shape[-2]], | |
| "disagreement_pixels": int(self.disagreement.sum()), | |
| "trusted_disagreement_pixels": int(self.trusted_disagreement.sum()), | |
| "ambiguous_target_pixels": int(self.target_overlap.sum()), | |
| "ambiguous_sim_pixels": int(self.sim_overlap.sum()), | |
| "worst_actor": self.names[worst_actor] if worst_actor is not None else None, | |
| "physical_cause_identified": False, | |
| "top_frames": [ | |
| { | |
| "frame_index": int(self.frame_indices[index]), | |
| "local_frame_index": int(index), | |
| "timestamp": float(self.timestamps[index]), | |
| "weighted_error": float(frame_scores[index]), | |
| "actor_errors": { | |
| name: float(per_frame[actor, index]) | |
| if self.observed[actor, index] | |
| else None | |
| for actor, name in enumerate(self.names) | |
| }, | |
| } | |
| for index in ranked_frames[:top_frames] | |
| if frame_weights[index] > 0 | |
| ], | |
| } | |
| def query(self, frame_index, x, y): | |
| matches = np.flatnonzero(self.frame_indices == frame_index) | |
| if len(matches) != 1: | |
| raise ValueError(f"Source frame {frame_index} is absent from the trace") | |
| index = int(matches[0]) | |
| if ( | |
| not 0 <= x < self.target_masks.shape[-1] | |
| or not 0 <= y < self.target_masks.shape[-2] | |
| ): | |
| raise ValueError("Pixel coordinates are outside the trace image") | |
| target_id = int(self.target_actor_ids[index, y, x]) | |
| sim_id = int(self.sim_actor_ids[index, y, x]) | |
| actor_name = lambda actor_id: ( | |
| self.names[actor_id - 1] if actor_id else "background" | |
| ) | |
| result = { | |
| "frame_index": int(frame_index), | |
| "local_frame_index": index, | |
| "timestamp": float(self.timestamps[index]), | |
| "pixel": [int(x), int(y)], | |
| "coordinate_space": "rendered_mask", | |
| "image_size": [self.target_masks.shape[-1], self.target_masks.shape[-2]], | |
| "real_object": actor_name(target_id), | |
| "sim_object": actor_name(sim_id), | |
| "target_actor_id": target_id, | |
| "sim_actor_id": sim_id, | |
| "disagreement": bool(self.disagreement[index, y, x]), | |
| "trusted_disagreement": bool(self.trusted_disagreement[index, y, x]), | |
| "ambiguous_target": bool(self.target_overlap[index, y, x]), | |
| "ambiguous_sim": bool(self.sim_overlap[index, y, x]), | |
| "confidence": { | |
| name: float(self.confidence[actor, index]) | |
| for actor, name in enumerate(self.names) | |
| }, | |
| "observed": { | |
| name: bool(self.observed[actor, index]) | |
| for actor, name in enumerate(self.names) | |
| }, | |
| "actor_errors": { | |
| name: { | |
| "false_positive": bool(self.false_positive[actor, index, y, x]), | |
| "false_negative": bool(self.false_negative[actor, index, y, x]), | |
| } | |
| for actor, name in enumerate(self.names) | |
| }, | |
| "physical_cause_identified": False, | |
| } | |
| if self.geom_ids is not None: | |
| geom_id = int(self.geom_ids[index, y, x]) | |
| result["sim_geom_id"] = geom_id | |
| result["sim_geom"] = self.geom_mapping.get(str(geom_id)) | |
| return result | |
| def arrays(self): | |
| values = { | |
| "target_masks": self.target_masks, | |
| "sim_masks": self.sim_masks, | |
| "target_actor_ids": self.target_actor_ids, | |
| "sim_actor_ids": self.sim_actor_ids, | |
| "false_positive": self.false_positive, | |
| "false_negative": self.false_negative, | |
| "disagreement": self.disagreement, | |
| "trusted_disagreement": self.trusted_disagreement, | |
| "confidence": self.confidence, | |
| "observed": self.observed, | |
| "timestamps": self.timestamps, | |
| "frame_indices": self.frame_indices, | |
| "names": np.asarray(self.names), | |
| "geom_mapping_json": json.dumps(self.geom_mapping), | |
| } | |
| if self.geom_ids is not None: | |
| values["geom_ids"] = self.geom_ids | |
| return values | |
| def save(self, path): | |
| np.savez_compressed(Path(path), **self.arrays()) | |
| def load(cls, path): | |
| with np.load(path, allow_pickle=False) as saved: | |
| return cls( | |
| saved["target_masks"], | |
| saved["sim_masks"], | |
| saved["confidence"], | |
| saved["observed"], | |
| saved["timestamps"], | |
| saved["frame_indices"], | |
| saved["names"].tolist(), | |
| saved.get("geom_ids", None), | |
| json.loads(str(saved["geom_mapping_json"].item())), | |
| ) | |
Xet Storage Details
- Size:
- 11.1 kB
- Xet hash:
- 5cf92b92430cad5e1d3a1529c62f39319212009589a1c790168f4a7a647c733e
·
Xet efficiently stores files, intelligently splitting them into unique chunks and accelerating uploads and downloads. More info.