运行 使用 dogwatch 生成新图像后的 TensorFlow 脚本

Question

我有一个 Raspberry Pi 4 和珊瑚加速器。

这是我正在尝试做的事情：

1. 我在 Pi 上有一个目录，我可以在其中自动上传我 phone

   中的照片

2. 我有看门狗工作，可以检测何时添加（上传）新照片

3. [我的问题领域我需要一些帮助或指导]：在 elif 中为新创建的文件（图片已上传）我想 运行 python 脚本和命令 运行 图像处理以检测图片中的内容（珊瑚加速器）

问题区域被一排星号包围。

import time
from watchdog.observers import Observer
from watchdog.events import FileSystemEventHandler

class Watcher:
    DIRECTORY_TO_WATCH = "/path/to/img/directory"

    def __init__(self):
        self.observer = Observer()

    def run(self):
        event_handler = Handler()
        self.observer.schedule(event_handler, self.DIRECTORY_TO_WATCH, recursive=True)
        self.observer.start()
        try:
            while True:
                time.sleep(5)
        except:
            self.observer.stop()
            print("Error")

        self.observer.join()


class Handler(FileSystemEventHandler):

    @staticmethod
    def on_any_event(event):
        if event.is_directory:
            return None

        elif event.event_type == 'created':
            # Take any action here when a file is first created.
            print (event.src_path)
            ******************************
            ******************************
            python3 classify_image.py \
            --model models/mobilenet_v2_1.0_224_inat_bird_quant_edgetpu.tflite \
            --labels models/inat_bird_labels.txt \
            --input event.src_path
            ******************************
            ******************************

        # elif event.event_type == 'modified':
            # Taken any action here when a file is modified.
            # print("Received modified event - %s." % event.src_path)

        elif event.event_type == 'deleted':
            # Taken any action here when a file is deleted.
            print("Received deleted event - %s." % event.src_path)


if __name__ == '__main__':
    w = Watcher()
    w.run()

提前致谢。

Answer 1

您可以使用 subprocess.run().

执行 python 脚本

import subprocess

args = [
    'python3',
    'classify_image.py',
    '--model models/mobilenet_v2_1.0_224_inat_bird_quant_edgetpu.tflite',
    '--labels models/inat_bird_labels.txt',
    '--input {}'.format(event.src_path)
]
sp = subprocess.run(args, capture_output=True) # Blocked until script execution is complete
print(sp.stdout)

Answer 2

扩展上一个答案，这非常好并且可以工作，但效率不高。问题是每次有新文件时 interpreter/model/engine/labels 都需要初始化。这需要很长时间（几乎几秒钟），因为每个推理调用在 pi4 + 珊瑚加速器上应该只需要 ~5ms。

由于 edgetpu's API 带有模块，您可以在 python 脚本中直接调用，下面是我的处理方式：

# Add these imports
from edgetpu.classification.engine import ClassificationEngine
from edgetpu.utils import dataset_utils
from PIL import Image

# [EDIT] since user had issue calling "self" in the class, I'm making this a global variable so that it could be called anywhere.

print('Initializing engine and labels')
engine = ClassificationEngine('models/mobilenet_v2_1.0_224_inat_bird_quant_edgetpu.tflite')
labels = dataset_utils.read_label_file('models/inat_bird_labels.txt')

# Keep class Watcher the same
class Watcher:
.....

class Handler(FileSystemEventHandler):
    @staticmethod
    def on_any_event(event):
        if event.is_directory:
            return None

        elif event.event_type == 'created':
            # Take any action here when a file is first created.
            print(event.src_path)
            img = Image.open(event.src_path)
            for result in engine.classify_with_image(img, top_k=3):
              print('---------------------------')
              print(labels[result[0]])
              print('Score : ', result[1])

        # elif event.event_type == 'modified':
            # Taken any action here when a file is modified.
            # print("Received modified event - %s." % event.src_path)

        elif event.event_type == 'deleted':
            # Taken any action here when a file is deleted.
            print("Received deleted event - %s." % event.src_path)