selfhostyourtech/reflector
2
0
Fork 0
mirror of https://github.com/Monadical-SAS/reflector.git synced 2025-12-24 14:19:05 +00:00
Code Issues Packages Projects Releases Wiki Activity

fix hatchet parallel syntax (#810)

Browse Source 
Co-authored-by: Igor Loskutov <igor.loskutoff@gmail.com>
This commit is contained in:
Igor Monadical
2025-12-23 18:45:06 -05:00
committed by GitHub
parent 7b352f465e
commit e886153ae1
2 changed files with 10 additions and 1 deletions
Download Patch File Download Diff File Expand all files Collapse all files

2
server/reflector/hatchet/workflows/topic_chunk_processing.py
View File

@@ -35,7 +35,7 @@ topic_chunk_workflow = hatchet.workflow(
name="TopicChunkProcessing", name="TopicChunkProcessing",
input_validator=TopicChunkInput, input_validator=TopicChunkInput,
concurrency=ConcurrencyExpression( concurrency=ConcurrencyExpression(
expression="true", # constant CEL expression = global limit expression="'global'", # constant string = global limit across all runs
max_runs=20, max_runs=20,
limit_strategy=ConcurrencyLimitStrategy.GROUP_ROUND_ROBIN, limit_strategy=ConcurrencyLimitStrategy.GROUP_ROUND_ROBIN,
), ),

9
server/reflector/hatchet/workflows/track_processing.py
View File

@@ -83,6 +83,15 @@ async def pad_track(input: TrackInput, ctx: Context) -> PadTrackResult:
) )
with av.open(source_url) as in_container: with av.open(source_url) as in_container:
if in_container.duration:
try:
duration = timedelta(seconds=in_container.duration // 1_000_000)
ctx.log(
f"pad_track: track {input.track_index}, duration={duration}"
)
except Exception:
ctx.log(f"pad_track: track {input.track_index}, duration=ERROR")
start_time_seconds = extract_stream_start_time_from_container( start_time_seconds = extract_stream_start_time_from_container(
in_container, input.track_index, logger=logger in_container, input.track_index, logger=logger
) )