moonq/tsmark
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

have interpolation run in a thread

Browse Source
This commit is contained in:
q
2025-08-21 14:06:24 +03:00
parent 3fc383382b
commit af70258261
2 changed files with 111 additions and 54 deletions
Download Patch File Download Diff File Expand all files Collapse all files

2
tsmark/__init__.py
View File

@@ -2,7 +2,7 @@ import argparse
from tsmark.video_annotator import Marker
VERSION = "0.7.9"
VERSION = "0.7.10"
class SmartFormatter(argparse.HelpFormatter):

147
tsmark/video_annotator.py
View File

@@ -49,7 +49,9 @@ class Marker:
self.points_interpolated = {}
self.point_index = None
self.points_interpolation_enabled = True
self.points_interpolation_required = False
self.points_interpolation_required = {}
self.points_interpolation_thread = None
self.points_interpolation_thread_exit = False
self.message = None
self.message_timer = time.time()
@@ -257,7 +259,7 @@ class Marker:
if index == self.point_index and self.point_click == 1:
continue
current = self.get_interpolated_point(index=index)
if current["type"] in ("pre", "post"):
if current["type"] in ("pre", "post", None):
continue
if current["visible"] == "hidden":
continue
@@ -283,14 +285,21 @@ class Marker:
)
# Show current track
x, y = [20, 70]
intrp_str = (
""
if not self.points_interpolation_enabled
else " i*" if True in self.points_interpolation_required.values() else " i"
)
self.shadow_text(
frame,
"P:" + str(self.point_index),
f"P:{str(self.point_index)}{intrp_str}",
(x, y),
0.5,
1,
(255, 255, 255),
)
try:
current = self.get_interpolated_point()
if current["type"] is not None:
@@ -306,6 +315,7 @@ class Marker:
cv2.circle(frame, (current["cx"], current["cy"]), 10, color, 1)
if self.points_interpolation_enabled:
history = []
for p in range(max(1, int(self.nr - self.viewer_fps)), self.nr + 1):
po = self.get_interpolated_point(p)
@@ -349,7 +359,7 @@ class Marker:
if direction == "previous":
for ts in reversed(sorted(list(self.points[self.point_index].keys()))):
if ts < self.nr - 1:
if ts < self.nr:
return set_nr(ts)
except Exception:
@@ -372,7 +382,7 @@ class Marker:
"y1": ip["y1"],
"visible": POINT_VISIBILITY[0],
}
self.interpolate_points()
self.points_interpolation_required[self.point_index] = True
except Exception:
pass
@@ -415,6 +425,17 @@ class Marker:
if index is None:
index = self.point_index
if index in self.points:
if nr in self.points[index]:
value = self.get_point(nr=nr, index=index)
value.update({"type": "key" if value["x0"] is not None else None, "age": 0})
return value
if not self.points_interpolation_enabled:
value = self.get_point(nr=nr, index=index)
value.update({"type": "key" if value["x0"] is not None else None, "age": 0})
return value
if index in self.points_interpolated:
if nr in self.points_interpolated[index]:
value = self.points_interpolated[index][nr].copy()
@@ -441,6 +462,8 @@ class Marker:
def convert_interpolated_points(self):
if self.point_click == 1 and self.point_index in self.points:
self.toggle_interpolation(True)
for nr in range(self.frames):
ip = self.get_interpolated_point(nr=nr)
if ip["type"] == "interp" and ip["visible"] == POINT_VISIBILITY[0]:
@@ -451,7 +474,7 @@ class Marker:
"y1": ip["y1"],
"visible": POINT_VISIBILITY[0],
}
self.interpolate_points()
# self.interpolate_points()
def modify_point(self, position, x, y):
"""position: tl topleft, br bottomright, c center"""
@@ -536,7 +559,8 @@ class Marker:
self.points[self.point_index][self.nr]["y1"],
)
self.interpolate_points()
# self.interpolate_points()
self.points_interpolation_required[self.point_index] = True
def modify_point_wh(self):
@@ -558,7 +582,7 @@ class Marker:
self.points[self.point_index][self.nr]["y1"] = int(curr_point["cy"] + new_hh)
self.points[self.point_index][self.nr]["visible"] = POINT_VISIBILITY[0]
self.interpolate_points()
self.points_interpolation_required[self.point_index] = True
def toggle_point_visibility(self):
@@ -585,6 +609,7 @@ class Marker:
except Exception as e:
print(e)
pass
self.points_interpolation_required[self.point_index] = True
def track_point(self):
@@ -594,11 +619,14 @@ class Marker:
if self.opts.output_points is None:
return
self.toggle_interpolation(True)
tracker_gui = TrackerGUI(self)
if len(tracker_gui.points) > 0:
for nr in tracker_gui.points:
self.points[self.point_index][nr] = tracker_gui.points[nr]
self.interpolate_points()
# self.interpolate_points()
self.points_interpolation_required[self.point_index] = True
self.nr = max(tracker_gui.points) - 1
self.read_next = True
@@ -655,13 +683,13 @@ class World:
post: after any keyframes
"""
if self.points_interpolation_enabled:
process = threading.Thread(target=self.interpolate_points_in_thread, args=(point_index,))
process.start()
if self.points_interpolation_thread is None:
self.points_interpolation_thread = threading.Thread(target=self.interpolate_points_in_thread, args=())
self.points_interpolation_thread.start()
def interpolate_points_in_thread(self, point_index=None):
if not self.points_interpolation_thread.is_alive():
self.points_interpolation_thread = threading.Thread(target=self.interpolate_points_in_thread, args=())
self.points_interpolation_thread.start()
if point_index is None:
point_index = self.point_index
@@ -675,11 +703,10 @@ class World:
if not point_index in self.points:
return
self.points_interpolation_required[point_index] = False
if not point_index in self.points_interpolated:
self.points_interpolated[point_index] = {key: {} for key in range(self.frames)}
# ~ self.points_interpolation_required = False
new_points = {k: v for k, v in self.points_interpolated[point_index].items()}
if len(self.points[point_index]) == 1: # only one point added
@@ -742,7 +769,25 @@ class World:
self.points_interpolated[point_index] = new_points
def toggle_interpolation(self):
def interpolate_points_in_thread(self):
self.points_interpolation_frequency = 1
while True:
if self.points_interpolation_thread_exit:
return
time.sleep(self.points_interpolation_frequency)
if not self.points_interpolation_enabled:
continue
for point_index in self.points_interpolation_required:
if self.points_interpolation_required[point_index]:
self.interpolate_points(point_index)
def toggle_interpolation(self, value=None):
if value is not None:
self.points_interpolation_enabled = not value
self.points_interpolation_enabled = not self.points_interpolation_enabled
if self.points_interpolation_enabled:
@@ -1341,6 +1386,21 @@ class World:
self.modify_point_wh()
elif k & 0xFF == ord("u"): # toggle interpolation
self.toggle_interpolation()
if self.point_click == 0:
self.shadow_text(
frame_visu,
(
"Point interpolation turned on"
if self.points_interpolation_enabled
else "Point interpolation turned off"
),
(20, 70),
0.9,
2,
(255, 255, 255),
)
cv2.imshow("tsmark", frame_visu)
k2 = cv2.waitKey(1000)
elif k & 0xFF == ord("x"): # toggle ts
if self.point_click == 1:
self.toggle_point(self.nr)
@@ -1385,6 +1445,7 @@ class World:
except Exception as e:
print(e)
self.points_interpolation_thread_exit = True
self.video_reader.release()
cv2.destroyAllWindows()
self.print_timestamps()
@@ -1432,6 +1493,8 @@ class TrackerGUI:
tracked[0] = [*bbox, 1]
for i in range(max_frames):
# Read a new frame
self.marker.video_reader.set(cv2.CAP_PROP_POS_FRAMES, self.marker.nr + i)
ok, frame = self.marker.video_reader.read()
frame = cv2.resize(frame.copy(), self.marker.video_res)
if not ok:
@@ -1440,10 +1503,10 @@ class TrackerGUI:
ok, bbox = tracker.update(frame)
if ok:
# Tracking success
if self.marker.nr + i + 1 in self.marker.points[self.marker.point_index]:
point = self.marker.get_point(nr=self.marker.nr + i + 1)
if self.marker.nr + i in self.marker.points[self.marker.point_index]:
point = self.marker.get_point(nr=self.marker.nr + i)
bbox = tuple([point["x0"], point["y0"], point["w"], point["h"]])
tracked[i + 1] = [*bbox, 1]
tracked[i] = [*bbox, 1]
show_message = f"Tracking... ({i}/{max_frames})"
else:
# Tracking failure
@@ -1471,20 +1534,16 @@ class TrackerGUI:
while True:
if done:
break
self.marker.video_reader.set(cv2.CAP_PROP_POS_FRAMES, self.marker.nr)
i = -1
i = 0
while True:
show_time = time.time()
self.marker.video_reader.set(cv2.CAP_PROP_POS_FRAMES, self.marker.nr + i)
if done:
break
if paused:
frame = frame_copy.copy()
if (not paused) or seek:
ok, frame = self.marker.video_reader.read()
frame = cv2.resize(frame.copy(), self.marker.video_res)
frame_copy = frame.copy()
i += 1
seek = False
self.marker.shadow_text(frame, f"Accept? ({i+1}/{max_frames})", (100, 80), 0.75, 2, (255, 255, 255))
if i in tracked:
bbox = tracked[i]
@@ -1495,7 +1554,7 @@ class TrackerGUI:
cv2.rectangle(frame, p1, p2, color, thicc, 1)
cv2.imshow("tsmark - tracker", frame)
# speed up fps by 2
time_to_wait = self.marker.viewer_spf / 2 - time.time() + show_time
time_to_wait = 0.2 if paused else (self.marker.viewer_spf / 2 - time.time() + show_time)
k = cv2.waitKey(max(1, int(time_to_wait * 1000)))
if k & 0xFF == ord("q") or k & 0xFF == 13: # accept with q or enter
done = True
@@ -1508,46 +1567,44 @@ class TrackerGUI:
paused = not paused
# Movement =================
elif k & 0xFF == 83 or k & 0xFF == ord("l"): # right arrow
i += int(self.marker.fps) - 1
i += int(self.marker.fps)
seek = True
elif k & 0xFF == 81 or k & 0xFF == ord("j"): # left arrow
i -= int(self.marker.fps) + 1
seek = True
i -= int(self.marker.fps)
# Move by frame
elif k & 0xFF == ord(".") or k & 0xFF == ord("c"):
paused = True
seek = True
i += 1
elif k & 0xFF == ord(",") or k & 0xFF == ord("z"):
paused = True
i -= 2
seek = True
i -= 1
elif k & 0xFF == ord("x"):
cut_after = i
cut_after = i + 1
# TODO: ord("h") for help!
if i >= max_frames - 1:
i = max_frames - 2
paused = True
seek = True
if i < 0:
i = -1
i = 0
paused = True
seek = True
if seek:
self.marker.video_reader.set(cv2.CAP_PROP_POS_FRAMES, self.marker.nr + i + 1)
if not paused:
i += 1
cv2.destroyWindow("tsmark - tracker")
self.marker.nr = old_nr - 1
self.marker.read_next = True
self.points = {}
for i in sorted(list(tracked.keys())):
if i >= cut_after:
continue
self.points[self.marker.nr + i + 1] = {
self.points[old_nr + i] = {
"x0": tracked[i][0],
"y0": tracked[i][1],
"x1": tracked[i][0] + tracked[i][2],
"y1": tracked[i][1] + tracked[i][3],
"visible": POINT_VISIBILITY[0],
}
self.marker.nr = old_nr + cut_after - 1
self.marker.read_next = True