Method _decode

myriad/data_2d.py:313–429

(self, sample: dict[str, bytes])

        return target_sample

    def _decode(self, sample: dict[str, bytes]) -> dict[str, torch.Tensor]:
        try:
            assert all(
                k in sample
                for k in [
                    "video.mp4",
                    # "times.npy",
                    f"{self.track_key}.npy",
                    "logits_visible.npy",
                    "certainty.npy"
                ]
            ), f"{sample.keys()=}"

            tracks = decode_npy(sample[f"{self.track_key}.npy"])
            tracks = torch.from_numpy(tracks).float()
            tracks = (tracks + 1) / 2 # to [0, 1]
            tracks = tracks[..., [1, 0]] # (yx) to (xy)

            num_tracks_unfiltered = tracks.shape[1]
            certainty = torch.from_numpy(decode_npy(sample["certainty.npy"]))

            d = {
                "tracks": tracks,  # [t, n_t, 2]
                "visibility": torch.sigmoid(torch.from_numpy(decode_npy(sample["logits_visible.npy"]))),  # [t, n_t]
                "certainty": certainty,
            }

            # get fps and real time
            got_fps = False
            if "meta.json" in sample.keys():
                meta_bytes = sample["meta.json"]
                meta_str = meta_bytes.decode("utf-8")
                meta = json.loads(meta_str)
                if "fps" in meta.keys():
                    fps = float(meta["fps"])
                    got_fps = True
            if "fps" in sample.keys() and not got_fps:
                fps = float(sample["fps"])
                got_fps = True
            if not got_fps:
                fps = 30
            d["times"] = torch.arange(d["tracks"].shape[0]) / fps

            # get sample and valid tracks
            sample_out = self.extract_training_sample(d)
            if not sample_out.get("valid", True):
                return {"valid": False}

            # camera static filtering
            camera_static = sample_out["camera_static"]
            if self.filter_static_camera and not camera_static.item():
                return {"valid": False}

            # get frame
            i_f = sample_out.get("i_frame", 0) if not self.return_video_without_cutting else 0
            x, pos, visibility, track_in_frame = self._load_adjusted_frame(sample["video.mp4"], i_f, sample_out["pos"], sample_out["visibility"])

            if x is None: