ysn-rfd/text-dataset-tiny-code-script-py-format

1import cv2
2import numpy as np
3import time
4from sklearn.neighbors import KNeighborsClassifier
5from collections import defaultdict, deque
6
7# Create background subtractor for motion detection
8back_sub = cv2.createBackgroundSubtractorKNN(history=500, dist2Threshold=400, detectShadows=True)
9cap = cv2.VideoCapture(0)
10
11# Store object traces
12object_traces = defaultdict(lambda: deque(maxlen=30))  # Last 30 points of each object
13object_last_seen = {}
14object_id_counter = 0
15
16# For real-time learning
17knn = KNeighborsClassifier(n_neighbors=3)
18features_set = []
19labels_set = []
20
21# Timer for real-time learning and training interval
22start_time = time.time()
23training_interval = 5  # 5 seconds for real-time training
24
25# Variable to avoid predicting before training
26is_trained = False
27
28# Memory storage for past predictions and features
29memory = defaultdict(list)  # Store memory of features and predictions for each object
30
31def apply_noise_reduction(mask):
32    kernel = cv2.getStructuringElement(cv2.MORPH_ELLIPSE, (5, 5))
33    mask = cv2.morphologyEx(mask, cv2.MORPH_OPEN, kernel, iterations=2)
34    mask = cv2.dilate(mask, kernel, iterations=1)
35    return mask
36
37def get_centroid(x, y, w, h):
38    return (int(x + w / 2), int(y + h / 2))
39
40def calculate_direction(trace):
41    if len(trace) < 2:
42        return "-"
43    dx = trace[-1][0] - trace[0][0]
44    dy = trace[-1][1] - trace[0][1]
45    if abs(dx) > abs(dy):
46        return "Left" if dx < 0 else "Right"
47    else:
48        return "Up" if dy < 0 else "Down"
49
50def calculate_speed(trace, duration):
51    if len(trace) < 2 or duration == 0:
52        return 0
53    dist = np.linalg.norm(np.array(trace[-1]) - np.array(trace[0]))
54    return dist / duration
55
56def count_direction_changes(trace):
57    changes = 0
58    for i in range(2, len(trace)):
59        dx1 = trace[i-1][0] - trace[i-2][0]
60        dx2 = trace[i][0] - trace[i-1][0]
61        if dx1 * dx2 < 0:  # Horizontal direction change
62            changes += 1
63    return changes
64
65while True:
66    ret, frame = cap.read()
67    if not ret:
68        break
69
70    gray = cv2.cvtColor(frame, cv2.COLOR_BGR2GRAY)
71    fg_mask = back_sub.apply(frame)
72    fg_mask = apply_noise_reduction(fg_mask)
73
74    contours, _ = cv2.findContours(fg_mask, cv2.RETR_EXTERNAL, cv2.CHAIN_APPROX_SIMPLE)
75
76    current_ids = []
77    predicted = 1  # Default prediction value (if no prediction is made)
78    for cnt in contours:
79        area = cv2.contourArea(cnt)
80        if area < 150:
81            continue
82
83        x, y, w, h = cv2.boundingRect(cnt)
84        centroid = get_centroid(x, y, w, h)
85
86        # Identify or create a new ID for the object
87        matched_id = None
88        for oid, trace in object_traces.items():
89            if np.linalg.norm(np.array(trace[-1]) - np.array(centroid)) < 50:
90                matched_id = oid
91                break
92
93        if matched_id is None:
94            matched_id = object_id_counter
95            object_id_counter += 1
96
97        object_traces[matched_id].append(centroid)
98        object_last_seen[matched_id] = time.time()
99        current_ids.append(matched_id)
100
101        trace = object_traces[matched_id]
102        duration = time.time() - object_last_seen[matched_id] + 0.001
103        speed = calculate_speed(trace, duration)
104        direction = calculate_direction(trace)
105        direction_changes = count_direction_changes(trace)
106        total_move = sum(np.linalg.norm(np.array(trace[i]) - np.array(trace[i-1])) for i in range(1, len(trace)))
107
108        # Feature for the model
109        feature = [w, h, centroid[0], centroid[1], area, speed, direction_changes]
110        label = 1  # Default label: Normal
111
112        # Simple automatic labeling:
113        if speed > 100 or direction_changes > 4:
114            label = 2  # Suspicious
115
116        features_set.append(feature)
117        labels_set.append(label)
118
119        # Store features and predictions in memory
120        memory[matched_id].append({
121            'features': feature,
122            'prediction': label
123        })
124
125        # Retrain the model every 5 seconds
126        if time.time() - start_time > training_interval:
127            if len(features_set) > 10:
128                knn.fit(features_set, labels_set)  # Train the model
129                is_trained = True  # Model is trained
130                print("Model updated.")
131                start_time = time.time()  # Reset the timer after retraining
132
133        # Prediction only after training
134        if is_trained:
135            predicted = knn.predict([feature])[0]
136
137        # Draw information on the frame
138        cv2.rectangle(frame, (x, y), (x + w, y + h), (0, 255, 0) if label == 1 else (0, 0, 255), 2)
139        cv2.circle(frame, centroid, 4, (255, 255, 255), -1)
140        cv2.putText(frame, f"ID: {matched_id} | Direction: {direction} | Speed: {int(speed)}", (x, y - 25),
141                    cv2.FONT_HERSHEY_SIMPLEX, 0.5, (255, 255, 0), 1)
142        cv2.putText(frame, f"Behavior: {'Normal' if predicted == 1 else 'Suspicious'}", (x, y - 5),
143                    cv2.FONT_HERSHEY_SIMPLEX, 0.5, (0, 255, 255), 1)
144
145    # Remove old object IDs from memory
146    for oid in list(object_last_seen):
147        if time.time() - object_last_seen[oid] > 2:
148            object_traces.pop(oid, None)
149            object_last_seen.pop(oid, None)
150            memory.pop(oid, None)  # Remove from memory as well
151
152    cv2.imshow("Behavioral Intelligence", frame)
153    if cv2.waitKey(1) & 0xFF == 27:
154        break
155
156cap.release()
157cv2.destroyAllWindows()
158