wx44wx/icassp2027-experiment
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Fix transformers version

Browse Source
This commit is contained in:
Xin Wang
2026-08-27 15:24:59 +08:00
parent 694473b475
commit c4feeb47f4
5 changed files with 185 additions and 19 deletions
Download Patch File Download Diff File Expand all files Collapse all files

131
test_local_turn_features.py
View File

@@ -12,6 +12,7 @@ import sys
import tempfile
import unittest
from types import SimpleNamespace
from unittest.mock import patch
import numpy as np
import torch
@@ -23,6 +24,8 @@ from extract_turn_local_features import (
validate_source_timeline,
model_config_digest,
box_coordinate_offset, frame_crops, parse_args, preview_boxes,
TESTED_TRANSFORMERS_VERSIONS, validate_local_transformers_version,
main as extract_main,
)
from local_turn_features import (
LOCAL_FORMAT, causal_frame_ticks, crop_box, easycom_boxes, load_local_features,
@@ -52,6 +55,13 @@ class LocalTests(unittest.TestCase):
def setUp(self):
torch.set_num_threads(1)
def test_tested_transformers_versions(self):
for version in ("4.52.3", "4.57.6"):
self.assertEqual(validate_local_transformers_version(version), version)
for version in ("4.50.0.dev0", "4.57.6.dev0", "5.0.0"):
with self.subTest(version=version), self.assertRaisesRegex(RuntimeError, "found " + version):
validate_local_transformers_version(version)
def test_half_open_grid_and_pooling(self):
self.assertEqual(list(causal_frame_ticks(.2, 2)), [0])
self.assertEqual(list(causal_frame_ticks(1., 2)), [0, 1])
@@ -263,6 +273,67 @@ class LocalTests(unittest.TestCase):
get_image_features=lambda *_: torch.tensor([[1., 1.], [3., 3.], [5., 5.]]))
actual = encoder.images([object(), object()])
self.assertTrue(torch.equal(actual, torch.tensor([[1., 1.], [4., 4.]])))
# A future API returning per-image tensors must not silently change
# the pooling rule or crash with an uninformative AttributeError.
encoder.model.get_image_features = lambda *_: (torch.ones(1, 2), torch.ones(2, 2))
with self.assertRaisesRegex(RuntimeError, "Unexpected image encoder output"):
encoder.images([object(), object()])
def test_extraction_saves_runtime_version_and_rejects_mixed_versions(self):
with tempfile.TemporaryDirectory() as directory:
root = Path(directory)
for folder in ("features", "targets", "local"):
(root / folder).mkdir()
reference, side, labels = fixture()
torch.save(reference, root / "features/Session_4.pt")
np.save(root / "targets/Session_4.npy", labels)
media = SimpleNamespace(clips=[], metadata=lambda: {})
argv = ["extract_turn_local_features.py", "--data-root", str(root),
"--feature-root", str(root / "features"), "--target-root", str(root / "targets"),
"--output-root", str(root / "local"), "--samples", "4", "--overwrite"]
contracts = []
for version in TESTED_TRANSFORMERS_VERSIONS:
fake_encoder = SimpleNamespace(transformers_version=version, visual_dim=6, audio_dim=5,
model_config_sha256="same-checkpoint")
with patch.object(sys, "argv", argv), \
patch.dict(sys.modules, {"transformers": SimpleNamespace(__version__=version)}), \
patch("extract_turn_local_features.build_media_and_boxes", return_value=(media, [])), \
patch("extract_turn_local_features.FrozenQwenLocalEncoder", return_value=fake_encoder), \
patch("extract_turn_local_features.extract_local", return_value=copy.deepcopy(side)):
extract_main()
output = root / "local/Session_4.pt"
_, contract = load_local_features(output, reference, sample_id="Session_4",
dataset="easycom", modalities="av")
self.assertEqual(contract["config"]["transformers"], version)
contracts.append(contract)
with self.assertRaisesRegex(ValueError, "contract differs"):
load_local_features(output, reference, sample_id="Session_4", dataset="easycom",
modalities="av", expected_contract=contracts[0])
def test_unsupported_version_fails_before_media_and_preview_still_works(self):
with tempfile.TemporaryDirectory() as directory:
root = Path(directory)
reference, _, labels = fixture()
torch.save(reference, root / "Session_4.pt")
np.save(root / "Session_4.npy", labels)
argv = ["extract_turn_local_features.py", "--data-root", str(root),
"--feature-root", str(root), "--target-root", str(root),
"--output-root", str(root / "preview"), "--samples", "4"]
with patch.object(sys, "argv", argv), \
patch.dict(sys.modules, {"transformers": SimpleNamespace(__version__="5.0.0")}), \
patch("extract_turn_local_features.build_media_and_boxes") as build:
with self.assertRaisesRegex(RuntimeError, "found 5.0.0"):
extract_main()
build.assert_not_called()
media = SimpleNamespace(clips=[], metadata=lambda: {})
with patch.object(sys, "argv", argv + ["--preview-only"]), \
patch.dict(sys.modules, {"transformers": SimpleNamespace(__version__="5.0.0")}), \
patch("extract_turn_local_features.build_media_and_boxes", return_value=(media, [])), \
patch("extract_turn_local_features.preview_boxes") as preview, \
patch("extract_turn_local_features.FrozenQwenLocalEncoder") as encoder:
extract_main()
preview.assert_called_once()
encoder.assert_not_called()
def test_extraction_never_encodes_future_audio_and_saves_ragged_people(self):
class Media:
@@ -294,17 +365,19 @@ class LocalTests(unittest.TestCase):
self.assertTrue(result["audio_valid"].all())
@unittest.skipUnless(importlib.util.find_spec("transformers"), "optional: install transformers==4.52.3")
@unittest.skipUnless(importlib.util.find_spec("transformers"), "optional: install a tested Transformers version")
class TinyQwenEncoderTests(unittest.TestCase):
def test_real_encoder_apis_without_downloading_weights(self):
def setUp(self):
import transformers
if transformers.__version__ != "4.52.3":
self.skipTest("Pinned encoder API test requires transformers==4.52.3")
if transformers.__version__ not in TESTED_TRANSFORMERS_VERSIONS:
self.skipTest(f"Encoder API tests require one of {TESTED_TRANSFORMERS_VERSIONS}")
torch.set_num_threads(1)
def tiny_model_and_processor(self):
from transformers import (
Qwen2_5OmniThinkerConfig, Qwen2_5OmniThinkerForConditionalGeneration,
Qwen2VLImageProcessor, WhisperFeatureExtractor,
)
torch.set_num_threads(1)
config = Qwen2_5OmniThinkerConfig(
audio_config=dict(d_model=16, encoder_layers=1, encoder_attention_heads=2,
encoder_ffn_dim=32, output_dim=8),
@@ -313,14 +386,18 @@ class TinyQwenEncoderTests(unittest.TestCase):
text_config=dict(vocab_size=32, hidden_size=8, intermediate_size=16,
num_hidden_layers=1, num_attention_heads=2, num_key_value_heads=2),
)
model = Qwen2_5OmniThinkerForConditionalGeneration(config).eval().requires_grad_(False)
processor = SimpleNamespace(image_processor=Qwen2VLImageProcessor(),
feature_extractor=WhisperFeatureExtractor(feature_size=128))
return model, processor
def test_real_encoder_apis_without_downloading_weights(self):
encoder = FrozenQwenLocalEncoder.__new__(FrozenQwenLocalEncoder)
encoder.model = Qwen2_5OmniThinkerForConditionalGeneration(config).eval().requires_grad_(False)
encoder.model, encoder.processor = self.tiny_model_and_processor()
encoder.model.model = None
encoder.model.lm_head = None
encoder.visual_dim = encoder.audio_dim = 8
encoder.device, encoder.max_pixels, encoder.image_batch_size = torch.device("cpu"), 224 * 224, 8
encoder.processor = SimpleNamespace(image_processor=Qwen2VLImageProcessor(),
feature_extractor=WhisperFeatureExtractor(feature_size=128))
crops = [Image.new("RGB", (56, 84), "red"), Image.new("RGB", (112, 56), "blue")]
batch = encoder.images(crops)
single = encoder.images(crops[:1])
@@ -332,6 +409,44 @@ class TinyQwenEncoderTests(unittest.TestCase):
self.assertTrue(torch.isfinite(audio).all())
self.assertFalse(audio.requires_grad)
@unittest.skipUnless(importlib.util.find_spec("accelerate"), "optional: install accelerate for checkpoint loading")
def test_constructor_loads_local_checkpoint_and_preserves_encoder_weights(self):
import transformers
model, processor = self.tiny_model_and_processor()
expected = {key: value.clone() for key, value in model.state_dict().items()
if key.startswith(("visual.", "audio_tower."))}
with tempfile.TemporaryDirectory() as directory:
model.save_pretrained(directory)
args = SimpleNamespace(model_id=directory, device="cpu", dtype="float32",
flash_attention_2=False, max_pixels=224 * 224, image_batch_size=8)
with patch.object(transformers.Qwen2_5OmniProcessor, "from_pretrained", return_value=processor) as load:
encoder = FrozenQwenLocalEncoder(args)
load.assert_called_once_with(directory, use_fast=False)
self.assertEqual(encoder.transformers_version, transformers.__version__)
self.assertIsNone(encoder.model.model)
self.assertIsNone(encoder.model.lm_head)
self.assertFalse(encoder.model.training)
self.assertFalse(any(p.requires_grad for p in encoder.model.parameters()))
for key, value in encoder.model.state_dict().items():
self.assertTrue(torch.equal(value, expected[key]), key)
self.assertEqual(encoder.images([Image.new("RGB", (56, 84), "red")]).shape, (1, 8))
self.assertEqual(encoder.audio(np.zeros(3200, dtype=np.float32)).shape, (8,))
def test_constructor_rejects_incomplete_encoder_weights(self):
import transformers
args = SimpleNamespace(model_id="unused", device="cpu", dtype="float32", flash_attention_2=False)
for info in ({"missing_keys": ["visual.patch_embed.proj.weight"]},
{"missing_keys": ["audio_tower.conv1.weight"]},
{"mismatched_keys": [("visual.weight", (2,), (3,))]},
{"error_msgs": ["broken checkpoint"]}):
with self.subTest(info=info), \
patch.object(transformers.Qwen2_5OmniThinkerForConditionalGeneration,
"from_pretrained", return_value=(None, info)), \
self.assertRaisesRegex(RuntimeError, "incompletely loaded weights"):
FrozenQwenLocalEncoder(args)
class RoundTripTests(unittest.TestCase):
def test_train_infer_easycom_and_ego4d(self):