Skip to content

Check LlamaVideoSummarization timestamps #400

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Closed
wants to merge 2 commits into from
Closed

Check LlamaVideoSummarization timestamps #400

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
2 commits
Select commit Hold shift + click to select a range
ec6153d
Validate timestamps.
jrobble May 28, 2025
14f30ad
Fix bug.
jrobble May 28, 2025
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
58 changes: 31 additions & 27 deletions python/LlamaVideoSummarization/llama_video_summarization_component/__init__.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -120,36 +120,40 @@ def _get_response_from_subprocess(self, job_config: dict) -> dict:

def _check_response(self, attempts: dict, max_attempts: int, schema_json: dict, response: str
) -> Tuple[Union[dict, None], Union[str, None]]:
error = None
response_json = None

if not response:
error = 'Empty response.'
log.warning(error)
log.warning(f'Failed {attempts["base"] + 1} of {max_attempts} base attempts.')
attempts['base'] += 1
return None, error

try:
response_json = json.loads(response)
except ValueError as ve:
error = 'Response is not valid JSON.'
if not error:
try:
response_json = json.loads(response)
except ValueError as ve:
error = f'Response is not valid JSON. {str(ve)}'

if not error and response_json:
try:
validate(response_json, schema_json)
except ValidationError as ve:
error = f'Response JSON is not in the desired format. {str(ve)}'

if not error and response_json:
try:
event_timeline = response_json['video_event_timeline']
for event in event_timeline:
# update values for later use
event["timestamp_start"] = _get_timestamp_value(event["timestamp_start"])
event["timestamp_end"] = _get_timestamp_value(event["timestamp_end"])
except ValueError as ve:
error = f'Response JSON is not in the desired format. {str(ve)}'

if error:
log.warning(error)
log.warning(str(ve))
log.warning(f'Failed {attempts["base"] + 1} of {max_attempts} base attempts.')
attempts['base'] += 1
return response_json, error

try:
validate(response_json, schema_json)
except ValidationError as ve:
error = 'Response JSON is not in the desired format.'
log.warning(error)
log.warning(str(ve))
log.warning(f'Failed {attempts["base"] + 1} of {max_attempts} base attempts.')
attempts['base'] += 1
return response_json, error

return response_json, None
return response_json, error


def _check_timeline(self, threshold: float, attempts: dict, max_attempts: int,
Expand All @@ -158,8 +162,8 @@ def _check_timeline(self, threshold: float, attempts: dict, max_attempts: int,

error = None
for event in event_timeline:
timestamp_start = _get_timestamp_value(event["timestamp_start"])
timestamp_end = _get_timestamp_value(event["timestamp_end"])
timestamp_start = event["timestamp_start"]
timestamp_end = event["timestamp_end"]

if timestamp_start < 0:
error = (f'Timeline event start time of {timestamp_start} < 0.')
Expand All @@ -185,15 +189,15 @@ def _check_timeline(self, threshold: float, attempts: dict, max_attempts: int,
break

if not error:
min_event_start = min(list(map(lambda d: _get_timestamp_value(d.get('timestamp_start')),
min_event_start = min(list(map(lambda d: d.get('timestamp_start'),
filter(lambda d: 'timestamp_start' in d, event_timeline))))

if abs(segment_start_time - min_event_start) > threshold:
error = (f'Min timeline event start time not close enough to segment start time. '
f'abs({segment_start_time} - {min_event_start}) > {threshold}.')

if not error:
max_event_end = max(list(map(lambda d: _get_timestamp_value(d.get('timestamp_end')),
max_event_end = max(list(map(lambda d: d.get('timestamp_end'),
filter(lambda d: 'timestamp_end' in d, event_timeline))))

if abs(max_event_end - segment_stop_time) > threshold:
Expand Down Expand Up @@ -263,8 +267,8 @@ def _create_tracks(self, job: mpf.VideoJob, response_json: dict) -> Iterable[mpf

for event in response_json['video_event_timeline']:
# get offset start/stop times in milliseconds
event_start_time = int(_get_timestamp_value(event['timestamp_start']) * 1000)
event_stop_time = int(_get_timestamp_value(event['timestamp_end']) * 1000)
event_start_time = int(event['timestamp_start'] * 1000)
event_stop_time = int(event['timestamp_end'] * 1000)

offset_start_frame = int((event_start_time * video_fps) / 1000)
offset_stop_frame = int((event_stop_time * video_fps) / 1000) - 1
Expand Down
56 changes: 55 additions & 1 deletion python/LlamaVideoSummarization/tests/test_llama_video_summarization.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -69,7 +69,7 @@
},
{
"timestamp_start": "5.0",
"timestamp_end": "6.8",
"timestamp_end": "6.8s",
"description": "The cat looks back at the camera and then walks away."
}
]
Expand Down Expand Up @@ -340,6 +340,59 @@ def test_invalid_json_response(self):
self.assertEqual(mpf.DetectionError.DETECTION_FAILED, cm.exception.error_code)
self.assertIn("not valid JSON", str(cm.exception))


def test_schema_check(self):
component = LlamaVideoSummarizationComponent()

job = mpf.VideoJob('cat job', str(TEST_DATA / 'cat.mp4'), 0, 171,
{
"GENERATION_MAX_ATTEMPTS" : "1"
},
CAT_VIDEO_PROPERTIES, {})

with self.assertRaises(mpf.DetectionException) as cm:
self.run_patched_job(component, job, json.dumps(
{
"video_summary": "This is a video of a cat.",
"video_event_timeline": [
{
"timestamp_start": "0.00",
"bad": "8.04",
"description": "The cat is sitting on the cobblestone street, looking around."
}
]
})) # don't care about results

self.assertEqual(mpf.DetectionError.DETECTION_FAILED, cm.exception.error_code)
self.assertIn("'timestamp_end' is a required property", str(cm.exception))


def test_invalid_timestamp(self):
component = LlamaVideoSummarizationComponent()

job = mpf.VideoJob('cat job', str(TEST_DATA / 'cat.mp4'), 0, 171,
{
"GENERATION_MAX_ATTEMPTS" : "1"
},
CAT_VIDEO_PROPERTIES, {})

with self.assertRaises(mpf.DetectionException) as cm:
self.run_patched_job(component, job, json.dumps(
{
"video_summary": "This is a video of a cat.",
"video_event_timeline": [
{
"timestamp_start": "7:12",
"timestamp_end": "8:04",
"description": "The cat is sitting on the cobblestone street, looking around."
}
]
})) # don't care about results

self.assertEqual(mpf.DetectionError.DETECTION_FAILED, cm.exception.error_code)
self.assertIn("could not convert string to float", str(cm.exception))


def test_empty_response(self):
component = LlamaVideoSummarizationComponent()

Expand All @@ -355,6 +408,7 @@ def test_empty_response(self):
self.assertEqual(mpf.DetectionError.DETECTION_FAILED, cm.exception.error_code)
self.assertIn("Empty response", str(cm.exception))


def test_timeline_integrity(self):
component = LlamaVideoSummarizationComponent()

Expand Down