Update src/streamlit_app.py
Browse files- src/streamlit_app.py +43 -85
src/streamlit_app.py
CHANGED
@@ -1,41 +1,36 @@
|
|
1 |
import streamlit as st
|
2 |
-
import
|
3 |
-
import cv2 # OpenCV for image processing
|
4 |
import numpy as np
|
5 |
-
import mediapipe as mp
|
6 |
-
from
|
7 |
|
8 |
-
# Initialize MediaPipe
|
9 |
mp_pose = mp.solutions.pose
|
10 |
mp_drawing = mp.solutions.drawing_utils
|
11 |
|
12 |
-
#
|
|
|
|
|
|
|
|
|
|
|
|
|
13 |
if 'posture_status' not in st.session_state:
|
14 |
-
st.session_state.posture_status = "Please enable camera
|
15 |
|
16 |
def analyze_posture(image):
|
17 |
-
"""
|
18 |
-
Analyze posture using MediaPipe Pose
|
19 |
-
Args:
|
20 |
-
image: Input frame from camera
|
21 |
-
Returns:
|
22 |
-
annotated_image: Frame with pose landmarks drawn
|
23 |
-
posture_status: Analysis results text
|
24 |
-
"""
|
25 |
with mp_pose.Pose(
|
26 |
min_detection_confidence=0.5,
|
27 |
min_tracking_confidence=0.5,
|
28 |
model_complexity=1
|
29 |
) as pose:
|
30 |
|
31 |
-
# Convert color space and process frame
|
32 |
image_rgb = cv2.cvtColor(image, cv2.COLOR_BGR2RGB)
|
33 |
results = pose.process(image_rgb)
|
34 |
|
35 |
annotated_image = image.copy()
|
36 |
-
|
37 |
if results.pose_landmarks:
|
38 |
-
# Draw pose landmarks on the frame
|
39 |
mp_drawing.draw_landmarks(
|
40 |
annotated_image,
|
41 |
results.pose_landmarks,
|
@@ -45,22 +40,15 @@ def analyze_posture(image):
|
|
45 |
)
|
46 |
posture_status = check_posture(results.pose_landmarks, image.shape)
|
47 |
else:
|
48 |
-
posture_status = "No pose detected
|
49 |
|
50 |
return annotated_image, posture_status
|
51 |
|
52 |
def check_posture(landmarks, image_shape):
|
53 |
-
"""
|
54 |
-
Analyze body landmarks and generate posture report
|
55 |
-
Args:
|
56 |
-
landmarks: Detected pose landmarks
|
57 |
-
image_shape: Dimensions of input image
|
58 |
-
Returns:
|
59 |
-
Formatted posture analysis report
|
60 |
-
"""
|
61 |
h, w, _ = image_shape
|
62 |
|
63 |
-
# Get key
|
64 |
left_shoulder = landmarks.landmark[mp_pose.PoseLandmark.LEFT_SHOULDER]
|
65 |
right_shoulder = landmarks.landmark[mp_pose.PoseLandmark.RIGHT_SHOULDER]
|
66 |
left_hip = landmarks.landmark[mp_pose.PoseLandmark.LEFT_HIP]
|
@@ -69,18 +57,18 @@ def check_posture(landmarks, image_shape):
|
|
69 |
right_ear = landmarks.landmark[mp_pose.PoseLandmark.RIGHT_EAR]
|
70 |
nose = landmarks.landmark[mp_pose.PoseLandmark.NOSE]
|
71 |
|
72 |
-
# Determine
|
73 |
sitting = left_hip.y < left_shoulder.y + 0.1 or right_hip.y < right_shoulder.y + 0.1
|
74 |
|
75 |
messages = []
|
76 |
|
77 |
-
# Forward head
|
78 |
head_forward = (left_ear.y > left_shoulder.y + 0.1 or right_ear.y > right_shoulder.y + 0.1) and \
|
79 |
(nose.y > left_shoulder.y or nose.y > right_shoulder.y)
|
80 |
if head_forward:
|
81 |
-
messages.append("• Forward head tilt detected
|
82 |
|
83 |
-
#
|
84 |
shoulders_rounded = left_shoulder.x > left_hip.x + 0.05 or right_shoulder.x < right_hip.x - 0.05
|
85 |
if shoulders_rounded:
|
86 |
messages.append("• Rounded shoulders detected")
|
@@ -91,80 +79,50 @@ def check_posture(landmarks, image_shape):
|
|
91 |
if shoulder_diff > 0.05 or hip_diff > 0.05:
|
92 |
messages.append("• Body leaning to one side")
|
93 |
|
94 |
-
#
|
95 |
-
if sitting and (left_hip.y < left_shoulder.y + 0.15 or right_hip.y < right_shoulder.y + 0.15):
|
96 |
-
messages.append("• Pelvis tilted forward (sitting posture)")
|
97 |
-
|
98 |
-
# Generate final report
|
99 |
if messages:
|
100 |
report = [
|
101 |
-
f"**{'Sitting' if sitting else 'Standing'} posture issues
|
102 |
*messages,
|
103 |
"\n**Recommendations:**",
|
104 |
-
"• Keep head straight
|
105 |
-
"• Pull shoulders back
|
106 |
-
"• Maintain straight back
|
107 |
-
"•
|
108 |
]
|
109 |
else:
|
110 |
report = [
|
111 |
f"**Excellent {'sitting' if sitting else 'standing'} posture!**",
|
112 |
-
"All key points are properly aligned",
|
113 |
"\n**Tips:**",
|
114 |
-
"• Continue monitoring your posture
|
115 |
]
|
116 |
|
117 |
return "\n\n".join(report)
|
118 |
|
119 |
-
def video_frame_callback(frame):
|
120 |
-
"""
|
121 |
-
Callback function for processing each video frame
|
122 |
-
Args:
|
123 |
-
frame: Incoming video frame from WebRTC
|
124 |
-
Returns:
|
125 |
-
Processed video frame with pose landmarks
|
126 |
-
"""
|
127 |
-
img = frame.to_ndarray(format="bgr24")
|
128 |
-
|
129 |
-
try:
|
130 |
-
# Analyze posture and update session state
|
131 |
-
analyzed_img, posture_status = analyze_posture(img)
|
132 |
-
st.session_state.posture_status = posture_status
|
133 |
-
return av.VideoFrame.from_ndarray(analyzed_img, format="bgr24")
|
134 |
-
except Exception as e:
|
135 |
-
st.error(f"Processing error: {str(e)}")
|
136 |
-
return frame
|
137 |
-
|
138 |
def main():
|
139 |
-
# Configure Streamlit page
|
140 |
-
st.set_page_config(layout="wide")
|
141 |
-
st.title("📷 Real-time Posture Analysis")
|
142 |
-
|
143 |
-
# Create two-column layout
|
144 |
col1, col2 = st.columns([2, 1])
|
145 |
|
146 |
with col1:
|
147 |
-
st.header("Camera
|
148 |
|
149 |
-
#
|
150 |
-
|
151 |
-
|
152 |
-
|
153 |
-
video_frame_callback=video_frame_callback,
|
154 |
-
media_stream_constraints={"video": True, "audio": False},
|
155 |
-
async_processing=True,
|
156 |
-
rtc_configuration={
|
157 |
-
"iceServers": [{"urls": ["stun:stun.l.google.com:19302"]}]
|
158 |
-
}
|
159 |
-
)
|
160 |
|
161 |
-
|
162 |
-
|
163 |
-
|
164 |
-
st.
|
|
|
|
|
|
|
|
|
|
|
|
|
165 |
|
166 |
with col2:
|
167 |
-
st.header("
|
168 |
st.markdown(st.session_state.posture_status)
|
169 |
|
170 |
if __name__ == "__main__":
|
|
|
1 |
import streamlit as st
|
2 |
+
import cv2
|
|
|
3 |
import numpy as np
|
4 |
+
import mediapipe as mp
|
5 |
+
from PIL import Image
|
6 |
|
7 |
+
# Initialize MediaPipe
|
8 |
mp_pose = mp.solutions.pose
|
9 |
mp_drawing = mp.solutions.drawing_utils
|
10 |
|
11 |
+
# Configure page
|
12 |
+
st.set_page_config(layout="wide")
|
13 |
+
st.title("📷 Posture Analysis App")
|
14 |
+
|
15 |
+
# Initialize session state
|
16 |
+
if 'camera_on' not in st.session_state:
|
17 |
+
st.session_state.camera_on = False
|
18 |
if 'posture_status' not in st.session_state:
|
19 |
+
st.session_state.posture_status = "Please enable camera"
|
20 |
|
21 |
def analyze_posture(image):
|
22 |
+
"""Analyze posture using MediaPipe"""
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
23 |
with mp_pose.Pose(
|
24 |
min_detection_confidence=0.5,
|
25 |
min_tracking_confidence=0.5,
|
26 |
model_complexity=1
|
27 |
) as pose:
|
28 |
|
|
|
29 |
image_rgb = cv2.cvtColor(image, cv2.COLOR_BGR2RGB)
|
30 |
results = pose.process(image_rgb)
|
31 |
|
32 |
annotated_image = image.copy()
|
|
|
33 |
if results.pose_landmarks:
|
|
|
34 |
mp_drawing.draw_landmarks(
|
35 |
annotated_image,
|
36 |
results.pose_landmarks,
|
|
|
40 |
)
|
41 |
posture_status = check_posture(results.pose_landmarks, image.shape)
|
42 |
else:
|
43 |
+
posture_status = "No pose detected"
|
44 |
|
45 |
return annotated_image, posture_status
|
46 |
|
47 |
def check_posture(landmarks, image_shape):
|
48 |
+
"""Analyze body landmarks and generate posture report"""
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
49 |
h, w, _ = image_shape
|
50 |
|
51 |
+
# Get key points
|
52 |
left_shoulder = landmarks.landmark[mp_pose.PoseLandmark.LEFT_SHOULDER]
|
53 |
right_shoulder = landmarks.landmark[mp_pose.PoseLandmark.RIGHT_SHOULDER]
|
54 |
left_hip = landmarks.landmark[mp_pose.PoseLandmark.LEFT_HIP]
|
|
|
57 |
right_ear = landmarks.landmark[mp_pose.PoseLandmark.RIGHT_EAR]
|
58 |
nose = landmarks.landmark[mp_pose.PoseLandmark.NOSE]
|
59 |
|
60 |
+
# Determine posture
|
61 |
sitting = left_hip.y < left_shoulder.y + 0.1 or right_hip.y < right_shoulder.y + 0.1
|
62 |
|
63 |
messages = []
|
64 |
|
65 |
+
# Forward head check
|
66 |
head_forward = (left_ear.y > left_shoulder.y + 0.1 or right_ear.y > right_shoulder.y + 0.1) and \
|
67 |
(nose.y > left_shoulder.y or nose.y > right_shoulder.y)
|
68 |
if head_forward:
|
69 |
+
messages.append("• Forward head tilt detected")
|
70 |
|
71 |
+
# Shoulders check
|
72 |
shoulders_rounded = left_shoulder.x > left_hip.x + 0.05 or right_shoulder.x < right_hip.x - 0.05
|
73 |
if shoulders_rounded:
|
74 |
messages.append("• Rounded shoulders detected")
|
|
|
79 |
if shoulder_diff > 0.05 or hip_diff > 0.05:
|
80 |
messages.append("• Body leaning to one side")
|
81 |
|
82 |
+
# Generate report
|
|
|
|
|
|
|
|
|
83 |
if messages:
|
84 |
report = [
|
85 |
+
f"**{'Sitting' if sitting else 'Standing'} posture issues:**",
|
86 |
*messages,
|
87 |
"\n**Recommendations:**",
|
88 |
+
"• Keep head straight",
|
89 |
+
"• Pull shoulders back",
|
90 |
+
"• Maintain straight back",
|
91 |
+
"• Sit on sitting bones"
|
92 |
]
|
93 |
else:
|
94 |
report = [
|
95 |
f"**Excellent {'sitting' if sitting else 'standing'} posture!**",
|
|
|
96 |
"\n**Tips:**",
|
97 |
+
"• Continue monitoring your posture"
|
98 |
]
|
99 |
|
100 |
return "\n\n".join(report)
|
101 |
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
102 |
def main():
|
|
|
|
|
|
|
|
|
|
|
103 |
col1, col2 = st.columns([2, 1])
|
104 |
|
105 |
with col1:
|
106 |
+
st.header("Camera View")
|
107 |
|
108 |
+
# Camera toggle
|
109 |
+
if st.button("Enable Camera" if not st.session_state.camera_on else "Disable Camera"):
|
110 |
+
st.session_state.camera_on = not st.session_state.camera_on
|
111 |
+
st.rerun()
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
112 |
|
113 |
+
if st.session_state.camera_on:
|
114 |
+
# Placeholder for local camera implementation
|
115 |
+
demo_image = np.zeros((480, 640, 3), dtype=np.uint8)
|
116 |
+
st.image(demo_image, caption="Camera feed placeholder")
|
117 |
+
st.warning("For full camera functionality, please run locally")
|
118 |
+
|
119 |
+
# Simulate analysis
|
120 |
+
if st.button("Analyze Current Frame"):
|
121 |
+
_, posture_status = analyze_posture(demo_image)
|
122 |
+
st.session_state.posture_status = posture_status
|
123 |
|
124 |
with col2:
|
125 |
+
st.header("Analysis Results")
|
126 |
st.markdown(st.session_state.posture_status)
|
127 |
|
128 |
if __name__ == "__main__":
|