import hashlib import shutil from pathlib import Path from flandre.utils.RfMeta import RfFrameMeta, RfSequenceMeta from flandre.utils.RfSequence import RfSequence from flandre.utils.archive import to_zip def f1(): # rr = RfSequenceMeta tempdst = Path('/mnt/16T/private_dataset/ustemp') arr = [] for (i, file) in enumerate(Path('/mnt/16T/private_dataset/New Folder/steel-top/').glob('*pwi.bin')): file = Path(file) x, y, _ = file.name.split('_') r = RfFrameMeta(encoder=0, robot_x=int(x) * 100, robot_y=int(y) * 100) arr.append((file, r)) arr.sort(key=lambda item: (item[1].robot_y, item[1].robot_x)) arg = [] for i, item in enumerate(arr): file, meta = item meta.sequence_id = i print(file.name, meta.sequence_id, meta.robot_x, meta.robot_y) pic = file.with_suffix('.png') farr = [] if pic.exists(): farr.append((pic, '.png')) arg.append((file, meta, farr)) to_zip(arg, tempdst, Path('/mnt/16T/private_dataset/us/steel-top,U=30,M=PWI,S=(256 1502).zip')) def f2(): tempdst = Path('/mnt/16T/private_dataset/ustemp') src = Path('/mnt/16T/private_dataset/New Folder/T1,U=30,M=FMC,S=(256 256 1502)/') arr = [] for (i, file) in enumerate(Path('/mnt/16T/private_dataset/New Folder/steel-top/').glob('*pwi.bin')): file = Path(file) x, _ = file.name.split('_') r = RfFrameMeta(encoder=0, robot_x=int(x) * 100, robot_y=int(y) * 100) arr.append((file, r)) arr.sort(key=lambda item: (item[1].robot_y, item[1].robot_x)) print(arr) WRITE_TMP = Path('/run/media/lambda/040fb2b8-1584-4683-bac8-fec3b264167d/write_cache/') def proc(src: Path | str, dst: Path | str, fx, tmp=WRITE_TMP): src = Path(src) dst = Path(dst) assert not dst.exists() shutil.rmtree(tmp) tmp.mkdir(parents=True) arr = fx(src) print(arr) # to_zip(arr, tmp, dst) def f_ok(src: Path) -> list[tuple[Path, RfFrameMeta, list[tuple[Path, str]]]]: arr: list[tuple[Path, RfFrameMeta, list[tuple[Path, str]]]] = [] for p in src.glob('*'): m = RfFrameMeta.from_name(p.stem) arr.append((p, m, [])) arr.sort(key=lambda item: item[1].sequence_id) return arr def f_x(src: Path) -> list[tuple[Path, RfFrameMeta, list[tuple[Path, str]]]]: arr: list[tuple[Path, RfFrameMeta, list[tuple[Path, str]]]] = [] for p in src.glob('*'): m = RfFrameMeta.from_name(p.stem) arr.append((p, m, [])) arr.sort(key=lambda item: item[1].robot_x) for i, a in enumerate(arr): a[1].sequence_id = i a[1].encoder = 0 return arr def f_xy(src: Path) -> list[tuple[Path, RfFrameMeta, list[tuple[Path, str]]]]: arr: list[tuple[Path, RfFrameMeta, list[tuple[Path, str]]]] = [] for p in src.glob('*'): m = RfFrameMeta.from_name(p.stem) arr.append((p, m, [])) arr.sort(key=lambda item: (item[1].robot_y, item[1].robot_x)) for i, a in enumerate(arr): a[1].sequence_id = i a[1].encoder = 0 return arr def f_legacy4(src: Path, t, xx=100, yy=100) -> list[tuple[Path, RfFrameMeta, list[tuple[Path, str]]]]: arr = [] for p in src.glob(f'*{t}.bin'): x, y, _ = p.name.split('_') r = RfFrameMeta(encoder=0, robot_x=int(x) * xx, robot_y=int(y) * yy) arr.append((p, r)) arr.sort(key=lambda item: (item[1].robot_y, item[1].robot_x)) arg: list[tuple[Path, RfFrameMeta, list[tuple[Path, str]]]] = [] for i, item in enumerate(arr): file, meta = item meta.sequence_id = i pic = file.with_suffix('.png') farr = [] if pic.exists(): farr.append((pic, '.png')) arg.append((file, meta, farr)) return arg def f_legacy4_pwi(src: Path) -> list[tuple[Path, RfFrameMeta, list[tuple[Path, str]]]]: return f_legacy4(src, 'pwi') def f_legacy4_tfm(src: Path) -> list[tuple[Path, RfFrameMeta, list[tuple[Path, str]]]]: return f_legacy4(src, 'fmc') def f_legacy4_pwi_y10(src: Path) -> list[tuple[Path, RfFrameMeta, list[tuple[Path, str]]]]: return f_legacy4(src, 'pwi', yy=10) def f_legacy4_tfm_y10(src: Path) -> list[tuple[Path, RfFrameMeta, list[tuple[Path, str]]]]: return f_legacy4(src, 'fmc', yy=10) def f_cap(src: Path) -> list[tuple[Path, RfFrameMeta, list[tuple[Path, str]]]]: arr = [] for p in src.glob(f'*parallel.bin'): y, _ = p.name.split('_') r = RfFrameMeta(encoder=0, robot_y=int(y) * 100) arr.append((p, r)) arr.sort(key=lambda item: item[1].robot_y) arg: list[tuple[Path, RfFrameMeta, list[tuple[Path, str]]]] = [] for i, item in enumerate(arr): file, meta = item meta.sequence_id = i arg.append((file, meta, [])) return arg def f_legacy_unknown(src: Path, t) -> list[tuple[Path, RfFrameMeta, list[tuple[Path, str]]]]: arr = [] for p in src.glob(f'*{t}.bin'): x, y, _ = p.name.split('_') r = RfFrameMeta(encoder=0, robot_x=int(x) * xx, robot_y=int(y) * yy) arr.append((p, r)) arr.sort(key=lambda item: (item[1].robot_y, item[1].robot_x)) arg: list[tuple[Path, RfFrameMeta, list[tuple[Path, str]]]] = [] for i, item in enumerate(arr): file, meta = item meta.sequence_id = i pic = file.with_suffix('.png') farr = [] if pic.exists(): farr.append((pic, '.png')) arg.append((file, meta, farr)) return arg if __name__ == '__main__': b2b = hashlib.blake2b( Path('/run/media/lambda/040fb2b8-1584-4683-bac8-fec3b264167d/c1/remilia/cap1/34_4040_fmc.bin').read_bytes(), digest_size=4).hexdigest() print(b2b) print('--') for f in Path('/mnt/16T/private_dataset/us/').glob('*.zip'): m = RfSequenceMeta.from_path(f) if m.mode == RfSequenceMeta.RfSequenceMode.TFM and m.us == 30: for f in RfSequence(f).frames: print(m.commit, f.meta.sequence_id, f.meta.encoder, f.meta.robot_x, f.meta.robot_y, f.meta.blake2b) # RfSequence() # proc( # '/run/media/lambda/040fb2b8-1584-4683-bac8-fec3b264167d/c2/Desktop/New folder/4d/cap1', # '/mnt/16T/private_dataset/us/cap1,U=30,M=PWI,S=(256 1502).zip', # f_legacy_unknown, # )