jgrewe/fixtracks
1
0
Fork 0
Code Issues 4 Pull Requests Actions Packages Projects Releases Wiki Activity

[classifier] working but not really...

Browse Source
This commit is contained in:
Jan Grewe 2025-02-25 18:45:37 +01:00
parent f1a4f4dc84
commit 430ee4fac7
1 changed files with 151 additions and 99 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

238
fixtracks/widgets/classifier.py
View File

@ -12,6 +12,17 @@ from fixtracks.utils.trackingdata import TrackingData
from IPython import embed from IPython import embed
class Detection():
def __init__(self, id, frame, track, position, orientation, length, userlabeled):
self.id = id
self.frame = frame
self.track = track
self.position = position
self.score = 0.0
self.angle = orientation
self.length = length
self.userlabeled = userlabeled
class WorkerSignals(QObject): class WorkerSignals(QObject):
error = Signal(str) error = Signal(str)
running = Signal(bool) running = Signal(bool)
@ -24,7 +35,8 @@ class ConsitencyDataLoader(QRunnable):
super().__init__() super().__init__()
self.signals = WorkerSignals() self.signals = WorkerSignals()
self.data = data self.data = data
self.bendedness = self.positions = None self.bendedness = None
self.positions = None
self.lengths = None self.lengths = None
self.orientations = None self.orientations = None
self.userlabeled = None self.userlabeled = None
@ -70,6 +82,18 @@ class ConsistencyWorker(QRunnable):
@Slot() @Slot()
def run(self): def run(self):
def get_detections(frame, indices):
detections = []
for i in indices:
if np.any(self.positions[i] < 0.1):
logging.debug("Encountered probably invalid position %s", str(self.positions[i]))
continue
d = Detection(i, frame, self.tracks[i], self.positions[i],
self.orientations[i], self.lengths[i],
self.userlabeled[i])
detections.append(d)
return detections
def needs_checking(original, new): def needs_checking(original, new):
res = False res = False
for n, o in zip(new, original): for n, o in zip(new, original):
@ -82,112 +106,135 @@ class ConsistencyWorker(QRunnable):
print("all detections would be assigned to one track!") print("all detections would be assigned to one track!")
return res return res
def assign_by_distance(f, p): def assign_by_distance(d):
t1_step = f - last_frame[0] t1_step = d.frame - last_detections[1].frame
t2_step = f - last_frame[1] t2_step = d.frame - last_detections[2].frame
if t1_step == 0 or t2_step == 0: if t1_step == 0 or t2_step == 0:
print(f"framecount is zero! current frame {f}, last frame {last_frame[0]} and {last_frame[1]}") print(f"framecount is zero! current frame {f}, last frame {last_detections[1].frame} and {last_detections[2].frame}")
distance_to_trackone = np.linalg.norm(d.position - last_detections[1].position)/t1_step
distance_to_trackone = np.linalg.norm(p - last_pos[0])/t1_step distance_to_tracktwo = np.linalg.norm(d.position - last_detections[2].position)/t2_step
distance_to_tracktwo = np.linalg.norm(p - last_pos[1])/t2_step
most_likely_track = np.argmin([distance_to_trackone, distance_to_tracktwo]) + 1 most_likely_track = np.argmin([distance_to_trackone, distance_to_tracktwo]) + 1
distances = np.zeros(2) distances = np.zeros(2)
distances[0] = distance_to_trackone distances[0] = distance_to_trackone
distances[1] = distance_to_tracktwo distances[1] = distance_to_tracktwo
return most_likely_track, distances return most_likely_track, distances
def assign_by_orientation(f, o): def assign_by_orientation(d):
t1_step = f - last_frame[0] t1_step = d.frame - last_detections[1].frame
t2_step = f - last_frame[1] t2_step = d.frame - last_detections[2].frame
orientationchange = (last_angle - o) orientationchanges = np.zeros(2)
orientationchange[orientationchange > 180] = 360 - orientationchange[orientationchange > 180] for i in [1, 2]:
orientationchange /= np.array([t1_step, t2_step]) orientationchanges[i-1] = (last_detections[i].angle - d.angle)
# orientationchange = np.abs(np.unwrap((last_angle - o)/np.array([t1_step, t2_step])))
most_likely_track = np.argmin(np.abs(orientationchange)) + 1 orientationchanges[orientationchanges > 180] = 360 - orientationchanges[orientationchanges > 180]
return most_likely_track, orientationchange orientationchanges /= np.array([t1_step, t2_step])
most_likely_track = np.argmin(np.abs(orientationchanges)) + 1
def assign_by_length(o): return most_likely_track, orientationchanges
length_difference = np.abs((last_length - o))
most_likely_track = np.argmin(length_difference) + 1 def assign_by_length(d):
return most_likely_track, length_difference length_differences = np.zeros(2)
length_differences[0] = np.abs((last_detections[1].length - d.length))
def do_assignment(f, indices, assignments): length_differences[1] = np.abs((last_detections[2].length - d.length))
for i, idx in enumerate(indices): most_likely_track = np.argmin(length_differences) + 1
self.tracks[idx] = assignments[i] return most_likely_track, length_differences
last_pos[assignments[i]-1] = pp[i]
last_frame[assignments[i]-1] = f unique_frames = np.unique(self.frames)
last_angle[assignments[i]-1] = self.orientations[idx] steps = int((len(unique_frames) - self._startframe) // 100)
last_length[assignments[i]-1] += ((self.lengths[idx] - last_length[assignments[i]-1])/processed)
# self.userlabeled
last_pos = [self.positions[(self.tracks == 1) & (self.frames <= self._startframe)][-1],
self.positions[(self.tracks == 2) & (self.frames <= self._startframe)][-1]]
last_frame = [self.frames[(self.tracks == 1) & (self.frames <= self._startframe)][-1],
self.frames[(self.tracks == 2) & (self.frames <= self._startframe)][-1]]
last_angle = [self.orientations[(self.tracks == 1) & (self.frames <= self._startframe)][-1],
self.orientations[(self.tracks == 2) & (self.frames <= self._startframe)][-1]]
last_length = [self.lengths[(self.tracks == 1) & (self.frames <= self._startframe)][-1],
self.lengths[(self.tracks == 2) & (self.frames <= self._startframe)][-1]]
errors = 0 errors = 0
processed = 1 processed = 1
progress = 0 progress = 0
self._stoprequest = False self._stoprequest = False
maxframes = np.max(self.frames) last_detections = {1: None, 2: None, -1: None}
startframe = np.max(last_frame)
steps = int((maxframes - startframe) // 200)
for f in np.unique(self.frames[self.frames > startframe]): for f in unique_frames[unique_frames >= self._startframe]:
processed += 1
self.signals.currentframe.emit(f)
if self._stoprequest: if self._stoprequest:
break break
error = False
self.signals.currentframe.emit(f)
indices = np.where(self.frames == f)[0] indices = np.where(self.frames == f)[0]
pp = self.positions[indices] detections = get_detections(f, indices)
originaltracks = self.tracks[indices] done = [False, False]
dist_assignments = np.zeros_like(originaltracks) if len(detections) == 0:
angle_assignments = np.zeros_like(originaltracks)
length_assignments = np.zeros_like(originaltracks)
userlabeled = np.zeros_like(originaltracks)
distances = np.zeros((len(originaltracks), 2))
orientations = np.zeros_like(distances)
lengths = np.zeros_like(distances)
for i, (idx, p) in enumerate(zip(indices, pp)):
if self.userlabeled[idx]:
print("user")
userlabeled[i] = True
last_pos[originaltracks[i]-1] = pp[i]
last_frame[originaltracks[i]-1] = f
last_angle[originaltracks[i]-1] = self.orientations[idx]
last_length[originaltracks[i]-1] += ((self.lengths[idx] - last_length[originaltracks[i]-1]) / processed)
continue continue
dist_assignments[i], distances[i, :] = assign_by_distance(f, p)
angle_assignments[i], orientations[i,:] = assign_by_orientation(f, self.orientations[idx]) if len(detections) > 1 and np.any([detections[0].userlabeled, detections[1].userlabeled]):
length_assignments[i], lengths[i, :] = assign_by_length(self.lengths[idx]) # more than one detection
if np.any(userlabeled): if detections[0].userlabeled and detections[1].userlabeled:
if detections[0].track == detections[1].track:
error = True
logging.info("Classification error both detections in the same frame are assigned to the same track!")
elif detections[0].userlabeled and not detections[1].userlabeled:
detections[1].track = 1 if detections[0].track == 2 else 2
elif not detections[0].userlabeled and detections[1].userlabeled:
detections[0].track = 1 if detections[1].track == 2 else 2
if not error:
last_detections[detections[0].track] = detections[0]
last_detections[detections[1].track] = detections[1]
self.tracks[detections[0].id] = detections[0].track
self.tracks[detections[1].id] = detections[1].track
done[0] = True
done[1] = True
elif len(detections) == 1 and detections[0].userlabeled: # ony one detection and labeled
last_detections[detections[0].track] = detections[0]
done[0] = True
if np.sum(done) == len(detections):
continue continue
# check (re) assignment, update, and proceed # if f == 2088:
if not needs_checking(originaltracks, dist_assignments): # embed()
do_assignment(f, indices, dist_assignments) # return
if error and self._stoponerror:
self.signals.error.emit("Classification error both detections in the same frame are assigned to the same track!")
break
dist_assignments = np.zeros(2, dtype=int)
orientation_assignments = np.zeros_like(dist_assignments)
length_assignments = np.zeros_like(dist_assignments)
distances = np.zeros((2, 2))
orientations = np.zeros_like(distances)
lengths = np.zeros_like(distances)
assignments = np.zeros((2, 2))
for i, d in enumerate(detections):
dist_assignments[i], distances[i, :] = assign_by_distance(d)
orientation_assignments[i], orientations[i,:] = assign_by_orientation(d)
length_assignments[i], lengths[i, :] = assign_by_length(d)
assignments[i, :] = dist_assignments # (dist_assignments * 10 + orientation_assignments + length_assignments) / 3
diffs = np.diff(assignments, axis=1)
error = False
temp = {}
message = ""
for i, d in enumerate(detections):
temp = {}
if diffs[i] == 0: # both are equally likely
d.track = -1
error = True
message = "Classification error both detections in the same frame are assigned to the same track!"
break
if diffs[i] < 0:
d.track = 1
else: else:
if not (np.all(length_assignments == 1) or np.all(length_assignments == 2)): # if I find a solution by body length d.track = 2
logging.debug("frame %i: Decision based on body length", f) self.tracks[d.id] = d.track
do_assignment(f, indices, length_assignments) if d.track not in temp:
elif not (np.all(angle_assignments == 1) or np.all(angle_assignments == 2)): # else there is a solution based on orientation temp[d.track] = d
logging.info("frame %i: Decision based on orientation", f)
do_assignment(f, indices, angle_assignments)
else: else:
logging.info("frame %i: Cannot decide who is who") error = True
message = "Double assignment to the same track!"
break
if not error:
for k in temp:
last_detections[temp[k].track] = temp[k]
else:
logging.info("frame %i: Cannot decide who is who! %s", f, message)
for idx in indices: for idx in indices:
self.tracks[idx] = -1 self.tracks[idx] = -1
errors += 1 errors += 1
if self._stoponerror: if self._stoponerror:
self.signals.error.emit(message)
break break
processed += 1
if steps > 0 and f % steps == 0: if steps > 0 and f % steps == 0:
progress += 1 progress += 1
@ -486,17 +533,24 @@ class ConsistencyClassifier(QWidget):
self._all_scores = self._dataworker.scores self._all_scores = self._dataworker.scores
self._frames = self._dataworker.frames self._frames = self._dataworker.frames
self._tracks = self._dataworker.tracks self._tracks = self._dataworker.tracks
self._dataworker = None
if np.sum(self._userlabeled) < 1:
logging.error("ConsistencyTracker: I need at least 1 user-labeled frame to start with!")
self.setEnabled(False)
else:
t1_userlabeled = self._frames[self._userlabeled & (self._tracks == 1)]
t2_userlabeled = self._frames[self._userlabeled & (self._tracks == 2)]
max_startframe = np.min([t1_userlabeled[-1], t2_userlabeled[-1]])
min_startframe = np.max([t1_userlabeled[0], t2_userlabeled[0]])
self._maxframes = np.max(self._frames) self._maxframes = np.max(self._frames)
# FIXME the following line causes an error when there are no detections in the range
min_frame = max([self._frames[self._tracks == 1][0], self._frames[self._tracks == 2][0]]) + 1
self._maxframeslabel.setText(str(self._maxframes)) self._maxframeslabel.setText(str(self._maxframes))
self._startframe_spinner.setMinimum(min_frame) self._startframe_spinner.setMinimum(min_startframe)
self._startframe_spinner.setMaximum(self._frames[-1]) self._startframe_spinner.setMaximum(max_startframe)
self._startframe_spinner.setValue(self._frames[0] + 1) self._startframe_spinner.setValue(min_startframe)
self._startframe_spinner.setSingleStep(20)
self._startbtn.setEnabled(True) self._startbtn.setEnabled(True)
self._assignedlabel.setText("0") self._assignedlabel.setText("0")
self._errorlabel.setText("0") self._errorlabel.setText("0")
self._dataworker = None
self.setEnabled(True) self.setEnabled(True)
@Slot(float) @Slot(float)
@ -612,16 +666,14 @@ def main():
import pickle import pickle
from fixtracks.info import PACKAGE_ROOT from fixtracks.info import PACKAGE_ROOT
datafile = PACKAGE_ROOT / "data/merged2.pkl" datafile = PACKAGE_ROOT / "data/merged_small_beginning.pkl"
with open(datafile, "rb") as f: with open(datafile, "rb") as f:
df = pickle.load(f) df = pickle.load(f)
data = TrackingData() data = TrackingData(as_dict(df))
data.setData(as_dict(df))
coords = data.coordinates() coords = data.coordinates()
cogs = data.centerOfGravity() cogs = data.centerOfGravity()
userlabeled = data["userlabeled"] userlabeled = data["userlabeled"]
embed()
app = QApplication([]) app = QApplication([])
window = QWidget() window = QWidget()
window.setMinimumSize(200, 200) window.setMinimumSize(200, 200)