security_vision/trap
1
0
Fork 0
Code Issues Pull requests Projects Releases Packages Wiki Activity

Compare commits

base: security_vision:render_historic
Branches Tags
security_vision:main
security_vision:animation_window
security_vision:render_historic
..
compare: security_vision:main
Branches Tags
security_vision:animation_window
security_vision:main
security_vision:render_historic

1 commit
render_his ... main

Author SHA1 Message Date
Ruben van de Ven
3d8cb7ef70 fix crash on track delete 2024-11-05 15:52:02 +01:00
2 changed files with 81 additions and 191 deletions
Show stats Expand all files Collapse all files

2
trap/frame_emitter.py
View file

@ -95,7 +95,9 @@ class Track:
coords = self.get_projected_history(H) coords = self.get_projected_history(H)
return [{"x":c[0], "y":c[1]} for c in coords] return [{"x":c[0], "y":c[1]} for c in coords]
@dataclass @dataclass
class Frame: class Frame:

270
trap/renderer.py
View file

@ -18,31 +18,16 @@ import tempfile
from pathlib import Path from pathlib import Path
import shutil import shutil
import math import math
from collections import deque
from pyglet import shapes from pyglet import shapes
from PIL import Image from PIL import Image
from trap.frame_emitter import DetectionState, Frame, Track from trap.frame_emitter import DetectionState, Frame, Track
from dataclasses import dataclass
logger = logging.getLogger("trap.renderer") logger = logging.getLogger("trap.renderer")
@dataclass
class LineAnimConfig:
start_opacity: float
target_opacity: float
decay: float
@dataclass
class AnimConfig:
clear_color = (0,0,0,0)
video_opacity = 100 # on 255 scale
tracks: LineAnimConfig = LineAnimConfig(80, 180, 60)
predictions: LineAnimConfig = LineAnimConfig(20, 50, 10)
class FrameAnimation: class FrameAnimation:
def __init__(self, frame: Frame): def __init__(self, frame: Frame):
self.start_time = time.time() self.start_time = time.time()
@ -77,48 +62,27 @@ class DrawnTrack:
self.update_at = self.created_at = time.time() self.update_at = self.created_at = time.time()
self.track_id = track_id self.track_id = track_id
self.renderer = renderer self.renderer = renderer
self.drawn_positions = []
# self.drawn_predictions = []
self.predictions: deque[DrawnPrediction] = deque(maxlen=2) # TODO; make configurable
self.shapes: list[pyglet.shapes.Line] = []
self.set_track(track, H) self.set_track(track, H)
self.set_prediction(track) self.drawn_positions = []
self.drawn_predictions = []
def __del__(self): self.shapes: list[pyglet.shapes.Line] = []
self.destroy() self.pred_shapes: list[list[pyglet.shapes.Line]] = []
def destroy(self):
# TODO)) not working yet
logger.warning(f'del track {self}')
for s in self.shapes:
s.delete()
for p in self.predictions:
p.destroy()
def set_track(self, track: Track, H): def set_track(self, track: Track, H):
self.update_at = time.time() self.update_at = time.time()
self.track = track self.track = track
self.H = H self.H = H
self.coords = [d.get_foot_coords() for d in track.history[-30:]] # maximum of 30 past positions (prevent collapse of FPS on loitering object) self.coords = [d.get_foot_coords() for d in track.history]
# perhaps only do in constructor: # perhaps only do in constructor:
self.inv_H = np.linalg.pinv(self.H) self.inv_H = np.linalg.pinv(self.H)
def set_prediction(self, track: Track):
# TODO: turn into add_prediction
pred_coords = [] pred_coords = []
if not track.predictions:
return
for pred_i, pred in enumerate(track.predictions): for pred_i, pred in enumerate(track.predictions):
pred_coords.append(cv2.perspectiveTransform(np.array([pred]), self.inv_H)[0].tolist()) pred_coords.append(cv2.perspectiveTransform(np.array([pred]), self.inv_H)[0].tolist())
# self.pred_coords = pred_coords self.pred_coords = pred_coords
self.predictions.append(DrawnPrediction(self, pred_coords))
# color = (128,0,128) if pred_i else (128, # color = (128,0,128) if pred_i else (128,
@ -128,32 +92,29 @@ class DrawnTrack:
''' '''
# TODO: make lerp, currently quick way to get results # TODO: make lerp, currently quick way to get results
for i, pos in enumerate(self.drawn_positions): for i, pos in enumerate(self.drawn_positions):
self.drawn_positions[i][0] = int(exponentialDecay(self.drawn_positions[i][0], self.coords[i][0], AnimConfig.tracks.decay, dt)) self.drawn_positions[i][0] = int(exponentialDecay(self.drawn_positions[i][0], self.coords[i][0], 16, dt))
self.drawn_positions[i][1] = int(exponentialDecay(self.drawn_positions[i][1], self.coords[i][1], AnimConfig.tracks.decay, dt)) self.drawn_positions[i][1] = int(exponentialDecay(self.drawn_positions[i][1], self.coords[i][1], 16, dt))
if len(self.coords) > len(self.drawn_positions): if len(self.coords) > len(self.drawn_positions):
self.drawn_positions.extend(self.coords[len(self.drawn_positions):]) self.drawn_positions.extend(self.coords[len(self.drawn_positions):])
# Superseded by individual drawnprediction elements for a, drawn_prediction in enumerate(self.drawn_predictions):
# for a, drawn_prediction in enumerate(self.drawn_predictions): for i, pos in enumerate(drawn_prediction):
# for i, pos in enumerate(drawn_prediction): # TODO: this should be done in polar space starting from origin (i.e. self.drawn_posision[-1])
# # TODO: this should be done in polar space starting from origin (i.e. self.drawn_posision[-1]) decay = max(3, (18/i) if i else 10) # points further away move with more delay
# decay = max(3, (18/i) if i else 10) # points further away move with more delay decay = 6
# decay = 6 origin = self.drawn_positions[-1]
# origin = self.drawn_positions[-1] drawn_r, drawn_angle = relativePointToPolar( origin, drawn_prediction[i])
# drawn_r, drawn_angle = relativePointToPolar( origin, drawn_prediction[i]) pred_r, pred_angle = relativePointToPolar(origin, self.pred_coords[a][i])
# pred_r, pred_angle = relativePointToPolar(origin, self.pred_coords[a][i]) r = exponentialDecay(drawn_r, pred_r, decay, dt)
# r = exponentialDecay(drawn_r, pred_r, decay, dt) angle = exponentialDecay(drawn_angle, pred_angle, decay, dt)
# angle = exponentialDecay(drawn_angle, pred_angle, decay, dt) x, y = relativePolarToPoint(origin, r, angle)
# x, y = relativePolarToPoint(origin, r, angle) self.drawn_predictions[a][i] = int(x), int(y)
# self.drawn_predictions[a][i] = int(x), int(y) # self.drawn_predictions[i][0] = int(exponentialDecay(self.drawn_predictions[i][0], self.pred_coords[i][0], decay, dt))
# # self.drawn_predictions[i][0] = int(exponentialDecay(self.drawn_predictions[i][0], self.pred_coords[i][0], decay, dt)) # self.drawn_predictions[i][1] = int(exponentialDecay(self.drawn_predictions[i][1], self.pred_coords[i][1], decay, dt))
# # self.drawn_predictions[i][1] = int(exponentialDecay(self.drawn_predictions[i][1], self.pred_coords[i][1], decay, dt))
# if len(self.pred_coords) > len(self.drawn_predictions):
# self.drawn_predictions.extend(self.pred_coords[len(self.drawn_predictions):])
if len(self.pred_coords) > len(self.drawn_predictions):
self.drawn_predictions.extend(self.pred_coords[len(self.drawn_predictions):])
# for a, drawn_prediction in self.drawn_predictions: # for a, drawn_prediction in self.drawn_predictions:
# if len(self.pred_coords) > len(self.drawn_predictions): # if len(self.pred_coords) > len(self.drawn_predictions):
# self.drawn_predictions.extend(self.pred_coords[len(self.drawn_predictions):]) # self.drawn_predictions.extend(self.pred_coords[len(self.drawn_predictions):])
@ -178,7 +139,7 @@ class DrawnTrack:
if ci >= len(self.shapes): if ci >= len(self.shapes):
# TODO: add color2 # TODO: add color2
line = self.renderer.gradientLine(x, y, x2, y2, 3, color, color, batch=self.renderer.batch_anim) line = self.renderer.gradientLine(x, y, x2, y2, 3, color, color, batch=self.renderer.batch_anim)
line.opacity = AnimConfig.tracks.start_opacity line.opacity = 5
self.shapes.append(line) self.shapes.append(line)
else: else:
@ -186,111 +147,50 @@ class DrawnTrack:
line.x, line.y = x, y line.x, line.y = x, y
line.x2, line.y2 = x2, y2 line.x2, line.y2 = x2, y2
line.color = color line.color = color
line.opacity = int(exponentialDecay(line.opacity, AnimConfig.tracks.target_opacity, AnimConfig.tracks.start_opacity, dt)) line.opacity = int(exponentialDecay(line.opacity, 180, 3, dt))
# TODO: basically a duplication of the above, do this smarter? # TODO: basically a duplication of the above, do this smarter?
# TODO: add intermediate segment # TODO: add intermediate segment
#color = colorset[self.track_id % len(colorset)] color = colorset[self.track_id % len(colorset)]
for drawn_prediction in self.predictions: for a, drawn_predictions in enumerate(self.drawn_predictions):
drawn_prediction.update_opacities(dt) if len(self.pred_shapes) <= a:
self.pred_shapes.append([])
# for a, drawn_predictions in enumerate(self.drawn_predictions):
# if len(self.pred_shapes) <= a:
# self.pred_shapes.append([])
# if len(self.pred_shapes[a]) > (len(drawn_predictions) +1): if len(self.pred_shapes[a]) > (len(drawn_predictions) +1):
# self.pred_shapes[a] = self.pred_shapes[a][:len(drawn_predictions)] self.pred_shapes[a] = self.pred_shapes[a][:len(drawn_predictions)]
# # for i, pos in drawn_predictions.enumerate(): # for i, pos in drawn_predictions.enumerate():
# for ci in range(0, len(drawn_predictions)): for ci in range(0, len(drawn_predictions)):
# if ci == 0:
# x, y = [int(p) for p in self.drawn_positions[-1]]
# else:
# x, y = [int(p) for p in drawn_predictions[ci-1]]
# x2, y2 = [int(p) for p in drawn_predictions[ci]]
# y, y2 = self.renderer.window.height - y, self.renderer.window.height - y2
# # color = [255,0,0]
# # print(x,y,x2,y2,color)
# if ci >= len(self.pred_shapes[a]):
# # TODO: add color2
# line = self.renderer.gradientLine(x, y, x2, y2, 3, color, color, batch=self.renderer.batch_anim)
# line.opacity = 5
# self.pred_shapes[a].append(line)
# else:
# line = self.pred_shapes[a][ci-1]
# line.x, line.y = x, y
# line.x2, line.y2 = x2, y2
# line.color = color
# decay = (16/ci) if ci else 16
# half = len(drawn_predictions) / 2
# if ci < half:
# target_opacity = 180
# else:
# target_opacity = (1 - ((ci - half) / half)) * 180
# line.opacity = int(exponentialDecay(line.opacity, target_opacity, decay, dt))
class DrawnPrediction:
def __init__(self, drawn_track: DrawnTrack, coords: list[list] = []):
self.created_at = time.time()
# self.renderer = renderer
self.drawn_track = drawn_track
self.coords = coords
self.color = colorset[self.drawn_track.track_id % len(colorset)]
self.pred_shapes: list[list[pyglet.shapes.Line]] = []
# coords is a list of predictions
for a, coords in enumerate(self.coords):
prediction_shapes = []
for ci in range(0, len(coords)):
if ci == 0: if ci == 0:
x, y = [int(p) for p in self.drawn_track.coords[-1]] x, y = [int(p) for p in self.drawn_positions[-1]]
else: else:
x, y = [int(p) for p in coords[ci-1]] x, y = [int(p) for p in drawn_predictions[ci-1]]
x2, y2 = [int(p) for p in coords[ci]] x2, y2 = [int(p) for p in drawn_predictions[ci]]
# flip in window: y, y2 = self.renderer.window.height - y, self.renderer.window.height - y2
y, y2 = self.drawn_track.renderer.window.height - y, self.drawn_track.renderer.window.height - y2 # color = [255,0,0]
# print(x,y,x2,y2,color)
line = self.drawn_track.renderer.gradientLine(x, y, x2, y2, 3, self.color, self.color, batch=self.drawn_track.renderer.batch_anim)
line.opacity = 5
prediction_shapes.append(line)
self.pred_shapes.append(prediction_shapes)
def destroy(self):
for pred in self.pred_shapes:
for shape in pred:
shape.delete()
if ci >= len(self.pred_shapes[a]):
def update_opacities(self, dt: float): # TODO: add color2
""" line = self.renderer.gradientLine(x, y, x2, y2, 3, color, color, batch=self.renderer.batch_anim)
Update the opacties of the drawn line, by only using the dt provided by the renderer line.opacity = 5
Done using exponential decal, with a different decay value per item self.pred_shapes[a].append(line)
"""
for a, coords in enumerate(self.coords):
for ci in range(0, len(coords)):
line = self.pred_shapes[a][ci-1]
# Positions of prediction no longer update
# line.x, line.y = x, y
# line.x2, line.y2 = x2, y2
# line.color = color
# lower decay for further points slows down fade in
decay = (AnimConfig.predictions.decay/(3*ci)) if ci else AnimConfig.predictions.decay
half = len(coords) / 2
if ci < half:
target_opacity = AnimConfig.predictions.target_opacity
else: else:
target_opacity = (1 - ((ci - half) / half)) * AnimConfig.predictions.target_opacity line = self.pred_shapes[a][ci-1]
line.opacity = int(exponentialDecay(line.opacity, target_opacity, decay, dt)) line.x, line.y = x, y
# logger.info(f"{target_opacity=}, {line.opacity=}") line.x2, line.y2 = x2, y2
line.color = color
decay = (16/ci) if ci else 16
half = len(drawn_predictions) / 2
if ci < half:
target_opacity = 180
else:
target_opacity = (1 - ((ci - half) / half)) * 180
line.opacity = int(exponentialDecay(line.opacity, target_opacity, decay, dt))
class FrameWriter: class FrameWriter:
@ -379,8 +279,7 @@ class Renderer:
self.window.set_handler('on_refresh', self.on_refresh) self.window.set_handler('on_refresh', self.on_refresh)
self.window.set_handler('on_close', self.on_close) self.window.set_handler('on_close', self.on_close)
# Purple background color: pyglet.gl.glClearColor(81./255, 20/255, 46./255, 0)
pyglet.gl.glClearColor(*AnimConfig.clear_color)
self.fps_display = pyglet.window.FPSDisplay(window=self.window, color=(255,255,255,255)) self.fps_display = pyglet.window.FPSDisplay(window=self.window, color=(255,255,255,255))
self.fps_display.label.x = self.window.width - 50 self.fps_display.label.x = self.window.width - 50
self.fps_display.label.y = self.window.height - 17 self.fps_display.label.y = self.window.height - 17
@ -516,32 +415,34 @@ class Renderer:
def check_frames(self, dt): def check_frames(self, dt):
new_tracks = False
try: try:
self.frame: Frame = self.frame_sock.recv_pyobj(zmq.NOBLOCK) self.frame: Frame = self.frame_sock.recv_pyobj(zmq.NOBLOCK)
if not self.first_time: if not self.first_time:
self.first_time = self.frame.time self.first_time = self.frame.time
img = cv2.GaussianBlur(self.frame.img, (15, 15), 0) img = cv2.GaussianBlur(self.frame.img, (15, 15), 0)
img = cv2.cvtColor(cv2.cvtColor(img, cv2.COLOR_BGR2GRAY), cv2.COLOR_GRAY2RGB) img = cv2.flip(cv2.cvtColor(img, cv2.COLOR_BGR2RGB), 0)
channels = 3 # unfortunately, pyglet seems to draw single channel as Red only img = pyglet.image.ImageData(self.frame_size[0], self.frame_size[1], 'RGB', img.tobytes())
img = pyglet.image.ImageData(self.frame_size[0], self.frame_size[1], 'RGB', img.tobytes(), pitch=self.frame_size[0] * -1 * channels)
# don't draw in batch, so that it is the background # don't draw in batch, so that it is the background
self.video_sprite = pyglet.sprite.Sprite(img=img, batch=self.batch_bg) self.video_sprite = pyglet.sprite.Sprite(img=img, batch=self.batch_bg)
self.video_sprite.opacity = AnimConfig.video_opacity self.video_sprite.opacity = 100
except zmq.ZMQError as e: except zmq.ZMQError as e:
# idx = frame.index if frame else "NONE" # idx = frame.index if frame else "NONE"
# logger.debug(f"reuse video frame {idx}") # logger.debug(f"reuse video frame {idx}")
pass pass
try: try:
self.prediction_frame: Frame = self.prediction_sock.recv_pyobj(zmq.NOBLOCK) self.prediction_frame: Frame = self.prediction_sock.recv_pyobj(zmq.NOBLOCK)
self.update_predictions() new_tracks = True
except zmq.ZMQError as e: except zmq.ZMQError as e:
pass pass
try: try:
self.tracker_frame: Frame = self.tracker_sock.recv_pyobj(zmq.NOBLOCK) self.tracker_frame: Frame = self.tracker_sock.recv_pyobj(zmq.NOBLOCK)
self.update_tracks() new_tracks = True
except zmq.ZMQError as e: except zmq.ZMQError as e:
pass pass
if new_tracks:
self.update_tracks()
def update_tracks(self): def update_tracks(self):
"""Updates the track objects and shapes. Called after setting `prediction_frame` """Updates the track objects and shapes. Called after setting `prediction_frame`
@ -553,31 +454,20 @@ class Renderer:
# # TODO fade out # # TODO fade out
# del self.drawn_tracks[track_id] # del self.drawn_tracks[track_id]
if self.tracker_frame: if self.prediction_frame:
for track_id, track in self.tracker_frame.tracks.items(): for track_id, track in self.prediction_frame.tracks.items():
if track_id not in self.drawn_tracks: if track_id not in self.drawn_tracks:
self.drawn_tracks[track_id] = DrawnTrack(track_id, track, self, self.tracker_frame.H) self.drawn_tracks[track_id] = DrawnTrack(track_id, track, self, self.prediction_frame.H)
else: else:
self.drawn_tracks[track_id].set_track(track, self.tracker_frame.H) self.drawn_tracks[track_id].set_track(track, self.prediction_frame.H)
# clean up # clean up
for track_id in list(self.drawn_tracks.keys()): for track_id in list(self.drawn_tracks.keys()):
# TODO make delay configurable # TODO make delay configurable
if self.drawn_tracks[track_id].update_at < time.time() - 5: if self.drawn_tracks[track_id].update_at < time.time() - 5:
# TODO fade out # TODO fade out
self.drawn_tracks[track_id].destroy()
del self.drawn_tracks[track_id] del self.drawn_tracks[track_id]
def update_predictions(self):
if self.prediction_frame:
for track_id, track in self.prediction_frame.tracks.items():
if track_id not in self.drawn_tracks:
self.drawn_tracks[track_id] = DrawnTrack(track_id, track, self, self.prediction_frame.H)
logger.warning("Prediction for uninitialised frame. This should not happen? (maybe huge delay in prediction?)")
else:
self.drawn_tracks[track_id].set_prediction(track)
def on_key_press(self, symbol, modifiers): def on_key_press(self, symbol, modifiers):
print('A key was pressed, use f to hide') print('A key was pressed, use f to hide')
@ -611,16 +501,14 @@ class Renderer:
self.window.clear() self.window.clear()
self.batch_bg.draw() self.batch_bg.draw()
for track in self.drawn_tracks.values(): for track in self.drawn_tracks.values():
for shape in track.shapes: for shape in track.shapes:
shape.draw() # for some reason the batches don't work shape.draw() # for some reason the batches don't work
for track in self.drawn_tracks.values(): for track in self.drawn_tracks.values():
for prediction in track.predictions: for shapes in track.pred_shapes:
for shapes in prediction.pred_shapes: for shape in shapes:
for shape in shapes: shape.draw()
shape.draw()
# self.batch_anim.draw() # self.batch_anim.draw()
self.batch_overlay.draw() self.batch_overlay.draw()
@ -782,12 +670,12 @@ def decorate_frame(frame: Frame, prediction_frame: Frame, first_time: float, con
# or https://api.arcade.academy/en/stable/index.html (supports gradient color in line -- "Arcade is built on top of Pyglet and OpenGL.") # or https://api.arcade.academy/en/stable/index.html (supports gradient color in line -- "Arcade is built on top of Pyglet and OpenGL.")
frame.img frame.img
# # Fill image with red color(set each pixel to red) overlay = np.zeros(frame.img.shape, np.uint8)
# overlay = np.zeros(frame.img.shape, np.uint8) # Fill image with red color(set each pixel to red)
# overlay[:] = (130, 0, 75) overlay[:] = (130, 0, 75)
# img = cv2.addWeighted(frame.img, .4, overlay, .6, 0) img = cv2.addWeighted(frame.img, .4, overlay, .6, 0)
img = frame.img.copy() # img = frame.img.copy()
# all not working: # all not working:
# if i == 1: # if i == 1: