"""PR13 one-source ServiceMaker composition. One RTSP capture/decode/Frame admission is forked to provider-specific preprocess/infer branches. Each bounded branch retains only its latest waiting buffer. SDK iterator wrappers must be consumed before advancing the iterator. """ from __future__ import annotations import os import configparser import threading import time from app.native_metadata import copy_detection_frame, copy_pose_frame from app.source_adapters.rtsp import RtspSourceAdapterError from app.stream_lifecycle import RoleRunHandle, classify_pipeline_failure, make_stream_failure from app.provider_branch import LATEST_QUEUE, load_completion_relay from app.pipeline_diagnostics import diagnostics def capture_size(providers, cfgs): if len(providers) != 1: return None path = str(cfgs[providers[0].key]) + '.preprocess' if not os.path.isfile(path): return None config = configparser.ConfigParser() config.read(path) size = (config.getint('property', 'processing-width'), config.getint('property', 'processing-height')) if min(size) <= 0 or max(size) > 4096: raise ValueError('Invalid capture dimensions.') return size def copy_single_envelope(batch, ingress): # ServiceMaker iterators reuse a borrowed wrapper and clear it at end. # Copy only owned values here, never list(batch.frame_items) itself. frames = [ingress.copy(frame) for frame in batch.frame_items] if len(frames) != 1: raise ValueError('Single-decode Runtime requires one Frame per batch.') return frames[0] def run_single_decode_pipeline( stream_id, providers, cfgs, component_ids, src, coordinator, *, pipes=None, run_generation=0, on_role_update=None, source_adapter=None, is_current=None): from pyservicemaker import BufferOperator, BufferRetriever, Flow, Pipeline, Probe handle = RoleRunHandle('shared', run_generation, on_role_update) handle.await_source_on_stop(5.0) gates = {} provider_diagnostics = {p.key: diagnostics.for_provider(p.key) for p in providers} def active(): return (not handle.stop_requested and (is_current is None or is_current())) class SharedIngress(BufferOperator): def handle_buffer(self, buffer): handle.note_source_ready() if not active(): return False try: selected_at = time.monotonic() if not coordinator.ingress.stamp(buffer.batch_meta): return False envelope = copy_single_envelope(buffer.batch_meta, coordinator.ingress) if not coordinator.admit(envelope, selected_at): return False if diagnostics.enabled: for frame in buffer.batch_meta.frame_items: key = (int(frame.frame_number), int(frame.buffer_pts)) for metric in provider_diagnostics.values(): metric.mark('selected', key, round(selected_at * 1_000_000_000)) handle.mark_source_stage('decoded_frame_received') handle.mark_frame() return True except Exception as error: handle.mark_failure(make_stream_failure( 'ROLE_RUNTIME_FAILED', 'frame_identity', 'shared', error)) return False class ProviderGate(BufferOperator): def __init__(self, key): super().__init__() self.key = key self._lock = threading.Lock() self._seq = None self.timing = (None, None, None) self.relay = None def handle_buffer(self, buffer): if not active() or not coordinator.provider_available(self.key): return False try: envelope = copy_single_envelope(buffer.batch_meta, coordinator.ingress) except Exception as error: handle.mark_failure(make_stream_failure( 'ROLE_RUNTIME_FAILED', 'frame_identity', 'shared', error)) return False if not coordinator.is_open(envelope.frame_seq): provider_diagnostics[self.key].mark('pre_infer_expired') return False with self._lock: if self._seq is not None: raise RuntimeError('Provider inference flight overlap.') self._seq = envelope.frame_seq entered_at = time.monotonic() selected_at = coordinator.provider_gate_time( self.key, envelope.frame_seq, entered_at) if selected_at is None: self._seq = None provider_diagnostics[self.key].mark('pre_infer_expired') return False self.timing = (selected_at, entered_at, None) return True def complete(self, seq=None): with self._lock: if seq is None or self._seq == seq: self._seq = None if self.relay is not None: self.relay.set({'complete': True}) class StageTiming(BufferOperator): def __init__(self, key, stage): super().__init__() self.metric = provider_diagnostics[key] self.key = key self.stage = stage def handle_buffer(self, buffer): if self.stage == 'infer_output': coordinator.note_provider_progress(self.key, 'infer_output') if self.metric.collecting(): for frame in buffer.batch_meta.frame_items: self.metric.mark(self.stage, (int(frame.frame_number), int(frame.buffer_pts))) return True class InferTiming(BufferOperator): def __init__(self, gate): super().__init__() self.gate = gate def handle_buffer(self, buffer): selected, entered, _ = self.gate.timing self.gate.timing = (selected, entered, time.monotonic()) for frame in buffer.batch_meta.frame_items: provider_diagnostics[self.gate.key].mark( 'preprocess', (int(frame.frame_number), int(frame.buffer_pts))) return True class ProviderRetriever(BufferRetriever): def __init__(self, provider, gate): super().__init__() self.provider = provider self.gate = gate def consume(self, buffer): callback_at = time.monotonic() coordinator.note_provider_progress(self.provider.key, 'callback') if not active(): self.gate.complete() return 1 for fm in buffer.batch_meta.frame_items: metric = provider_diagnostics[self.provider.key] metric.mark('callback', (int(fm.frame_number), int(fm.buffer_pts)), round(callback_at * 1_000_000_000)) seq = None try: try: envelope = coordinator.ingress.copy(fm) seq = envelope.frame_seq except Exception as error: handle.mark_failure(make_stream_failure( 'ROLE_RUNTIME_FAILED', 'frame_identity', 'shared', error)) return -1 if not coordinator.provider_available(self.provider.key): continue try: with metric.copy_scope(): copied = (copy_pose_frame if self.provider.key == 'pose' else copy_detection_frame)( fm, component_ids[self.provider.key]) except Exception: coordinator.mark_provider_unavailable( self.provider.key, 'provider_unavailable') continue copy_done = time.monotonic() coordinator.submit( self.provider.key, envelope, copied, copy_done, timing=(*self.gate.timing, callback_at, copy_done)) metric.mark('submitted', (copied.frame_number, copied.pts)) finally: self.gate.complete(seq) return 1 def worker(): from app.multi_pipeline import _pipeline_message_callback from app.start_diagnostics import StartDiagnostics diagnostic = StartDiagnostics('single-decode') prepared_source = None try: if handle.stop_requested: return with diagnostic.span('pipeline'): pipeline = Pipeline(name='single-' + str(stream_id)) handle.bind_pipeline(pipeline) if pipes is not None: pipes['shared'] = pipeline ingress_probe = Probe( 'shared-ingress-' + str(run_generation), SharedIngress()) frame_size = capture_size(providers, cfgs) if source_adapter is not None: with diagnostic.span('source_prepare'): prepared_source = source_adapter.prepare('shared', run_generation) capture = prepared_source.build_capture( Flow(pipeline), pipeline, decoded_probe=ingress_probe, rtp_time=bool(coordinator.ingress.frame_time is not None or coordinator.artifacts is not None), diagnostic=diagnostic, frame_size=frame_size) else: dimensions = {} if frame_size is None else dict(width=frame_size[0], height=frame_size[1]) capture = Flow(pipeline).batch_capture( [src], **{'file-loop': False}, **dimensions).attach(ingress_probe) root = capture.fork() load_completion_relay() # Flow.fork() returns a tee stream; probes must not attach to that tee. tee = root._streams[0].originator for provider in providers: if not coordinator.provider_available(provider.key): continue cfg = cfgs[provider.key] gate = ProviderGate(provider.key) gates[provider.key] = gate prefix = f'provider-{provider.key}-{run_generation}' queue, isolate, relay = prefix+'-latest', prefix+'-isolate', prefix+'-flight' pipeline.add('queue', queue, dict(LATEST_QUEUE)) # Force a branch-owned output buffer/batch meta before any mutating # preprocess/infer. This GPU copy is for isolation, not decoding. pipeline.add('nvvideoconvert', isolate, {'disable-passthrough': True, 'output-buffers': 2}) pipeline.add('contextra_completion_relay', relay) gate.relay = pipeline[relay] coordinator.register_branch_release( provider.key, lambda node=gate.relay: node.set({'disabled': True})) pipeline.link((tee, queue), ('src_%u', 'sink')) pipeline.link(queue, isolate, relay) pipeline.attach(isolate, Probe(prefix+'-admit', gate)) pipeline.attach(relay, Probe(prefix+'-preprocess-start', StageTiming(provider.key, 'preprocess_start'))) tail = relay preprocess_cfg = str(cfg) + '.preprocess' if os.path.isfile(preprocess_cfg): # Flow.preprocess() silently adds another queue. Insert the # same SDK element directly so only the latest queue owns backlog. tail = prefix+'-preprocess' pipeline.add('nvdspreprocess', tail, {'config-file': preprocess_cfg, 'target-unique-ids': component_ids[provider.key]}) pipeline.link(relay, tail) branch = Flow(pipeline, streams=[tail], parent=root) pipeline.attach(tail, Probe(prefix+'-infer-timing', InferTiming(gate))) branch = branch.infer(cfg, with_triton=True) branch = branch.attach(Probe(prefix+'-infer-output', StageTiming(provider.key, 'infer_output'))) branch.retrieve(ProviderRetriever(provider, gate)) if handle.stop_requested: return callback = _pipeline_message_callback(handle, finite=False) with diagnostic.span('sdk_start'): waiter = handle.start_pipeline(callback) if waiter is None: return handle.mark_initialized() with diagnostic.span('sdk_wait'): waiter.wait() if not handle.stop_requested: handle.mark_failure(make_stream_failure( 'ROLE_RUNTIME_FAILED', 'pipeline_exit', 'shared')) except RtspSourceAdapterError as error: handle.mark_failure(make_stream_failure( 'PIPELINE_INIT_FAILED', error.failure_stage, 'shared', error)) except Exception as error: handle.mark_failure(classify_pipeline_failure( error, 'shared', handle.initialized)) finally: diagnostic.close() if prepared_source is not None: prepared_source.cleanup() if handle.failure is None and not handle.stop_requested: handle.mark_stopped() thread = threading.Thread( target=worker, name=f'pipe-shared-{stream_id}', daemon=True) handle.bind_thread(thread) thread.start() return handle