"""Single-provider trial using the SDK's bounded linear preprocess/infer path.""" import threading import time from app.native_metadata import copy_detection_frame, copy_pose_frame from app.pipeline_diagnostics import diagnostics from app.single_decode_pipeline import capture_size, copy_single_envelope from app.stream_lifecycle import RoleRunHandle, classify_pipeline_failure, make_stream_failure # These are Gst queue defaults used by the historical Flow.preprocess path. LINEAR_QUEUE = {'max-size-buffers': 200, 'max-size-bytes': 10485760, 'max-size-time': 1000000000, 'leaky': 0} def run_provider_linear_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 if len(providers) != 1 or not getattr(coordinator, 'linear_trial', False): raise ValueError('Linear pipeline requires one provider trial coordinator.') provider = providers[0] metric = diagnostics.for_provider(provider.key) handle = RoleRunHandle('shared', run_generation, on_role_update) handle.await_source_on_stop(5) queue_node = None def active(): return not handle.stop_requested and (is_current is None or is_current()) class Ingress(BufferOperator): def handle_buffer(self, buffer): handle.note_source_ready() if not active(): return False try: selected = time.monotonic_ns() if not coordinator.ingress.stamp(buffer.batch_meta, selected_monotonic_ns=selected): return False envelope = copy_single_envelope(buffer.batch_meta, coordinator.ingress) if not coordinator.admit(envelope, selected/1e9): return False for frame in buffer.batch_meta.frame_items: key = (int(frame.frame_number), int(frame.buffer_pts)) metric.mark('selected', key, selected) if metric.collecting(): from app.input_timing import read metric.input_timing(key, read(frame), selected) 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 Stage(BufferOperator): def __init__(self, stage): super().__init__();self.stage = stage def handle_buffer(self, buffer): if metric.collecting() and queue_node is not None: metric.queue_sample({k: int(queue_node.get(k)) for k in ('current-level-buffers', 'current-level-bytes')}) if self.stage == 'preprocess': coordinator.note_request(provider.key) elif self.stage == 'infer_output': coordinator.note_provider_progress(provider.key, self.stage) for frame in buffer.batch_meta.frame_items: metric.mark(self.stage, (int(frame.frame_number), int(frame.buffer_pts))) return True class Retriever(BufferRetriever): def consume(self, buffer): coordinator.note_provider_progress(provider.key, 'callback') if not active(): return 1 try: for fm in buffer.batch_meta.frame_items: metric.mark('callback', (int(fm.frame_number), int(fm.buffer_pts))) envelope, selected = coordinator.ingress.copy(fm, with_timing=True) with metric.copy_scope(): copied = (copy_pose_frame if provider.key == 'pose' else copy_detection_frame)( fm, component_ids[provider.key]) coordinator.submit(provider.key, envelope, copied, timing=(selected/1e9,)) metric.mark('submitted', (copied.frame_number, copied.pts)) return 1 except Exception as error: handle.mark_failure(make_stream_failure('ROLE_RUNTIME_FAILED', 'native_metadata', 'shared', error)) return -1 def worker(): nonlocal queue_node from app.multi_pipeline import _pipeline_message_callback prepared = None try: if handle.stop_requested: return pipeline = Pipeline(name='linear-'+str(stream_id));handle.bind_pipeline(pipeline) if pipes is not None: pipes['shared'] = pipeline probe = Probe('linear-ingress', Ingress()) size = capture_size(providers, cfgs) if source_adapter is not None: prepared = source_adapter.prepare('shared', run_generation) capture = prepared.build_capture(Flow(pipeline), pipeline, decoded_probe=probe, rtp_time=bool(coordinator.ingress.frame_time is not None or coordinator.artifacts is not None), frame_size=size) else: capture = Flow(pipeline).batch_capture([src], **{'file-loop':False}, **({} if size is None else dict(width=size[0],height=size[1]))).attach(probe) capture = capture.attach(Probe('linear-preprocess-start', Stage('preprocess_start'))) # Same layout as Flow.preprocess: preprocess -> bounded queue -> infer. upstream = capture._streams[0].originator pp, queue = 'linear-preprocess', 'linear-queue' pipeline.add('nvdspreprocess', pp, {'config-file':str(cfgs[provider.key])+'.preprocess', 'target-unique-ids':component_ids[provider.key]}) pipeline.add('queue', queue, dict(LINEAR_QUEUE)) queue_node = pipeline[queue] pipeline.link(upstream, pp, queue) pipeline.attach(pp, Probe('linear-preprocess-output', Stage('preprocess_output'))) branch = Flow(pipeline, streams=[queue], parent=capture) branch = branch.attach(Probe('linear-infer-entry', Stage('preprocess'))) branch = branch.infer(cfgs[provider.key], with_triton=True) branch.attach(Probe('linear-infer-output', Stage('infer_output'))).retrieve(Retriever()) waiter = handle.start_pipeline(_pipeline_message_callback(handle, finite=False)) if waiter is None: return handle.mark_initialized();waiter.wait() if not handle.stop_requested: handle.mark_failure(make_stream_failure('ROLE_RUNTIME_FAILED','pipeline_exit','shared')) except Exception as error: handle.mark_failure(classify_pipeline_failure(error,'shared',handle.initialized)) finally: if prepared is not None: prepared.cleanup() thread = threading.Thread(target=worker,name='pipe-linear-'+str(stream_id),daemon=True) handle.bind_thread(thread);thread.start() return handle