Kawałek Pythona - jak czytam pliki schodzące z kamery

Behlur_Olderys · 4 Czerwca 2022

Hej,

W ramach wymiany myśli chciałem zaprezentować małą klaskę którą napisałem na Zlocie. Jej zadaniem jest wyłapywać najnowsze pliki z pewnego folderu (w domyśle - Capture z kamery) i wykonywać na nich dowolną czynność. Potrzebna jest dowolna klasa przetwarzająca z funkcjami "init" i "process" wykorzystująca nazwę pliku i jego timestamp

Z chęcią poczytałbym jakieś uwagi/sugestie w ramach takiego forumowego "code review"

Kod wygląda tak:

import time
from tkinter import filedialog
import os
from time import sleep
from threading import Thread


def is_file_wanted(f, extensions):
    """
    extensions is an array of wanted file extensions
    """
    is_any = any([f.lower().endswith(e) for e in extensions])
    return is_any


def is_file_fits(f):
    return is_file_wanted(f, ["fit", "fits"])


def is_file_png(f):
    return is_file_wanted(f, ["png"])


def get_last_files(d, filter_fun):
    """
    returns list of pairs (file, timestamp) sorted by timestamp
    """
    files = [os.path.join(d, f) for f in os.listdir(d) if filter_fun(f)]
    files = [(p, os.path.getctime(p)) for p in files]
    return sorted(files, key=lambda x: x[1])


class RecentImagesProvider:
    def __init__(self, processor, filter_fun):
        self._processor = processor
        self._filter_fun = filter_fun
        self._files = []
        self._kill = False
        self._thread = None
        self._filenames = None
        self._main_dir = None

    def start(self):
        self._thread = Thread(target=self._run)
        self._thread.start()

    def _run(self):
        self._main_dir = filedialog.askdirectory(title="Open dir with images")
        while not self._files:
            self._files = get_last_files(self._main_dir, self._filter_fun)
            if not self._files:
                print("Waiting 1s for new files...")
                sleep(1)
        self._filenames = [f[0] for f in self._files]
        if not self._processor.init(*self._files[-1]):
            print("Initialization failed, ending...")
            return

        self._process()

    def kill(self):
        self._kill = True
        self._thread.join()

    def __del__(self):
        if not self._kill:
            self._kill = True
            self._thread.join()

    def _process(self):
        while not self._kill:
            files = get_last_files(self._main_dir, self._filter_fun)
            new_files = [f for f in files if f[0] not in self._filenames]

            if not new_files:
                print("Waiting 1s for new files...")
                sleep(1)
                continue

            print(f"Acquired  {len(new_files)} new files")
            for f, t in new_files:
                self._processor.process(f, t)

            self._files += new_files
            self._filenames = [f[0] for f in files]


class TestProcessor:
    def __init__(self):
        pass

    def init(self, f, t):
        print(f"Initializing test processor with file {f}")

    def process(self, f, t):
        print(f"Processing file {f}")


# przykładowe użycie:
t = TestProcessor()
r = RecentImagesProvider(t, is_file_png)
r.start()
#(...) cokolwiek innego
#r.kill()

stfn · 4 Czerwca 2022

Hej, nie mam doświadczenia z wątkami i tktinterem, ale ogólnie wygląda ok, po pierwszym przejrzeniu usprawniłbym trochę nazewnictwo:

is_file_wanted -> is_file_of_right_extension

get_last_files - ta funkcja zwraca *wszystkie* pliki z folderu, ewentualnie przefiltrowane, ale niekoniecznie *ostatnie* co nie? Więc nie użyłbym _last_ tutaj. Może po prostu get_files()? get_sorted_files()?

filter_fun przemianowałbym na filter_func, wtedy imho to jaśniejsze że chodzi o funkcję, i dorzuciłbym do get_last_files opcję niepodawania jej, dać jej default value na None i potem t obsłużyć w kodzie funkcji.

Bolą mnie też trochę jednoliterowe zmienne i brak typowania, ale to są wszystko nitpicki

Behlur_Olderys · 4 Czerwca 2022

35 minut temu, stfn napisał:

get_last_files - ta funkcja zwraca *wszystkie* pliki z folderu, ewentualnie przefiltrowane, ale niekoniecznie *ostatnie* co nie?

To prawda. Ogólnie przyznaje się do postawionych zarzutów

Jeszcze tytułem objaśnienia:

Tej klasy można użyć np. by zrealizować prosty guiding.

Zasadzamy się na katalog gdzie kamera guidująca zrzuca pliki. Następnie wybieramy gwiazdkę (w init) a potem obliczmy np. centroid albo inną funkcję przybliżającą środek gwiazdy (np. najjaśniejszy piksel itp.)

W process można obliczyć te wartości dla każdej kolejnej klatki i wysyłać korekty do montażu.