diff --git a/plugins/coreml/src/pipeline b/plugins/coreml/src/pipeline deleted file mode 120000 index ca3760257..000000000 --- a/plugins/coreml/src/pipeline +++ /dev/null @@ -1 +0,0 @@ -../../tensorflow-lite/src/pipeline \ No newline at end of file diff --git a/plugins/coreml/src/requirements.txt b/plugins/coreml/src/requirements.txt index c94b349a2..ee3909f9a 100644 --- a/plugins/coreml/src/requirements.txt +++ b/plugins/coreml/src/requirements.txt @@ -1,10 +1,5 @@ -# plugin -Pillow>=5.4.1 -PyGObject>=3.30.4 coremltools~=6.1 -av>=10.0.0; sys_platform != 'linux' or platform_machine == 'x86_64' or platform_machine == 'aarch64' -# sort_oh -scipy -filterpy -numpy +# pillow for anything not intel linux, pillow-simd is available on x64 linux +Pillow>=5.4.1; sys_platform != 'linux' or platform_machine != 'x86_64' +pillow-simd; sys_platform == 'linux' and platform_machine == 'x86_64' diff --git a/plugins/python-codecs/src/thread.py b/plugins/python-codecs/src/thread.py index 4bded8bf0..059224f4c 100644 --- a/plugins/python-codecs/src/thread.py +++ b/plugins/python-codecs/src/thread.py @@ -1,5 +1,4 @@ import asyncio -from typing import Any import concurrent.futures # vips is already multithreaded, but needs to be kicked off the python asyncio thread. diff --git a/plugins/python-codecs/src/vipsimage.py b/plugins/python-codecs/src/vipsimage.py index bddfc8569..475f47304 100644 --- a/plugins/python-codecs/src/vipsimage.py +++ b/plugins/python-codecs/src/vipsimage.py @@ -6,7 +6,6 @@ try: except: Image = None pyvips = None - pass from thread import to_thread class VipsImage(scrypted_sdk.VideoFrame): diff --git a/plugins/tensorflow-lite/src/detect/__init__.py b/plugins/tensorflow-lite/src/detect/__init__.py index 0fec70736..45fe4eb27 100644 --- a/plugins/tensorflow-lite/src/detect/__init__.py +++ b/plugins/tensorflow-lite/src/detect/__init__.py @@ -1,118 +1,21 @@ from __future__ import annotations -from asyncio.events import AbstractEventLoop, TimerHandle -from asyncio.futures import Future -from typing import Any, Mapping, Tuple -from typing_extensions import TypedDict - -from pipeline import GstPipeline, GstPipelineBase, create_pipeline_sink, safe_set_result -import scrypted_sdk -import json import asyncio -import time -import os -import binascii -from urllib.parse import urlparse -import threading -from pipeline import run_pipeline -import platform -from .corohelper import run_coro_threadsafe -from PIL import Image -import math -import io +from typing import Any, Tuple -Gst = None -try: - from gi.repository import Gst -except: - pass - -av = None -try: - import av - av.logging.set_level(av.logging.PANIC) -except: - pass - -from scrypted_sdk.types import ObjectDetectionGeneratorSession, ObjectDetectionModel, Setting, FFmpegInput, MediaObject, ObjectDetection, ObjectDetectionCallbacks, ObjectDetectionSession, ObjectsDetected, ScryptedInterface, ScryptedMimeTypes - -def optional_chain(root, *keys): - result = root - for k in keys: - if isinstance(result, dict): - result = result.get(k, None) - else: - result = getattr(result, k, None) - if result is None: - break - return result - - -class DetectionSession: - id: str - timerHandle: TimerHandle - future: Future - loop: AbstractEventLoop - settings: Any - running: bool - plugin: DetectPlugin - callbacks: ObjectDetectionCallbacks - user_callback: Any - - def __init__(self) -> None: - self.timerHandle = None - self.future = Future() - self.running = False - self.mutex = threading.Lock() - self.last_sample = time.time() - self.user_callback = None - - def clearTimeoutLocked(self): - if self.timerHandle: - self.timerHandle.cancel() - self.timerHandle = None - - def clearTimeout(self): - with self.mutex: - self.clearTimeoutLocked() - - def timedOut(self): - self.plugin.end_session(self) - - def setTimeout(self, duration: float): - with self.mutex: - self.clearTimeoutLocked() - self.timerHandle = self.loop.call_later( - duration, lambda: self.timedOut()) - - -class DetectionSink(TypedDict): - pipeline: str - input_size: Tuple[float, float] +import scrypted_sdk +from scrypted_sdk.types import (MediaObject, ObjectDetection, + ObjectDetectionCallbacks, + ObjectDetectionGeneratorSession, + ObjectDetectionModel, ObjectDetectionSession, + ObjectsDetected, ScryptedMimeTypes, Setting) class DetectPlugin(scrypted_sdk.ScryptedDeviceBase, ObjectDetection): def __init__(self, nativeId: str | None = None): super().__init__(nativeId=nativeId) - self.detection_sessions: Mapping[str, DetectionSession] = {} - self.session_mutex = threading.Lock() - self.crop = False self.loop = asyncio.get_event_loop() - async def getSettings(self) -> list[Setting]: - activeSessions: Setting = { - 'key': 'activeSessions', - 'readonly': True, - 'title': 'Active Detection Sessions', - 'value': len(self.detection_sessions), - } - return [ - activeSessions - ] - - async def putSetting(self, key: str, value: scrypted_sdk.SettingValue) -> None: - pass - def getClasses(self) -> list[str]: pass @@ -138,165 +41,21 @@ class DetectPlugin(scrypted_sdk.ScryptedDeviceBase, ObjectDetection): 'settings': [], } - decoderSetting: Setting = { - 'title': "Decoder", - 'description': "The tool used to decode the stream. The may be libav or a gstreamer element.", - 'combobox': True, - 'value': 'Default', - 'placeholder': 'Default', - 'key': 'decoder', - 'subgroup': 'Advanced', - 'choices': [ - 'Default', - 'libav', - 'decodebin', - 'vtdec_hw', - 'nvh264dec', - 'vaapih264dec', - ], - } - d['settings'] += self.getModelSettings(settings) - d['settings'].append(decoderSetting) return d - async def detection_event(self, detection_session: DetectionSession, detection_result: ObjectsDetected, redetect: Any = None, mediaObject = None): - if not detection_session.running and detection_result.get('running'): - return - - detection_result['timestamp'] = int(time.time() * 1000) - if detection_session.callbacks: - if detection_session.running: - return await detection_session.callbacks.onDetection(detection_result, redetect, mediaObject) - else: - await detection_session.callbacks.onDetectionEnded(detection_result) - else: - # legacy path, nuke this pattern in opencv, pam diff, and full tensorflow. - detection_result['detectionId'] = detection_session.id - await self.onDeviceEvent(ScryptedInterface.ObjectDetection.value, detection_result) - - def end_session(self, detection_session: DetectionSession): - print('detection ended', detection_session.id) - - detection_session.clearTimeout() - # leave detection_session.running as True to avoid race conditions. - # the removal from detection_sessions will restart it. - safe_set_result(detection_session.loop, detection_session.future) - with self.session_mutex: - self.detection_sessions.pop(detection_session.id, None) - - detection_result: ObjectsDetected = {} - detection_result['running'] = False - - asyncio.run_coroutine_threadsafe(self.detection_event(detection_session, detection_result), loop=detection_session.loop) - - def create_detection_result_status(self, detection_id: str, running: bool): - detection_result: ObjectsDetected = {} - detection_result['detectionId'] = detection_id - detection_result['running'] = running - detection_result['timestamp'] = int(time.time() * 1000) - return detection_result - - def run_detection_jpeg(self, detection_session: DetectionSession, image_bytes: bytes, settings: Any) -> ObjectsDetected: - pass - def get_detection_input_size(self, src_size): pass - def create_detection_session(self): - return DetectionSession() - - def run_detection_gstsample(self, detection_session: DetectionSession, gst_sample, settings: Any, src_size, convert_to_src_size) -> Tuple[ObjectsDetected, Any]: + async def run_detection_videoframe(self, videoFrame: scrypted_sdk.VideoFrame, detection_session: ObjectDetectionSession) -> ObjectsDetected: pass - - async def run_detection_videoframe(self, videoFrame: scrypted_sdk.VideoFrame, detection_session: DetectionSession) -> ObjectsDetected: - pass - - async def run_detection_avframe(self, detection_session: DetectionSession, avframe, settings: Any, src_size, convert_to_src_size) -> Tuple[ObjectsDetected, Any]: - pil: Image.Image = avframe.to_image() - return await self.run_detection_image(detection_session, pil, settings, src_size, convert_to_src_size) - - async def run_detection_image(self, detection_session: DetectionSession, image: Image.Image, settings: Any, src_size, convert_to_src_size) -> Tuple[ObjectsDetected, Any]: - pass - - def run_detection_crop(self, detection_session: DetectionSession, sample: Any, settings: Any, src_size, convert_to_src_size, bounding_box: Tuple[float, float, float, float]) -> ObjectsDetected: - print("not implemented") - pass - - def ensure_session(self, mediaObjectMimeType: str, session: ObjectDetectionSession) -> Tuple[bool, DetectionSession, ObjectsDetected]: - settings = None - duration = None - detection_id = None - detection_session = None - - if session: - detection_id = session.get('detectionId', None) - duration = session.get('duration', None) - settings = session.get('settings', None) - - is_image = mediaObjectMimeType and mediaObjectMimeType.startswith( - 'image/') - - ending = False - new_session = False - with self.session_mutex: - if not is_image and not detection_id: - detection_id = binascii.b2a_hex(os.urandom(15)).decode('utf8') - - if detection_id: - detection_session = self.detection_sessions.get( - detection_id, None) - - if duration == None and not is_image: - ending = True - elif detection_id and not detection_session: - if not mediaObjectMimeType: - return (False, None, self.create_detection_result_status(detection_id, False)) - - new_session = True - detection_session = self.create_detection_session() - detection_session.plugin = self - detection_session.id = detection_id - detection_session.settings = settings - loop = asyncio.get_event_loop() - detection_session.loop = loop - self.detection_sessions[detection_id] = detection_session - - detection_session.future.add_done_callback( - lambda _: self.end_session(detection_session)) - - if not ending and detection_session and time.time() - detection_session.last_sample > 30 and not mediaObjectMimeType: - print('detection session has not received a sample in 30 seconds, terminating', - detection_session.id) - ending = True - - if ending: - if detection_session: - self.end_session(detection_session) - return (False, None, self.create_detection_result_status(detection_id, False)) - - if is_image: - return (False, detection_session, None) - - detection_session.setTimeout(duration / 1000) - if settings != None: - detection_session.settings = settings - - if not new_session: - print("existing session", detection_session.id) - return (False, detection_session, self.create_detection_result_status(detection_id, detection_session.running)) - - return (True, detection_session, None) async def generateObjectDetections(self, videoFrames: Any, session: ObjectDetectionGeneratorSession = None) -> Any: try: videoFrames = await scrypted_sdk.sdk.connectRPCObject(videoFrames) - detection_session = self.create_detection_session() - detection_session.plugin = self - detection_session.settings = session and session.get('settings') async for videoFrame in videoFrames: - detected = await self.run_detection_videoframe(videoFrame, detection_session) + detected = await self.run_detection_videoframe(videoFrame, session) yield { '__json_copy_serialize_children': True, 'detected': detected, @@ -309,261 +68,13 @@ class DetectPlugin(scrypted_sdk.ScryptedDeviceBase, ObjectDetection): pass async def detectObjects(self, mediaObject: MediaObject, session: ObjectDetectionSession = None, callbacks: ObjectDetectionCallbacks = None) -> ObjectsDetected: - is_image = mediaObject and (mediaObject.mimeType.startswith('image/') or mediaObject.mimeType.endswith('/x-raw-image')) - - settings = None - duration = None - - if session: - duration = session.get('duration', None) - settings = session.get('settings', None) - + vf: scrypted_sdk.VideoFrame if mediaObject and mediaObject.mimeType == ScryptedMimeTypes.Image.value: - vf: scrypted_sdk.VideoFrame = mediaObject - return await self.run_detection_videoframe(vf, settings) + vf = mediaObject + else: + vf = await scrypted_sdk.mediaManager.convertMediaObjectToBuffer(mediaObject, ScryptedMimeTypes.Image.value) - create, detection_session, objects_detected = self.ensure_session( - mediaObject and mediaObject.mimeType, session) - if detection_session: - detection_session.callbacks = callbacks - - if is_image: - stream = io.BytesIO(bytes(await scrypted_sdk.mediaManager.convertMediaObjectToBuffer(mediaObject, 'image/jpeg'))) - image = Image.open(stream) - if detection_session: - if not detection_session.user_callback: - detection_session.user_callback = self.create_user_callback(self.run_detection_image, detection_session, duration) - def convert_to_src_size(point, normalize = False): - x, y = point - return (int(math.ceil(x)), int(math.ceil(y)), True) - - detection_session.running = True - try: - return await detection_session.user_callback(image, image.size, convert_to_src_size) - finally: - detection_session.running = False - else: - return await self.run_detection_jpeg(detection_session, bytes(await scrypted_sdk.mediaManager.convertMediaObjectToBuffer(mediaObject, 'image/jpeg')), settings) - - if not create: - # a detection session may have been created, but not started - # if the initial request was for an image. - # however, attached sessions should be unchoked, as the pipeline - # is not managed here. - if not detection_session or detection_session.running or not mediaObject: - return objects_detected - - detection_id = detection_session.id - detection_session.running = True - - print('detection starting', detection_id) - b = await scrypted_sdk.mediaManager.convertMediaObjectToBuffer(mediaObject, ScryptedMimeTypes.FFmpegInput.value) - s = b.decode('utf8') - j: FFmpegInput = json.loads(s) - - container = j.get('container', None) - videosrc = j['url'] - videoCodec = optional_chain(j, 'mediaStreamOptions', 'video', 'codec') - - decoder = settings and settings.get('decoder') - if decoder == 'Default': - decoder = None - if decoder == 'libav' and not av: - decoder = None - elif decoder != 'libav' and not Gst: - decoder = None - - if not decoder: - if Gst: - if videoCodec == 'h264': - # hw acceleration is "safe" to use on mac, but not - # on other hosts where it may crash. - # defaults must be safe. - if platform.system() == 'Darwin': - decoder = 'vtdec_hw' - else: - decoder = 'avdec_h264' - else: - # decodebin may pick a hardware accelerated decoder, which isn't ideal - # so use a known software decoder for h264 and decodebin for anything else. - decoder = 'decodebin' - elif av: - decoder = 'libav' - - if decoder == 'libav': - user_callback = self.create_user_callback(self.run_detection_avframe, detection_session, duration) - - async def inference_loop(): - options = { - 'analyzeduration': '0', - 'probesize': '500000', - 'reorder_queue_size': '0', - } - container = av.open(videosrc, options = options) - stream = container.streams.video[0] - - start = 0 - for idx, frame in enumerate(container.decode(stream)): - if detection_session.future.done(): - container.close() - break - now = time.time() - if not start: - start = now - elapsed = now - start - if (frame.time or 0) < elapsed - 0.500: - # print('too slow, skipping frame') - continue - # print(frame) - size = (frame.width, frame.height) - def convert_to_src_size(point, normalize = False): - x, y = point - return (int(math.ceil(x)), int(math.ceil(y)), True) - await user_callback(frame, size, convert_to_src_size) - - def thread_main(): - loop = asyncio.new_event_loop() - loop.run_until_complete(inference_loop()) - - thread = threading.Thread(target=thread_main) - thread.start() - return self.create_detection_result_status(detection_id, True) - - if not Gst: - raise Exception('Gstreamer is unavailable') - - if videosrc.startswith('tcp://'): - parsed_url = urlparse(videosrc) - videosrc = 'tcpclientsrc port=%s host=%s' % ( - parsed_url.port, parsed_url.hostname) - if container == 'mpegts': - videosrc += ' ! tsdemux' - elif container == 'sdp': - videosrc += ' ! sdpdemux' - else: - raise Exception('unknown container %s' % container) - elif videosrc.startswith('rtsp'): - videosrc = 'rtspsrc buffer-mode=0 location=%s protocols=tcp latency=0 is-live=false' % videosrc - if videoCodec == 'h264': - videosrc += ' ! rtph264depay ! h264parse' - - videosrc += " ! %s" % decoder - - width = optional_chain(j, 'mediaStreamOptions', - 'video', 'width') or 1920 - height = optional_chain(j, 'mediaStreamOptions', - 'video', 'height') or 1080 - src_size = (width, height) - - self.run_pipeline(detection_session, duration, src_size, videosrc) - - return self.create_detection_result_status(detection_id, True) + return await self.run_detection_videoframe(vf, session) def get_pixel_format(self): return 'RGB' - - def create_pipeline_sink(self, src_size) -> DetectionSink: - inference_size = self.get_detection_input_size(src_size) - ret: DetectionSink = {} - - ret['input_size'] = inference_size - ret['pipeline'] = create_pipeline_sink( - type(self).__name__, inference_size, self.get_pixel_format()) - - return ret - - async def detection_event_notified(self, settings: Any): - pass - - async def createMedia(self, data: Any) -> MediaObject: - pass - - def invalidateMedia(self, detection_session: DetectionSession, data: Any): - pass - - def create_user_callback(self, run_detection: Any, detection_session: DetectionSession, duration: float): - first_frame = True - - current_data = None - current_src_size = None - current_convert_to_src_size = None - - async def redetect(boundingBox: Tuple[float, float, float, float]): - nonlocal current_data - nonlocal current_src_size - nonlocal current_convert_to_src_size - if not current_data: - raise Exception('no sample') - - detection_result = await self.run_detection_crop( - detection_session, current_data, detection_session.settings, current_src_size, current_convert_to_src_size, boundingBox) - - return detection_result['detections'] - - async def user_callback(sample, src_size, convert_to_src_size): - try: - detection_session.last_sample = time.time() - - nonlocal first_frame - if first_frame: - first_frame = False - print("first frame received", detection_session.id) - - detection_result, data = await run_detection( - detection_session, sample, detection_session.settings, src_size, convert_to_src_size) - if detection_result: - detection_result['running'] = True - - mo = None - retain = False - - def maybeInvalidate(): - if not retain: - self.invalidateMedia(detection_session, data) - # else: - # print('retaining') - - mo = await self.createMedia(data) - try: - nonlocal current_data - nonlocal current_src_size - nonlocal current_convert_to_src_size - try: - current_data = data - current_src_size = src_size - current_convert_to_src_size = convert_to_src_size - retain = await run_coro_threadsafe(self.detection_event(detection_session, detection_result, redetect, mo), other_loop=detection_session.loop) - finally: - current_data = None - current_convert_to_src_size = None - current_src_size = None - maybeInvalidate() - except Exception as e: - print(e) - self.invalidateMedia(detection_session, data) - - # asyncio.run_coroutine_threadsafe(, loop = self.loop).result() - await self.detection_event_notified(detection_session.settings) - - if not detection_session or duration == None: - safe_set_result(detection_session.loop, - detection_session.future) - - return detection_result - finally: - pass - - return user_callback - - def run_pipeline(self, detection_session: DetectionSession, duration, src_size, video_input): - inference_size = self.get_detection_input_size(src_size) - - pipeline = run_pipeline(detection_session.loop, detection_session.future, self.create_user_callback(self.run_detection_gstsample, detection_session, duration), - appsink_name=type(self).__name__, - appsink_size=inference_size, - video_input=video_input, - pixel_format=self.get_pixel_format(), - crop=self.crop, - ) - task = pipeline.run() - asyncio.ensure_future(task) diff --git a/plugins/tensorflow-lite/src/pipeline/__init__.py b/plugins/tensorflow-lite/src/pipeline/__init__.py deleted file mode 100644 index cbe4d1d22..000000000 --- a/plugins/tensorflow-lite/src/pipeline/__init__.py +++ /dev/null @@ -1,315 +0,0 @@ -from asyncio.events import AbstractEventLoop -from asyncio.futures import Future -import threading - -from .safe_set_result import safe_set_result -import math -import asyncio - -try: - import gi - gi.require_version('Gst', '1.0') - gi.require_version('GstBase', '1.0') - - from gi.repository import GObject, Gst - GObject.threads_init() - Gst.init(None) -except: - pass - -class GstPipelineBase: - def __init__(self, loop: AbstractEventLoop, finished: Future) -> None: - self.loop = loop - self.finished = finished - self.gst = None - - def attach_launch(self, gst): - self.gst = gst - - def parse_launch(self, pipeline: str): - self.attach_launch(Gst.parse_launch(pipeline)) - - # Set up a pipeline bus watch to catch errors. - self.bus = self.gst.get_bus() - self.watchId = self.bus.connect('message', self.on_bus_message) - self.bus.add_signal_watch() - - def on_bus_message(self, bus, message): - # seeing the following error on pi 32 bit - # OverflowError: Python int too large to convert to C long - t = str(message.type) - if t == str(Gst.MessageType.EOS): - safe_set_result(self.loop, self.finished) - elif t == str(Gst.MessageType.WARNING): - err, debug = message.parse_warning() - print('Warning: %s: %s\n' % (err, debug)) - elif t == str(Gst.MessageType.ERROR): - err, debug = message.parse_error() - print('Error: %s: %s\n' % (err, debug)) - safe_set_result(self.loop, self.finished) - return True - - async def run_attached(self): - try: - await self.finished - except: - pass - - async def attach(self): - pass - - async def detach(self): - pass - - async def run(self): - await self.attach() - - # Run pipeline. - self.gst.set_state(Gst.State.PLAYING) - - try: - await self.run_attached() - finally: - # Clean up. - self.bus.remove_signal_watch() - self.bus.disconnect(self.watchId) - self.gst.set_state(Gst.State.NULL) - self.bus = None - self.watchId = None - self.gst = None - await self.detach() - - - -class GstPipeline(GstPipelineBase): - def __init__(self, loop: AbstractEventLoop, finished: Future, appsink_name: str, user_callback, crop=False): - super().__init__(loop, finished) - self.appsink_name = appsink_name - self.user_callback = user_callback - self.running = False - self.gstsample = None - self.sink_size = None - self.src_size = None - self.dst_size = None - self.pad_size = None - self.scale_size = None - self.crop = crop - self.condition = None - - def attach_launch(self, gst): - super().attach_launch(gst) - - appsink = self.gst.get_by_name(self.appsink_name) - appsink.connect('new-preroll', self.on_new_sample, True) - appsink.connect('new-sample', self.on_new_sample, False) - - async def attach(self): - # Start inference worker. - self.running = True - worker = threading.Thread(target=self.inference_main) - worker.start() - while not self.condition: - await asyncio.sleep(.1) - - async def detach(self): - async def notifier(): - async with self.condition: - self.condition.notify_all() - self.running = False - asyncio.run_coroutine_threadsafe(notifier(), loop = self.selfLoop) - - def on_new_sample(self, sink, preroll): - sample = sink.emit('pull-preroll' if preroll else 'pull-sample') - if not self.sink_size: - s = sample.get_caps().get_structure(0) - self.sink_size = (s.get_value('width'), s.get_value('height')) - self.gstsample = sample - async def notifier(): - async with self.condition: - self.condition.notify_all() - try: - if self.running: - asyncio.run_coroutine_threadsafe(notifier(), loop = self.selfLoop).result() - except Exception as e: - # now what? - # print('sample error') - # print(e) - pass - return Gst.FlowReturn.OK - - def get_src_size(self): - if not self.src_size: - videoconvert = self.gst.get_by_name('videoconvert') - structure = videoconvert.srcpads[0].get_current_caps( - ).get_structure(0) - _, w = structure.get_int('width') - _, h = structure.get_int('height') - self.src_size = (w, h) - - videoscale = self.gst.get_by_name('videoscale') - structure = videoscale.srcpads[0].get_current_caps( - ).get_structure(0) - _, w = structure.get_int('width') - _, h = structure.get_int('height') - self.dst_size = (w, h) - - appsink = self.gst.get_by_name(self.appsink_name) - structure = appsink.sinkpads[0].get_current_caps().get_structure(0) - _, w = structure.get_int('width') - _, h = structure.get_int('height') - self.dst_size = (w, h) - - # the dimension with the higher scale value got cropped or boxed. - # use the other dimension to figure out the crop/box amount. - scales = (self.dst_size[0] / self.src_size[0], - self.dst_size[1] / self.src_size[1]) - if self.crop: - scale = max(scales[0], scales[1]) - else: - scale = min(scales[0], scales[1]) - self.scale_size = scale - - dx = self.src_size[0] * scale - dy = self.src_size[1] * scale - - px = math.ceil((self.dst_size[0] - dx) / 2) - py = math.ceil((self.dst_size[1] - dy) / 2) - - self.pad_size = (px, py) - - return self.src_size - - def convert_to_src_size(self, point, normalize=False): - valid = True - px, py = self.pad_size - x, y = point - - if normalize: - x = max(0, x) - x = min(x, self.src_size[0] - 1) - y = max(0, y) - y = min(y, self.src_size[1] - 1) - - x = (x - px) / self.scale_size - if x < 0: - x = 0 - valid = False - if x >= self.src_size[0]: - x = self.src_size[0] - 1 - valid = False - - y = (y - py) / self.scale_size - if y < 0: - y = 0 - valid = False - if y >= self.src_size[1]: - y = self.src_size[1] - 1 - valid = False - - return (int(math.ceil(x)), int(math.ceil(y)), valid) - - - def inference_main(self): - loop = asyncio.new_event_loop() - self.selfLoop = loop - try: - loop.run_until_complete(self.inference_loop()) - finally: - loop.close() - - async def inference_loop(self): - self.condition = asyncio.Condition() - while self.running: - async with self.condition: - while not self.gstsample and self.running: - await self.condition.wait() - if not self.running: - return - gstsample = self.gstsample - self.gstsample = None - try: - await self.user_callback(gstsample, self.get_src_size( - ), lambda p, normalize=False: self.convert_to_src_size(p, normalize)) - except Exception as e: - print("callback failure") - print(e) - raise - - -def get_dev_board_model(): - try: - model = open('/sys/firmware/devicetree/base/model').read().lower() - if 'mx8mq' in model: - return 'mx8mq' - if 'mt8167' in model: - return 'mt8167' - except: - pass - return None - - -def create_pipeline_sink( - appsink_name, - appsink_size, - pixel_format, - crop=False): - SINK_ELEMENT = 'appsink name={appsink_name} emit-signals=true max-buffers=-1 drop=true sync=false'.format( - appsink_name=appsink_name) - - (width, height) = appsink_size - - SINK_CAPS = 'video/x-raw,format={pixel_format}' - if width and height: - SINK_CAPS += ',width={width},height={height},pixel-aspect-ratio=1/1' - - sink_caps = SINK_CAPS.format( - width=width, height=height, pixel_format=pixel_format) - pipeline = " {sink_caps} ! {sink_element}".format( - sink_caps=sink_caps, - sink_element=SINK_ELEMENT) - - return pipeline - - -def create_pipeline( - appsink_name, - appsink_size, - video_input, - pixel_format, - crop=False, - parse_only=False): - if parse_only: - sink = 'appsink name={appsink_name} emit-signals=true sync=false'.format( - appsink_name=appsink_name) - PIPELINE = """ {video_input} - ! {sink} - """ - else: - sink = create_pipeline_sink( - appsink_name, appsink_size, pixel_format, crop=crop) - if crop: - PIPELINE = """ {video_input} ! queue leaky=downstream max-size-buffers=0 ! videoconvert name=videoconvert ! aspectratiocrop aspect-ratio=1/1 ! videoscale name=videoscale ! queue leaky=downstream max-size-buffers=0 - ! {sink} - """ - else: - PIPELINE = """ {video_input} ! queue leaky=downstream max-size-buffers=0 ! videoconvert name=videoconvert ! videoscale name=videoscale ! queue leaky=downstream max-size-buffers=0 - ! {sink} - """ - pipeline = PIPELINE.format(video_input=video_input, sink=sink) - print('Gstreamer pipeline:\n', pipeline) - return pipeline - - -def run_pipeline(loop, finished, - user_callback, - appsink_name, - appsink_size, - video_input, - pixel_format, - crop=False, - parse_only=False): - gst = GstPipeline(loop, finished, appsink_name, user_callback, crop=crop) - pipeline = create_pipeline( - appsink_name, appsink_size, video_input, pixel_format, crop=crop, parse_only=parse_only) - gst.parse_launch(pipeline) - return gst diff --git a/plugins/tensorflow-lite/src/pipeline/safe_set_result.py b/plugins/tensorflow-lite/src/pipeline/safe_set_result.py deleted file mode 100644 index 5eb33f359..000000000 --- a/plugins/tensorflow-lite/src/pipeline/safe_set_result.py +++ /dev/null @@ -1,11 +0,0 @@ -from asyncio.futures import Future -from asyncio import AbstractEventLoop - -def safe_set_result(loop: AbstractEventLoop, future: Future): - def loop_set_result(): - try: - if not future.done(): - future.set_result(None) - except: - pass - loop.call_soon_threadsafe(loop_set_result) diff --git a/plugins/tensorflow-lite/src/predict/__init__.py b/plugins/tensorflow-lite/src/predict/__init__.py index 3224c4b82..eda2c9c02 100644 --- a/plugins/tensorflow-lite/src/predict/__init__.py +++ b/plugins/tensorflow-lite/src/predict/__init__.py @@ -1,34 +1,30 @@ from __future__ import annotations -from scrypted_sdk.types import ObjectDetectionResult, ObjectsDetected, Setting -import io -from PIL import Image -import re -import scrypted_sdk -from typing import Any, List, Tuple, Mapping + import asyncio -import time -from .rectangle import Rectangle, intersect_area, intersect_rect, to_bounding_box, from_bounding_box, combine_rect -import urllib.request import os +import re +import urllib.request +from typing import Any, List, Tuple -from detect import DetectionSession, DetectPlugin +import scrypted_sdk +from PIL import Image +from scrypted_sdk.types import (ObjectDetectionResult, ObjectDetectionSession, + ObjectsDetected, Setting) -import numpy as np -import traceback +from detect import DetectPlugin -try: - from gi.repository import Gst -except: - pass +from .rectangle import (Rectangle, combine_rect, from_bounding_box, + intersect_area, intersect_rect, to_bounding_box) -class PredictSession(DetectionSession): - image: Image.Image - - def __init__(self, start_time: float) -> None: - super().__init__() - self.image = None - self.processed = 0 - self.start_time = start_time +def ensureRGBData(data: bytes, size: Tuple[int, int], format: str): + if format == 'rgba': + rgba = Image.frombuffer('RGBA', size, data) + try: + return rgba.convert('RGB') + finally: + rgba.close() + else: + return Image.frombuffer('RGB', size, data) def parse_label_contents(contents: str): lines = contents.splitlines() @@ -144,42 +140,6 @@ class PredictPlugin(DetectPlugin, scrypted_sdk.BufferConverter, scrypted_sdk.Set mo = await scrypted_sdk.mediaManager.createMediaObject(data, self.fromMimeType) return mo - def end_session(self, detection_session: PredictSession): - image = detection_session.image - if image: - detection_session.image = None - image.close() - - dps = detection_session.processed / (time.time() - detection_session.start_time) - print("Detections per second %s" % dps) - return super().end_session(detection_session) - - def invalidateMedia(self, detection_session: PredictSession, data: RawImage): - if not data: - return - image = data.image - data.image = None - if image: - if not detection_session.image: - detection_session.image = image - else: - image.close() - data.jpegMediaObject = None - - async def convert(self, data: RawImage, fromMimeType: str, toMimeType: str, options: scrypted_sdk.BufferConvertorOptions = None) -> Any: - mo = data.jpegMediaObject - if not mo: - image = data.image - if not image: - raise Exception('data is no longer valid') - - bio = io.BytesIO() - image.save(bio, format='JPEG') - jpegBytes = bio.getvalue() - mo = await scrypted_sdk.mediaManager.createMediaObject(jpegBytes, 'image/jpeg') - data.jpegMediaObject = mo - return mo - def requestRestart(self): asyncio.ensure_future(scrypted_sdk.deviceManager.requestRestart()) @@ -242,15 +202,6 @@ class PredictPlugin(DetectPlugin, scrypted_sdk.BufferConverter, scrypted_sdk.Set # print(detection_result) return detection_result - async def run_detection_jpeg(self, detection_session: PredictSession, image_bytes: bytes, settings: Any) -> ObjectsDetected: - stream = io.BytesIO(image_bytes) - image = Image.open(stream) - if image.mode == 'RGBA': - image = image.convert('RGB') - - detections, _ = await self.run_detection_image(detection_session, image, settings, image.size) - return detections - def get_detection_input_size(self, src_size): # signals to pipeline that any input size is fine # previous code used to resize to correct size and run detection that way. @@ -264,8 +215,8 @@ class PredictPlugin(DetectPlugin, scrypted_sdk.BufferConverter, scrypted_sdk.Set async def detect_once(self, input: Image.Image, settings: Any, src_size, cvss) -> ObjectsDetected: pass - async def run_detection_videoframe(self, videoFrame: scrypted_sdk.VideoFrame, detection_session: PredictSession) -> ObjectsDetected: - settings = detection_session and detection_session.settings + async def run_detection_videoframe(self, videoFrame: scrypted_sdk.VideoFrame, detection_session: ObjectDetectionSession) -> ObjectsDetected: + settings = detection_session and detection_session.get('settings') src_size = videoFrame.width, videoFrame.height w, h = self.get_input_size() iw, ih = src_size @@ -279,10 +230,7 @@ class PredictPlugin(DetectPlugin, scrypted_sdk.BufferConverter, scrypted_sdk.Set data = await videoFrame.toBuffer({ 'format': videoFrame.format or 'rgb', }) - if videoFrame.format == 'rgba': - image = Image.frombuffer('RGBA', (w, h), data).convert('RGB') - else: - image = Image.frombuffer('RGB', (w, h), data) + image = ensureRGBData(data, (w, h), videoFrame.format) try: ret = await self.detect_once(image, settings, src_size, cvss) return ret @@ -327,14 +275,8 @@ class PredictPlugin(DetectPlugin, scrypted_sdk.BufferConverter, scrypted_sdk.Set }) ) - if videoFrame.format == 'rgba': - first = Image.frombuffer('RGBA', (w, h), firstData).convert('RGB') - else: - first = Image.frombuffer('RGB', (w, h), firstData) - if videoFrame.format == 'rgba': - second = Image.frombuffer('RGBA', (w, h), secondData).convert('RGB') - else: - second = Image.frombuffer('RGB', (w, h), secondData) + first = ensureRGBData(firstData, (w, h), videoFrame.format) + second = ensureRGBData(secondData, (w, h), videoFrame.format) def cvss1(point, normalize=False): return point[0] / s, point[1] / s, True @@ -375,174 +317,3 @@ class PredictPlugin(DetectPlugin, scrypted_sdk.BufferConverter, scrypted_sdk.Set ret = ret1 ret['detections'] = dedupe_detections(ret1['detections'] + ret2['detections'], is_same_detection=is_same_detection_middle) return ret - - async def run_detection_image(self, detection_session: PredictSession, image: Image.Image, settings: Any, src_size, convert_to_src_size: Any = None, multipass_crop: Tuple[float, float, float, float] = None): - (w, h) = self.get_input_size() or image.size - (iw, ih) = image.size - - - # this a single pass or the second pass. detect once and return results. - if multipass_crop: - (l, t, dx, dy) = multipass_crop - - # find center - cx = l + dx / 2 - cy = t + dy / 2 - - # fix aspect ratio on box - if dx / w > dy / h: - dy = dx / w * h - else: - dx = dy / h * w - - if dx > image.width: - s = image.width / dx - dx = image.width - dy *= s - - if dy > image.height: - s = image.height / dy - dy = image.height - dx *= s - - # crop size to fit input size - if dx < w: - dx = w - if dy < h: - dy = h - - l = cx - dx / 2 - t = cy - dy / 2 - if l < 0: - l = 0 - if t < 0: - t = 0 - if l + dx > iw: - l = iw - dx - if t + dy > ih: - t = ih - dy - crop_box = (l, t, l + dx, t + dy) - if dx == w and dy == h: - input = image.crop(crop_box) - else: - input = image.resize((w, h), Image.ANTIALIAS, crop_box) - - def cvss(point, normalize=False): - unscaled = ((point[0] / w) * dx + l, (point[1] / h) * dy + t) - converted = convert_to_src_size(unscaled, normalize) if convert_to_src_size else (unscaled[0], unscaled[1], True) - return converted - - ret = await self.detect_once(input, settings, src_size, cvss) - input.close() - detection_session.processed = detection_session.processed + 1 - return ret, RawImage(image) - - ws = w / iw - hs = h / ih - s = max(ws, hs) - if ws == 1 and hs == 1: - def cvss(point, normalize=False): - converted = convert_to_src_size(point, normalize) if convert_to_src_size else (point[0], point[1], True) - return converted - - ret = await self.detect_once(image, settings, src_size, cvss) - if detection_session: - detection_session.processed = detection_session.processed + 1 - else: - sw = int(w / s) - sh = int(h / s) - first_crop = (0, 0, sw, sh) - first = image.resize((w, h), Image.ANTIALIAS, first_crop) - ow = iw - sw - oh = ih - sh - second_crop = (ow, oh, ow + sw, oh + sh) - second = image.resize((w, h), Image.ANTIALIAS, second_crop) - - def cvss1(point, normalize=False): - unscaled = (point[0] / s, point[1] / s) - converted = convert_to_src_size(unscaled, normalize) if convert_to_src_size else (unscaled[0], unscaled[1], True) - return converted - def cvss2(point, normalize=False): - unscaled = (point[0] / s + ow, point[1] / s + oh) - converted = convert_to_src_size(unscaled, normalize) if convert_to_src_size else (unscaled[0], unscaled[1], True) - return converted - - ret1 = await self.detect_once(first, settings, src_size, cvss1) - first.close() - if detection_session: - detection_session.processed = detection_session.processed + 1 - ret2 = await self.detect_once(second, settings, src_size, cvss2) - if detection_session: - detection_session.processed = detection_session.processed + 1 - second.close() - - two_intersect = intersect_rect(Rectangle(*first_crop), Rectangle(*second_crop)) - - def is_same_detection_middle(d1: ObjectDetectionResult, d2: ObjectDetectionResult): - same, ret = is_same_detection(d1, d2) - if same: - return same, ret - - if d1['className'] != d2['className']: - return False, None - - r1 = from_bounding_box(d1['boundingBox']) - m1 = intersect_rect(two_intersect, r1) - if not m1: - return False, None - - r2 = from_bounding_box(d2['boundingBox']) - m2 = intersect_rect(two_intersect, r2) - if not m2: - return False, None - - same, ret = is_same_box(to_bounding_box(m1), to_bounding_box(m2)) - if not same: - return False, None - c = to_bounding_box(combine_rect(r1, r2)) - return True, c - - ret = ret1 - ret['detections'] = dedupe_detections(ret1['detections'] + ret2['detections'], is_same_detection=is_same_detection_middle) - - if not len(ret['detections']): - return ret, RawImage(image) - - return ret, RawImage(image) - - async def run_detection_crop(self, detection_session: DetectionSession, sample: RawImage, settings: Any, src_size, convert_to_src_size, bounding_box: Tuple[float, float, float, float]) -> ObjectsDetected: - (ret, _) = await self.run_detection_image(detection_session, sample.image, settings, src_size, convert_to_src_size, bounding_box) - return ret - - async def run_detection_gstsample(self, detection_session: PredictSession, gstsample, settings: Any, src_size, convert_to_src_size) -> Tuple[ObjectsDetected, Image.Image]: - caps = gstsample.get_caps() - # can't trust the width value, compute the stride - height = caps.get_structure(0).get_value('height') - width = caps.get_structure(0).get_value('width') - gst_buffer = gstsample.get_buffer() - result, info = gst_buffer.map(Gst.MapFlags.READ) - if not result: - return - try: - image = detection_session.image - detection_session.image = None - - if image and (image.width != width or image.height != height): - image.close() - image = None - if image: - image.frombytes(bytes(info.data)) - else: - image = Image.frombuffer('RGB', (width, height), bytes(info.data)) - finally: - gst_buffer.unmap(info) - - try: - return await self.run_detection_image(detection_session, image, settings, src_size, convert_to_src_size) - except: - image.close() - traceback.print_exc() - raise - - def create_detection_session(self): - return PredictSession(start_time=time.time()) diff --git a/plugins/tensorflow-lite/src/predict/sort_oh b/plugins/tensorflow-lite/src/predict/sort_oh deleted file mode 120000 index afa067527..000000000 --- a/plugins/tensorflow-lite/src/predict/sort_oh +++ /dev/null @@ -1 +0,0 @@ -../../../sort-tracker/sort_oh/libs \ No newline at end of file diff --git a/plugins/tensorflow-lite/src/requirements.txt b/plugins/tensorflow-lite/src/requirements.txt index 075158305..0700565a2 100644 --- a/plugins/tensorflow-lite/src/requirements.txt +++ b/plugins/tensorflow-lite/src/requirements.txt @@ -1,10 +1,7 @@ --extra-index-url https://google-coral.github.io/py-repo/ +pycoral~=2.0 +tflite-runtime==2.5.0.post1 -# plugin -numpy>=1.16.2 -# pillow for anything not intel linux +# pillow for anything not intel linux, pillow-simd is available on x64 linux Pillow>=5.4.1; sys_platform != 'linux' or platform_machine != 'x86_64' pillow-simd; sys_platform == 'linux' and platform_machine == 'x86_64' -pycoral~=2.0 -PyGObject>=3.30.4; sys_platform != 'win32' -tflite-runtime==2.5.0.post1 diff --git a/plugins/tensorflow-lite/src/tflite/__init__.py b/plugins/tensorflow-lite/src/tflite/__init__.py index bd2f50e0d..91d755ca3 100644 --- a/plugins/tensorflow-lite/src/tflite/__init__.py +++ b/plugins/tensorflow-lite/src/tflite/__init__.py @@ -1,5 +1,4 @@ from __future__ import annotations -import threading from .common import * from PIL import Image from pycoral.adapters import detect @@ -21,7 +20,6 @@ from predict import PredictPlugin import concurrent.futures import queue import asyncio -from time import time def parse_label_contents(contents: str): lines = contents.splitlines() @@ -117,7 +115,6 @@ class TensorFlowLitePlugin(PredictPlugin, scrypted_sdk.BufferConverter, scrypted def predict(): interpreter = self.interpreters.get() try: - print('predict s %s' % time()) common.set_input( interpreter, input) scale = (1, 1) @@ -133,7 +130,6 @@ class TensorFlowLitePlugin(PredictPlugin, scrypted_sdk.BufferConverter, scrypted raise e finally: self.interpreters.put(interpreter) - print('predict e %s' % time()) objs = await asyncio.get_event_loop().run_in_executor(self.executor, predict)