基于 ① import numpy as np
from lib.test.evaluation.data import Sequence, BaseDataset, SequenceList
from lib.test.utils.load_text import load_text
import os

class COESOTDataset(BaseDataset):

def init(self, split):
super().init()
if split == ‘test’:
self.base_path = os.path.join(self.env_settings.coesot_path, split)
else:
self.base_path = os.path.join(self.env_settings.coesot_path, ‘train’)
self.sequence_list = self._get_sequence_list(split)
self.split = split

def get_sequence_list(self):
return SequenceList([self._construct_sequence(s) for s in self.sequence_list])

def _construct_sequence(self, sequence_name):
# Load annotation data
anno_path = ‘{}/{}/groundtruth.txt’.format(self.base_path, sequence_name)
ground_truth_rect = load_text(str(anno_path), delimiter=‘,’, dtype=np.float64)

# Load APS frames
frames_path = ‘{}/{}/{}’.format(self.base_path, sequence_name, sequence_name + ‘_aps’)
frame_list = [frame for frame in os.listdir(frames_path) if frame.endswith(‘.png’) or frame.endswith(‘.bmp’)]
frame_list.sort(key=lambda f: int(f[-8:-4]))
frames_list = [os.path.join(frames_path, frame) for frame in frame_list]

# Load DVS frames
event_img_path = ‘{}/{}/{}’.format(self.base_path, sequence_name, sequence_name + ‘_dvs’)
event_img_list = [frame for frame in os.listdir(event_img_path) if frame.endswith(‘.png’) or frame.endswith(‘.bmp’)]
event_img_list.sort(key=lambda f: int(f[-8:-4]))
event_img_list = [os.path.join(event_img_path, frame) for frame in event_img_list]

# Return the sequence without voxel data
return Sequence(sequence_name, frames_list, ‘coesot’, ground_truth_rect.reshape(-1, 4),
event_img_list=event_img_list) # Removed frame_event_list from the return

def len(self):
return len(self.sequence_list)

def _get_sequence_list(self, split):
with open(‘{}/list.txt’.format(self.base_path)) as f:
sequence_list = f.read().splitlines()

if split in (‘val’, ‘train’):
with open(‘{}/{}.txt’.format(self.env_settings.dataspec_path, split)) as f:
seq_ids = f.read().splitlines()
sequence_list = [sequence_list[int(x)] for x in seq_ids]

return sequence_list
② import os
import os.path
import numpy as np
import torch
import csv
import pandas
import random
from collections import OrderedDict
from .base_video_dataset import BaseVideoDataset
from lib.train.data import jpeg4py_loader
from lib.train.admin import env_settings
import scipy.io as scio

class Coesot(BaseVideoDataset):
def init(self, root=None, image_loader=jpeg4py_loader, split=None, seq_ids=None, data_fraction=None):
root = env_settings().got10k_dir if root is None else root
super().init(‘Coesot’, root, image_loader)
self.sequence_list = self._get_sequence_list()

if split is not None:
if seq_ids is not None:
raise ValueError(‘Cannot set both split_name and seq_ids.’)
if split == ‘train’:
file_path = os.path.join(self.root, ‘train.txt’)
elif split == ‘val’:
file_path = os.path.join(self.root, ‘val.txt’)
else:
raise ValueError(‘Unknown split name’)
# seq_ids = pandas.read_csv(file_path, header=None, dtype=np.int64).squeeze(“columns”).values.tolist()
seq_ids = pandas.read_csv(file_path, header=None, dtype=np.int64).squeeze(“columns”).values.tolist()
elif seq_ids is None:
seq_ids = list(range(0, len(self.sequence_list)))

self.sequence_list = [self.sequence_list[i] for i in seq_ids]

def get_name(self):
return ‘coesot’

def _get_sequence_list(self):
with open(os.path.join(self.root, ‘list.txt’)) as f:
dir_list = list(csv.reader(f))
return [dir_name[0] for dir_name in dir_list]

def _read_bb_anno(self, seq_path):
bb_anno_file = os.path.join(seq_path, ‘groundtruth.txt’)
gt = pandas.read_csv(bb_anno_file, delimiter=‘,’, header=None, dtype=np.float32, na_filter=False, low_memory=False).values
return torch.tensor(gt)

def _get_sequence_path(self, seq_id):
return os.path.join(self.root, self.sequence_list[seq_id], self.sequence_list[seq_id] + ‘_aps’)

def _get_event_img_sequence_path(self, seq_id):
return os.path.join(self.root, self.sequence_list[seq_id], self.sequence_list[seq_id] + ‘_dvs’)

def _get_grountgruth_path(self, seq_id):
return os.path.join(self.root, self.sequence_list[seq_id])

def get_sequence_info(self, seq_id):
bbox_path = self._get_grountgruth_path(seq_id)
bbox = self._read_bb_anno(bbox_path)
valid = (bbox[:, 2] > 0) & (bbox[:, 3] > 0)
visible = valid.clone().byte()
return {‘bbox’: bbox, ‘valid’: valid, ‘visible’: visible}

def _get_frame_path(self, seq_path, frame_id):
if os.path.exists(os.path.join(seq_path, ‘frame{:04}.png’.format(frame_id))):
return os.path.join(seq_path, ‘frame{:04}.png’.format(frame_id))
else:
return os.path.join(seq_path, ‘frame{:04}.bmp’.format(frame_id))

def _get_frame(self, seq_path, frame_id):
return self.image_loader(self._get_frame_path(seq_path, frame_id))

def get_frames(self, seq_id, frame_ids, anno=None):
seq_path = self._get_sequence_path(seq_id)
frame_list = [self._get_frame(seq_path, f_id) for f_id in frame_ids] # RGB_img
seq_event_path = self._get_event_img_sequence_path(seq_id)
frame_event_img_list = [self._get_frame(seq_event_path, f_id) for f_id in frame_ids] # Event_img
if anno is None:
anno = self.get_sequence_info(seq_id)

# anno_frames = {key: [value[f_id, …].clone() for f_id in frame_ids] for key, value in anno.items()}

anno_frames = {}
for key, value in anno.items():
anno_frames[key] = [value[f_id, …].clone() for f_id in frame_ids]

object_meta = OrderedDict({
‘object_class_name’: None,
‘motion_class’: None,
‘major_class’: None,
‘root_class’: None,
‘motion_adverb’: None
})

return frame_list, anno_frames, object_meta, frame_event_img_list 修改 ③ import numpy as np
from lib.test.evaluation.environment import env_settings
from lib.train.data.image_loader import imread_indexed
from collections import OrderedDict


class BaseDataset:
“”“Base class for all datasets.”“”
def init(self):
self.env_settings = env_settings()

def len(self):
“”“Overload this function in your dataset. This should return number of sequences in the dataset.”“”
raise NotImplementedError

def get_sequence_list(self):
“”“Overload this in your dataset. Should return the list of sequences in the dataset.”“”
raise NotImplementedError


class Sequence:
“”“Class for the sequence in an evaluation.”“”
def init(self, name, frames, dataset, ground_truth_rect, ground_truth_seg=None, init_data=None,
object_class=None, target_visible=None, object_ids=None, multiobj_mode=False, frame_event_list=None,
event_img_list=None):
self.name = name
self.frames = frames
self.dataset = dataset
self.ground_truth_rect = ground_truth_rect
self.ground_truth_seg = ground_truth_seg
self.object_class = object_class
self.target_visible = target_visible
self.object_ids = object_ids
self.multiobj_mode = multiobj_mode
self.init_data = self._construct_init_data(init_data)
self.event_frames = frame_event_list
self.event_img_list = event_img_list
self._ensure_start_frame()

def _ensure_start_frame(self):
# Ensure start frame is 0
start_frame = min(list(self.init_data.keys()))
if start_frame > 0:
self.frames = self.frames[start_frame:]
if self.ground_truth_rect is not None:
if isinstance(self.ground_truth_rect, (dict, OrderedDict)):
for obj_id, gt in self.ground_truth_rect.items():
self.ground_truth_rect[obj_id] = gt[start_frame:,:]
else:
self.ground_truth_rect = self.ground_truth_rect[start_frame:,:]
if self.ground_truth_seg is not None:
self.ground_truth_seg = self.ground_truth_seg[start_frame:]
assert len(self.frames) == len(self.ground_truth_seg)

if self.target_visible is not None:
self.target_visible = self.target_visible[start_frame:]
self.init_data = {frame-start_frame: val for frame, val in self.init_data.items()}

def construct_init_data(self, init_data):
if init_data is not None:
if not self.multiobj_mode:
assert self.object_ids is None or len(self.object_ids) == 1
for frame, init_val in init_data.items():
if ‘bbox’ in init_val and isinstance(init_val[‘bbox’], (dict, OrderedDict)):
init_val[‘bbox’] = init_val[‘bbox’][self.object_ids[0]]
# convert to list
for frame, init_val in init_data.items():
if ‘bbox’ in init_val:
if isinstance(init_val[‘bbox’], (dict, OrderedDict)):
init_val[‘bbox’] = OrderedDict({obj_id: list(init) for obj_id, init in init_val[‘bbox’].items()})
else:
init_val[‘bbox’] = list(init_val[‘bbox’])
else:
init_data = {0: dict()} # Assume start from frame 0

if self.object_ids is not None:
init_data[0][‘object_ids’] = self.object_ids

if self.ground_truth_rect is not None:
if self.multiobj_mode:
assert isinstance(self.ground_truth_rect, (dict, OrderedDict))
init_data[0][‘bbox’] = OrderedDict({obj_id: list(gt[0,:]) for obj_id, gt in self.ground_truth_rect.items()})
else:
assert self.object_ids is None or len(self.object_ids) == 1
if isinstance(self.ground_truth_rect, (dict, OrderedDict)):
init_data[0][‘bbox’] = list(self.ground_truth_rect[self.object_ids[0]][0, :])
else:
init_data[0][‘bbox’] = list(self.ground_truth_rect[0,:])

if self.ground_truth_seg is not None:
init_data[0][‘mask’] = self.ground_truth_seg[0]

return init_data

def init_info(self, frame_num=0):
info = self.frame_info(frame_num=frame_num)
return info

def frame_info(self, frame_num):
info = self.object_init_data(frame_num=frame_num)
return info

def init_bbox(self, frame_num=0):
return self.object_init_data(frame_num=frame_num).get(‘init_bbox’)

def init_mask(self, frame_num=0):
return self.object_init_data(frame_num=frame_num).get(‘init_mask’)

def get_info(self, keys, frame_num=None):
info = dict()
for k in keys:
val = self.get(k, frame_num=frame_num)
if val is not None:
info[k] = val
return info

def object_init_data(self, frame_num=None) -> dict:
if frame_num is None:
frame_num = 0
if frame_num not in self.init_data:
return dict()

init_data = dict()
for key, val in self.init_data[frame_num].items():
if val is None:
continue
init_data['init’+key] = val

if ‘init_mask’ in init_data and init_data[‘init_mask’] is not None:
anno = imread_indexed(init_data[‘init_mask’])
if not self.multiobj_mode and self.object_ids is not None:
assert len(self.object_ids) == 1
anno = (anno == int(self.object_ids[0])).astype(np.uint8)
init_data[‘init_mask’] = anno

if self.object_ids is not None:
init_data[‘object_ids’] = self.object_ids
init_data[‘sequence_object_ids’] = self.object_ids

return init_data

def target_class(self, frame_num=None):
return self.object_class

def get(self, name, frame_num=None):
return getattr(self, name)(frame_num)

def repr(self):
return “{self.class.name} {self.name}, length={len} frames”.format(self=self, len=len(self.frames))



class SequenceList(list):
“”“List of sequences. Supports the addition operator to concatenate sequence lists.”“”
def getitem(self, item):
if isinstance(item, str):
for seq in self:
if seq.name == item:
return seq
raise IndexError(‘Sequence name not in the dataset.’)
elif isinstance(item, int):
return super(SequenceList, self).getitem(item)
elif isinstance(item, (tuple, list)):
return SequenceList([super(SequenceList, self).getitem(i) for i in item])
else:
return SequenceList(super(SequenceList, self).getitem(item))

def add(self, other):
return SequenceList(super(SequenceList, self).add(other))

def copy(self):
return SequenceList(super(SequenceList, self).copy())