PunchSense is a wearable punch evaluation system built with an M5Stack Core2 and an IMU Pro Mini Unit. It measures acceleration and angular velocity during punching motions and provides immediate feedback through a score, an S-to-D rank, a gauge, a facial expression, and a short comment on the M5Stack display.
This system evaluates motion characteristics rather than directly measuring impact force.
Demonstration of the IMU mounted on the boxing glove and real-time personalized feedback displayed on the M5Stack Core2.
Why I Built It
A large acceleration value alone does not always represent a good punch. During early experiments, a short wrist-snapping motion could receive a high score even when the arm was not extended properly.
I therefore wanted to evaluate not only acceleration, but also fist rotation, acceleration rise, and the coordination between translational motion and rotation. I also wanted the evaluation to reflect the user's own perception of a good punch.
System Overview
The IMU Pro Mini Unit is attached firmly to the glove and connected to the M5Stack Core2 through an I2C Grove cable. The BMI270 sensor measures three-axis acceleration and angular velocity at a target sampling rate of 100 Hz.
The M5Stack detects each punch, calculates motion features, and displays the result immediately. After the calibrated parameters are embedded in the firmware, the system can perform scoring without a connected computer.
Punch Detection
The magnitude of acceleration is calculated as:
|a| = sqrt(ax^2 + ay^2 + az^2)
The dynamic acceleration is calculated by removing the effect of approximately 1 g of gravity:
dynamic = ||a| - 1 g|
A punch candidate is detected when the dynamic acceleration reaches a local peak of at least 1.5 g. Consecutive detected peaks must also be separated by at least 0.6 seconds.
Evaluation Features
Each detected punch is evaluated using the following features:
Dynamic acceleration peak
Acceleration rise before the peak
Peak fist rotation speed
Timing relationship between acceleration and rotation
Ratio and balance between acceleration and rotation
Coordination of acceleration, rotation, and timing
These features allow PunchSense to evaluate the overall motion pattern instead of relying only on acceleration.
Experimental Punch-Arrival Estimation
After each acceleration peak, the program analyzes the rapid decrease in dynamic acceleration. The black triangles indicate the post-peak dynamic minima, and the blue diamonds indicate the estimated punch-arrival timing.
This arrival proxy is intended to approximate the timing at which the arm reaches extension or the punch motion changes rapidly. However, it has not yet been validated against an external reference such as a high-speed camera or an impact sensor. Therefore, it is currently treated as an experimental feature rather than an exact measurement.
Measured acceleration and angular velocity waveforms. Red circles show detected punch peaks, black triangles show post-peak dynamic minima, and blue diamonds show experimental punch-arrival estimates.
Personalized Scoring
The Python script scripts/read_imu.py records the IMU data and saves the measurements as CSV files and waveform images.
The user then assigns a subjective score from 1 to 5 to each detected punch. The user_label and comment columns are also used to describe the punch, while user_score is used as the numerical calibration target.
The script scripts/calibrate_subjective_score.py calculates the relationship between each motion feature and the subjective scores. Pearson correlation coefficients are used to derive the feature weights, and the resulting personalized model is saved as subjective_score_config.json.
The calibrated model produces a learned score from 0 to 100. The score is converted into five ranks:
S: 85-100
A: 70-84.9
B: 50-69.9
C: 30-49.9
D: below 30
Real-Time Feedback
After a punch is detected, the M5Stack Core2 displays:
Personalized score
S-to-D rank
Score gauge
Acceleration in m/s^2
Rank-dependent facial expression
Short Japanese feedback comment
The display gives the user immediate and intuitive feedback without requiring them to inspect raw sensor data.
Software
The M5Stack firmware was developed using PlatformIO and the Arduino framework. The project uses M5Unified and the M5Stack IMU Pro Mini library.
Python is used for serial data recording, waveform visualization, subjective-score calibration, and real-time analysis during development.
Build Instructions
1. Connect the IMU Pro Mini Unit to Port A of the M5Stack Core2 using a Grove cable.
2. Secure the IMU firmly to the glove so that its position and orientation do not change during punching motions.
3. Open the project in Visual Studio Code with PlatformIO installed.
4. Build and upload the firmware using the m5stack-core2 environment defined in platformio.ini.
5. Upload the SPIFFS filesystem image to install the S-to-D facial-expression images.
6. Power on the Core2 and wait for the IMU initialization to complete.
7. Perform a punch. After detecting the acceleration peak, the Core2 calculates the motion features and displays the score, rank, gauge, acceleration, facial expression, and feedback comment.
For data collection and model calibration, install the Python dependencies and run:
pip install -r scripts/requirements.txt
python scripts/read_imu.py
Add a user_score from 1 to 5 to each detected punch in the generated summary CSV, then run:
PunchSense can detect punch motions and visualize clear acceleration and rotation peaks. Testing different sampling-rate settings showed that 100 Hz was a practical operating point for this implementation.
The next steps are to collect more labeled punch data, improve the personalized scoring accuracy, validate the estimated punch-arrival timing using an external reference such as video or an impact sensor, and improve the method used to secure the sensor to the glove.
Safety
This project is intended for motion analysis and training support. It does not directly measure impact force and should not be used as a medical or professional safety device.
Records IMU data and saves CSV files and waveform images.
#!/usr/bin/env python3"""Receive IMU CSV data over serial, then save CSV and waveform on Ctrl+C."""from__future__importannotationsimportargparseimportcsvimportjsonimportsysimporttimefromdataclassesimportdataclassfromdatetimeimportdatetimefrompathlibimportPathtry:importserialfromserial.toolsimportlist_portsexceptImportErrorasexc:raiseSystemExit("pyserial is required. Install it with: python -m pip install pyserial")fromexctry:importmatplotlib.pyplotaspltexceptImportErrorasexc:raiseSystemExit("matplotlib is required. Install it with: python -m pip install matplotlib")fromexcfromcalibrate_subjective_scoreimportpredictplt.rcParams["font.family"]=["Yu Gothic","Meiryo","MS Gothic","Noto Sans CJK JP","sans-serif",]plt.rcParams["axes.unicode_minus"]=FalseFIELDS=["sample_index","device_ms","time_sec","ax_g","ay_g","az_g","mag_g","dynamic_g","gx_dps","gy_dps","gz_dps","rotation_dps",]@dataclassclassSample:sample_index:intdevice_ms:inttime_sec:floatax_g:floatay_g:floataz_g:floatmag_g:floatdynamic_g:floatgx_dps:float=0.0gy_dps:float=0.0gz_dps:float=0.0rotation_dps:float=0.0defas_row(self)->dict[str,int|float]:return{"sample_index":self.sample_index,"device_ms":self.device_ms,"time_sec":self.time_sec,"ax_g":self.ax_g,"ay_g":self.ay_g,"az_g":self.az_g,"mag_g":self.mag_g,"dynamic_g":self.dynamic_g,"gx_dps":self.gx_dps,"gy_dps":self.gy_dps,"gz_dps":self.gz_dps,"rotation_dps":self.rotation_dps,}@dataclassclassPeak:sample:Samplevalue:float@dataclassclassPunchAnalysis:number:intpeak:Peakaccel_rise_g:floatrotation_peak_dps:floatrotation_time_sec:floattiming_offset_sec:floatdecel_time_sec:floatdecel_dynamic_g:floatdecel_offset_sec:floatarrival_time_sec:floatarrival_dynamic_g:floatarrival_offset_sec:floatscore:floatdefauto_detect_port()->str|None:ports=list(list_ports.comports())ifnotports:returnNoneusb_ports=[portforportinportsif"USB"in(port.descriptionor"").upper()or"UART"in(port.descriptionor"").upper()or"SERIAL"in(port.descriptionor"").upper()]return(usb_ports[0]ifusb_portselseports[0]).devicedefparse_sample(line:str,first_device_ms:int|None)->tuple[Sample|None,int|None]:text=line.strip()ifnottextortext.startswith("#")ortext.startswith("CSV_HEADER"):returnNone,first_device_msparts=text.split(",")iflen(parts)notin(7,11):returnNone,first_device_mstry:sample_index=int(parts[0])device_ms=int(parts[1])ax_g=float(parts[2])ay_g=float(parts[3])az_g=float(parts[4])mag_g=float(parts[5])dynamic_g=float(parts[6])gx_dps=float(parts[7])iflen(parts)==11else0.0gy_dps=float(parts[8])iflen(parts)==11else0.0gz_dps=float(parts[9])iflen(parts)==11else0.0rotation_dps=float(parts[10])iflen(parts)==11else0.0exceptValueError:returnNone,first_device_msiffirst_device_msisNone:first_device_ms=device_mssample=Sample(sample_index=sample_index,device_ms=device_ms,time_sec=(device_ms-first_device_ms)/1000.0,ax_g=ax_g,ay_g=ay_g,az_g=az_g,mag_g=mag_g,dynamic_g=dynamic_g,gx_dps=gx_dps,gy_dps=gy_dps,gz_dps=gz_dps,rotation_dps=rotation_dps,)returnsample,first_device_msdefsample_value(sample:Sample,field:str)->float:returnfloat(getattr(sample,field))defdetect_peaks(samples:list[Sample],field:str,threshold:float,min_interval_sec:float,max_peaks:int,max_neighbor_gap_sec:float,)->list[Peak]:iflen(samples)<3:return[]candidates:list[Peak]=[]forindexinrange(1,len(samples)-1):prev_gap=samples[index].time_sec-samples[index-1].time_secnext_gap=samples[index+1].time_sec-samples[index].time_secprev_value=(sample_value(samples[index-1],field)ifprev_gap<=max_neighbor_gap_secelsefloat("-inf"))value=sample_value(samples[index],field)next_value=(sample_value(samples[index+1],field)ifnext_gap<=max_neighbor_gap_secelsefloat("-inf"))ifvalue>=thresholdandvalue>=prev_valueandvalue>=next_value:candidates.append(Peak(samples[index],value))selected:list[Peak]=[]forpeakinsorted(candidates,key=lambdaitem:item.value,reverse=True):ifall(abs(peak.sample.time_sec-chosen.sample.time_sec)>=min_interval_secforchoseninselected):selected.append(peak)iflen(selected)>=max_peaks:breakreturnsorted(selected,key=lambdaitem:item.sample.time_sec)defwindow_samples(samples:list[Sample],center_sec:float,before_sec:float,after_sec:float,)->list[Sample]:start=center_sec-before_secend=center_sec+after_secreturn[sampleforsampleinsamplesifstart<=sample.time_sec<=end]defclamp(value:float,low:float,high:float)->float:returnmax(low,min(value,high))defload_subjective_model(model_path:Path)->dict|None:ifnotmodel_path.exists():returnNonereturnjson.loads(model_path.read_text(encoding="utf-8-sig"))defanalysis_row(analysis:PunchAnalysis)->dict[str,str]:return{"score":f"{analysis.score}","dynamic_peak_g":f"{analysis.peak.value}","accel_rise_g":f"{analysis.accel_rise_g}","rotation_peak_dps":f"{analysis.rotation_peak_dps}","timing_offset_sec":f"{analysis.timing_offset_sec}","decel_offset_sec":f"{analysis.decel_offset_sec}","arrival_offset_sec":f"{analysis.arrival_offset_sec}",}defdisplay_score(analysis:PunchAnalysis,model:dict|None)->float:ifmodelisNone:returnanalysis.scorereturnpredict(analysis_row(analysis),model)defanalyze_punches(samples:list[Sample],peaks:list[Peak],rotation_field:str,)->list[PunchAnalysis]:analyses:list[PunchAnalysis]=[]fornumber,peakinenumerate(peaks,start=1):pre_samples=window_samples(samples,peak.sample.time_sec,0.25,0.0)impact_samples=window_samples(samples,peak.sample.time_sec,0.25,0.08)baseline_dynamic=min((sample.dynamic_gforsampleinpre_samples),default=peak.value,)accel_rise_g=max(0.0,peak.value-baseline_dynamic)rotation_sample=max(impact_samples,key=lambdasample:abs(sample_value(sample,rotation_field)),default=peak.sample,)rotation_peak_dps=abs(sample_value(rotation_sample,rotation_field))timing_offset_sec=rotation_sample.time_sec-peak.sample.time_secdecel_start_sec=peak.sample.time_sec+0.04decel_end_sec=peak.sample.time_sec+0.20decel_samples=[sampleforsampleinsamplesifdecel_start_sec<=sample.time_sec<=decel_end_sec]decel_sample=min(decel_samples,key=lambdasample:sample.dynamic_g,default=peak.sample,)decel_offset_sec=rotation_sample.time_sec-decel_sample.time_secarrival_sample=decel_samplesteepest_drop=0.0forprev_sample,current_sampleinzip(samples,samples[1:]):ifnot(peak.sample.time_sec+0.02<=current_sample.time_sec<=peak.sample.time_sec+0.20):continuedt=current_sample.time_sec-prev_sample.time_secifdt<=0:continueslope=(current_sample.dynamic_g-prev_sample.dynamic_g)/dtifslope<steepest_drop:steepest_drop=slopearrival_sample=current_samplearrival_offset_sec=rotation_sample.time_sec-arrival_sample.time_secaccel_score=clamp(peak.value/2.0,0.0,1.0)*40.0rise_score=clamp(accel_rise_g/1.0,0.0,1.0)*25.0rotation_score=clamp(rotation_peak_dps/800.0,0.0,1.0)*25.0timing_score=clamp(1.0-abs(arrival_offset_sec)/0.20,0.0,1.0)*10.0score=accel_score+rise_score+rotation_score+timing_scoreanalyses.append(PunchAnalysis(number=number,peak=peak,accel_rise_g=accel_rise_g,rotation_peak_dps=rotation_peak_dps,rotation_time_sec=rotation_sample.time_sec,timing_offset_sec=timing_offset_sec,decel_time_sec=decel_sample.time_sec,decel_dynamic_g=decel_sample.dynamic_g,decel_offset_sec=decel_offset_sec,arrival_time_sec=arrival_sample.time_sec,arrival_dynamic_g=arrival_sample.dynamic_g,arrival_offset_sec=arrival_offset_sec,score=score,))returnanalysesdefload_csv(csv_path:Path)->list[Sample]:samples:list[Sample]=[]withcsv_path.open(newline="",encoding="utf-8")asfp:reader=csv.DictReader(fp)forrowinreader:sample=Sample(sample_index=int(row["sample_index"]),device_ms=int(row["device_ms"]),time_sec=float(row["time_sec"]),ax_g=float(row["ax_g"]),ay_g=float(row["ay_g"]),az_g=float(row["az_g"]),mag_g=float(row["mag_g"]),dynamic_g=float(row["dynamic_g"]),gx_dps=float(row.get("gx_dps")or0.0),gy_dps=float(row.get("gy_dps")or0.0),gz_dps=float(row.get("gz_dps")or0.0),rotation_dps=float(row.get("rotation_dps")or0.0),)samples.append(sample)returnsamplesdefprepare_live_plot()->tuple[plt.Figure,plt.Axes]:plt.ion()fig,ax=plt.subplots(figsize=(10,5))fig.canvas.manager.set_window_title("IMU")ax.set_title("IMU",fontsize=16)ax.set_xlabel(" [s]",fontsize=13)ax.set_ylabel(" [g]",fontsize=13)ax.grid(True,alpha=0.3)returnfig,axdefupdate_plot(ax:plt.Axes,samples:list[Sample],window_sec:float)->None:ifnotsamples:returnlatest_time=samples[-1].time_secstart_time=max(0.0,latest_time-window_sec)visible=[sampleforsampleinsamplesifsample.time_sec>=start_time]t=[sample.time_secforsampleinvisible]ax.clear()ax.plot(t,[sample.dynamic_gforsampleinvisible],label="dynamic",linewidth=2.5)ax.plot(t,[sample.rotation_dps/1000.0forsampleinvisible],label="rotation / 1000",linewidth=2)ax.plot(t,[sample.ax_gforsampleinvisible],label="ax",alpha=0.25)ax.plot(t,[sample.ay_gforsampleinvisible],label="ay",alpha=0.25)ax.plot(t,[sample.az_gforsampleinvisible],label="az",alpha=0.25)ax.set_title("IMU",fontsize=16)ax.set_xlabel(" [s]",fontsize=13)ax.set_ylabel(" [g]",fontsize=13)ax.tick_params(axis="both",labelsize=11)ax.grid(True,alpha=0.3)ax.legend(loc="upper right")ax.figure.tight_layout()ax.figure.canvas.draw_idle()plt.pause(0.001)defsave_csv(samples:list[Sample],csv_path:Path)->None:csv_path.parent.mkdir(parents=True,exist_ok=True)withcsv_path.open("w",newline="",encoding="utf-8")asfp:writer=csv.DictWriter(fp,fieldnames=FIELDS)writer.writeheader()writer.writerows(sample.as_row()forsampleinsamples)defsave_plot(samples:list[Sample],image_path:Path,peak_field:str,peak_threshold:float,peak_min_interval_sec:float,max_peaks:int,rotation_field:str,peak_neighbor_gap_sec:float,)->list[PunchAnalysis]:image_path.parent.mkdir(parents=True,exist_ok=True)fig,(ax_accel,ax_rotation)=plt.subplots(2,1,figsize=(12,8),sharex=True,gridspec_kw={"height_ratios":[2,1]},)t=[sample.time_secforsampleinsamples]peaks=detect_peaks(samples,peak_field,peak_threshold,peak_min_interval_sec,max_peaks,peak_neighbor_gap_sec,)analyses=analyze_punches(samples,peaks,rotation_field)ax_accel.plot(t,[sample.ax_gforsampleinsamples],label="ax",alpha=0.25)ax_accel.plot(t,[sample.ay_gforsampleinsamples],label="ay",alpha=0.25)ax_accel.plot(t,[sample.az_gforsampleinsamples],label="az",alpha=0.25)ax_accel.plot(t,[sample.mag_gforsampleinsamples],label="|a|",color="0.35",alpha=0.55)ax_accel.plot(t,[sample.dynamic_gforsampleinsamples],label="dynamic",color="tab:purple",linewidth=2.5,)ax_accel.axhline(peak_threshold,color="tab:purple",linestyle=":",linewidth=1,alpha=0.8,label=f"peak threshold ({peak_threshold:g}g)",)ifpeaks:peak_t=[peak.sample.time_secforpeakinpeaks]peak_y=[peak.valueforpeakinpeaks]ax_accel.scatter(peak_t,peak_y,s=80,color="tab:red",edgecolors="white",linewidths=1.5,zorder=5,label="punch peak",)ax_accel.scatter([analysis.decel_time_secforanalysisinanalyses],[analysis.decel_dynamic_gforanalysisinanalyses],s=55,color="black",marker="v",edgecolors="white",linewidths=1.0,zorder=5,label="post-peak dynamic min",)ax_accel.scatter([analysis.arrival_time_secforanalysisinanalyses],[analysis.arrival_dynamic_gforanalysisinanalyses],s=60,color="tab:blue",marker="D",edgecolors="white",linewidths=1.0,zorder=6,label="arrival proxy",)foranalysisinanalyses:ax_accel.annotate(f"{analysis.number}",xy=(analysis.peak.sample.time_sec,analysis.peak.value),xytext=(0,12),textcoords="offset points",ha="center",va="bottom",fontsize=9,color="tab:red",)rotation_values=[abs(sample_value(sample,rotation_field))forsampleinsamples]ax_rotation.plot(t,rotation_values,label=rotation_field,color="tab:orange",linewidth=2,)ifanalyses:ax_rotation.scatter([analysis.rotation_time_secforanalysisinanalyses],[analysis.rotation_peak_dpsforanalysisinanalyses],s=60,color="tab:red",edgecolors="white",linewidths=1.2,zorder=5,label="rotation peak",)ax_accel.set_title("IMU",fontsize=18)ax_accel.set_ylabel(" [g]",fontsize=14)ax_accel.grid(True,alpha=0.3)ax_accel.legend(loc="upper right")ax_rotation.set_xlabel(" [s]",fontsize=14)ax_rotation.set_ylabel(" [deg/s]",fontsize=14)ax_accel.tick_params(axis="both",labelsize=12)ax_rotation.tick_params(axis="both",labelsize=12)ax_rotation.grid(True,alpha=0.3)ax_rotation.legend(loc="upper right")fig.tight_layout()fig.savefig(image_path,dpi=150)plt.close(fig)returnanalysesdefbuild_paths(output_dir:Path)->tuple[Path,Path]:stamp=datetime.now().strftime("%Y%m%d_%H%M%S")returnoutput_dir/f"imu_{stamp}.csv",output_dir/f"imu_{stamp}.png"defprint_analysis(analyses:list[PunchAnalysis],rotation_field:str,model:dict|None)->None:ifnotanalyses:print("Marked 0 punch peaks")returnprint(f"Marked {len(analyses)} punch peaks")print("No. time[s] score old dynamic[g] rise[g] rotation[dps] ""peak_offset[s] decel_offset[s] arrival_offset[s]")foranalysisinanalyses:score=display_score(analysis,model)print(f"{analysis.number:>2} "f"{analysis.peak.sample.time_sec:>7.3f} "f"{score:>5.1f} "f"{analysis.score:>5.1f} "f"{analysis.peak.value:>10.3f} "f"{analysis.accel_rise_g:>7.3f} "f"{analysis.rotation_peak_dps:>13.1f} "f"{analysis.timing_offset_sec:>14.3f} "f"{analysis.decel_offset_sec:>15.3f} "f"{analysis.arrival_offset_sec:>17.3f}")print(f"Rotation field: {rotation_field}")defsave_analysis_csv(analyses:list[PunchAnalysis],csv_path:Path,model:dict|None)->None:csv_path.parent.mkdir(parents=True,exist_ok=True)withcsv_path.open("w",newline="",encoding="utf-8")asfp:fieldnames=["punch_number","time_sec","score","old_score","dynamic_peak_g","accel_rise_g","rotation_peak_dps","rotation_time_sec","timing_offset_sec","decel_time_sec","decel_dynamic_g","decel_offset_sec","arrival_time_sec","arrival_dynamic_g","arrival_offset_sec",]writer=csv.DictWriter(fp,fieldnames=fieldnames)writer.writeheader()foranalysisinanalyses:score=display_score(analysis,model)writer.writerow({"punch_number":analysis.number,"time_sec":analysis.peak.sample.time_sec,"score":score,"old_score":analysis.score,"dynamic_peak_g":analysis.peak.value,"accel_rise_g":analysis.accel_rise_g,"rotation_peak_dps":analysis.rotation_peak_dps,"rotation_time_sec":analysis.rotation_time_sec,"timing_offset_sec":analysis.timing_offset_sec,"decel_time_sec":analysis.decel_time_sec,"decel_dynamic_g":analysis.decel_dynamic_g,"decel_offset_sec":analysis.decel_offset_sec,"arrival_time_sec":analysis.arrival_time_sec,"arrival_dynamic_g":analysis.arrival_dynamic_g,"arrival_offset_sec":analysis.arrival_offset_sec,})defprint_sample_rate_warning(samples:list[Sample],expected_hz:float)->None:iflen(samples)<2:returnduration_sec=samples[-1].time_sec-samples[0].time_secifduration_sec<=0:returnactual_hz=(len(samples)-1)/duration_secprint(f"Recorded sample rate: {actual_hz:.1f} Hz ({len(samples)} samples)")ifactual_hz<expected_hz*0.5:print("Warning: sample rate is very low. Punch peaks may be missed. ""Close other serial monitors and upload the latest firmware.")defmain()->int:parser=argparse.ArgumentParser(description="Read IMU serial CSV from M5Stack and save CSV/plot on Ctrl+C.")parser.add_argument("-p","--port",help="Serial port, for example COM3")parser.add_argument("-b","--baud",type=int,default=921600,help="Serial baud rate")parser.add_argument("-o","--output-dir",default="data",help="Output directory")parser.add_argument("--from-csv",type=Path,help="Create a peak-marked PNG from an existing CSV instead of recording serial data",)parser.add_argument("--image-path",type=Path,help="Output PNG path when using --from-csv",)parser.add_argument("--model",type=Path,default=Path("data/subjective_score_config.json"),help="Subjective score model used for the summary score column",)parser.add_argument("--peak-field",choices=["dynamic_g","mag_g"],default="dynamic_g",help="Signal used for punch peak detection",)parser.add_argument("--rotation-field",choices=["gx_dps","gy_dps","gz_dps","rotation_dps"],default="rotation_dps",help="Gyro signal used for fist rotation evaluation",)parser.add_argument("--peak-threshold",type=float,default=1.5,help="Minimum peak value in g",)parser.add_argument("--peak-min-interval-sec",type=float,default=0.6,help="Minimum time between separate punch peaks",)parser.add_argument("--max-peaks",type=int,default=30,help="Maximum number of peaks to mark",)parser.add_argument("--peak-neighbor-gap-sec",type=float,default=0.25,help="Do not compare peak neighbors across larger time gaps",)parser.add_argument("--expected-hz",type=float,default=100.0,help="Expected sensor sample rate for warning messages",)parser.add_argument("--window-sec",type=float,default=10.0,help="Live plot time window in seconds",)parser.add_argument("--plot-interval",type=float,default=0.2,help="Live plot refresh interval in seconds",)parser.add_argument("--live-plot",action="store_true",help="Show a live plot while recording. This can reduce the serial sample rate.",)parser.add_argument("--no-live-plot",action="store_true",help="Deprecated; live plot is disabled by default",)parser.add_argument("--no-data-warning-sec",type=float,default=5.0,help="Seconds before warning that no valid sample has arrived",)parser.add_argument("--show-raw",action="store_true",help="Print every non-sample serial line for troubleshooting",)parser.add_argument("--show-invalid",action="store_true",help="Print serial lines that are not valid CSV samples",)args=parser.parse_args()model=load_subjective_model(args.model)ifargs.from_csvisnotNone:samples=load_csv(args.from_csv)ifnotsamples:print(f"No samples found in {args.from_csv}",file=sys.stderr)return1print_sample_rate_warning(samples,args.expected_hz)image_path=args.image_pathorargs.from_csv.with_name(f"{args.from_csv.stem}_peaks.png")analyses=save_plot(samples,image_path,args.peak_field,args.peak_threshold,args.peak_min_interval_sec,args.max_peaks,args.rotation_field,args.peak_neighbor_gap_sec,)print(f"Saved punch evaluation plot to {image_path}")summary_path=image_path.with_name(f"{image_path.stem}_summary.csv")save_analysis_csv(analyses,summary_path,model)print(f"Saved punch summary to {summary_path}")print_analysis(analyses,args.rotation_field,model)return0port=args.portorauto_detect_port()ifnotport:print("No serial port found. Specify one with --port COMx.",file=sys.stderr)return2output_dir=Path(args.output_dir)csv_path,image_path=build_paths(output_dir)samples:list[Sample]=[]first_device_ms:int|None=Nonelast_plot_time=0.0last_status_time=0.0last_warning_time=time.monotonic()fig=Noneax=Noneifargs.live_plotandnotargs.no_live_plot:fig,ax=prepare_live_plot()print(f"Opening {port} at {args.baud} bps.")print("Recording IMU data. Press Ctrl+C to save CSV and waveform PNG.")try:withserial.Serial(port,args.baud,timeout=1)asser:whileTrue:raw_line=ser.readline()ifnotraw_line:now=time.monotonic()if(notsamplesandnow-last_warning_time>=args.no_data_warning_sec):print("\nNo valid samples yet. Check that the sensor is connected, ""the firmware is uploaded, and no other Serial Monitor is open.")last_warning_time=nowcontinueline=raw_line.decode("utf-8",errors="replace")sample,first_device_ms=parse_sample(line,first_device_ms)ifsampleisNone:text=line.strip()iftextand(args.show_raworargs.show_invalidortext.startswith("#")ortext.startswith("CSV_HEADER")):print(text)continuesamples.append(sample)last_warning_time=time.monotonic()now=time.monotonic()ifnow-last_status_time>=0.5:print(f"\rSamples: {len(samples)} "f"t={sample.time_sec:8.3f}s "f"ax={sample.ax_g: .3f} ay={sample.ay_g: .3f} az={sample.az_g: .3f}",end="",flush=True,)last_status_time=nowifaxisnotNoneandnow-last_plot_time>=args.plot_interval:update_plot(ax,samples,args.window_sec)last_plot_time=nowexceptKeyboardInterrupt:print()finally:iffigisnotNone:plt.close(fig)ifnotsamples:print("No samples were recorded; nothing was saved.")print("Try: python scripts/read_imu.py --baud 115200 --show-raw\n""If that shows CSV data, upload the latest firmware or keep using 115200.")return1print_sample_rate_warning(samples,args.expected_hz)save_csv(samples,csv_path)analyses=save_plot(samples,image_path,args.peak_field,args.peak_threshold,args.peak_min_interval_sec,args.max_peaks,args.rotation_field,args.peak_neighbor_gap_sec,)print(f"Saved {len(samples)} samples to {csv_path}")print(f"Saved waveform plot to {image_path}")summary_path=image_path.with_name(f"{image_path.stem}_summary.csv")save_analysis_csv(analyses,summary_path,model)print(f"Saved punch summary to {summary_path}")print_analysis(analyses,args.rotation_field,model)return0if__name__=="__main__":raiseSystemExit(main())
Personalized Score Calibration
Python
Builds a personalized scoring model from user-labeled punch data.
#!/usr/bin/env python3"""Build and apply a simple personal punch score from labeled summaries."""from__future__importannotationsimportargparseimportcsvimportjsonimportmathfrompathlibimportPathRAW_FEATURES=["dynamic_peak_g","accel_rise_g","rotation_peak_dps",]MODEL_FEATURES=[*RAW_FEATURES,"timing_closeness","rotation_per_dynamic","rotation_speed_balance","coordinated_rotation",]FEATURE_WEIGHT_MULTIPLIERS={"dynamic_peak_g":1.45,"accel_rise_g":1.45,"rotation_peak_dps":0.85,"timing_closeness":1.10,"rotation_per_dynamic":0.55,"rotation_speed_balance":1.00,"coordinated_rotation":0.80,}defrow_float(row:dict[str,str],name:str)->float:returnfloat(row[name])defrow_float_optional(row:dict[str,str],name:str,fallback:float)->float:value=row.get(name,"")ifvalueisNoneorstr(value).strip()=="":returnfallbackreturnfloat(value)deftiming_closeness(row:dict[str,str])->float:fallback_offset=row_float(row,"timing_offset_sec")decel_offset=row_float_optional(row,"decel_offset_sec",fallback_offset)offset=abs(row_float_optional(row,"arrival_offset_sec",decel_offset))returnmax(0.0,1.0-offset/0.20)deffeature_value(row:dict[str,str],name:str)->float:ifname=="timing_closeness":returntiming_closeness(row)ifname=="rotation_per_dynamic":returnrow_float(row,"rotation_peak_dps")/max(row_float(row,"dynamic_peak_g"),0.001)ifname=="rotation_speed_balance":return0.0ifname=="coordinated_rotation":return0.0returnrow_float(row,name)defvalid_labeled_row(row:dict[str,str])->bool:returnrow.get("user_score","").strip()in{"1","2","3","4","5"}defread_rows(paths:list[Path],labeled_only:bool)->list[dict[str,str]]:rows:list[dict[str,str]]=[]forpathinpaths:withpath.open(newline="",encoding="utf-8-sig")asfp:reader=csv.DictReader(fp)forrowinreader:iflabeled_onlyandnotvalid_labeled_row(row):continuerow["source_file"]=path.namerows.append(row)returnrowsdefpearson(xs:list[float],ys:list[float])->float:iflen(xs)<2:return0.0mx=sum(xs)/len(xs)my=sum(ys)/len(ys)num=sum((x-mx)*(y-my)forx,yinzip(xs,ys))den_x=math.sqrt(sum((x-mx)**2forxinxs))den_y=math.sqrt(sum((y-my)**2foryinys))ifden_x==0orden_y==0:return0.0returnnum/(den_x*den_y)defnormalize(value:float,low:float,high:float)->float:ifhigh<=low:return0.5returnmax(0.0,min(1.0,(value-low)/(high-low)))defbuild_model(rows:list[dict[str,str]])->dict:user_scores=[row_float(row,"user_score")forrowinrows]feature_stats={}raw_weights={}dynamic_values=[row_float(row,"dynamic_peak_g")forrowinrows]rotation_values=[row_float(row,"rotation_peak_dps")forrowinrows]dynamic_min=min(dynamic_values)dynamic_max=max(dynamic_values)rotation_min=min(rotation_values)rotation_max=max(rotation_values)defderived_value(row:dict[str,str],feature:str)->float:dynamic_norm=normalize(row_float(row,"dynamic_peak_g"),dynamic_min,dynamic_max,)rotation_norm=normalize(row_float(row,"rotation_peak_dps"),rotation_min,rotation_max,)close=timing_closeness(row)iffeature=="rotation_speed_balance":return1.0-abs(rotation_norm-dynamic_norm)iffeature=="coordinated_rotation":returnmath.sqrt(dynamic_norm*rotation_norm)*closereturnfeature_value(row,feature)forfeatureinMODEL_FEATURES:values=[derived_value(row,feature)forrowinrows]corr=pearson(values,user_scores)raw_weights[feature]=max(0.0,corr)*FEATURE_WEIGHT_MULTIPLIERS[feature]feature_stats[feature]={"min":min(values),"max":max(values),"correlation":corr,"weight_multiplier":FEATURE_WEIGHT_MULTIPLIERS[feature],}feature_stats["_normalization"]={"dynamic_peak_g":{"min":dynamic_min,"max":dynamic_max},"rotation_peak_dps":{"min":rotation_min,"max":rotation_max},}total_weight=sum(raw_weights.values())iftotal_weight<=0:weights={feature:1.0/len(MODEL_FEATURES)forfeatureinMODEL_FEATURES}else:weights={feature:raw_weights[feature]/total_weightforfeatureinMODEL_FEATURES}return{"version":3,"score_min":1.0,"score_max":5.0,"features":feature_stats,"weights":weights,}defpredict(row:dict[str,str],model:dict)->float:total=0.0forfeature,weightinmodel["weights"].items():stats=model["features"][feature]value=model_feature_value(row,feature,model)total+=weight*normalize(value,stats["min"],stats["max"])return100.0*totaldefmodel_feature_value(row:dict[str,str],feature:str,model:dict)->float:normalization=model["features"].get("_normalization",{})dynamic_stats=normalization.get("dynamic_peak_g",{})rotation_stats=normalization.get("rotation_peak_dps",{})dynamic_norm=normalize(row_float(row,"dynamic_peak_g"),dynamic_stats.get("min",0.0),dynamic_stats.get("max",1.0),)rotation_norm=normalize(row_float(row,"rotation_peak_dps"),rotation_stats.get("min",0.0),rotation_stats.get("max",1.0),)close=timing_closeness(row)iffeature=="rotation_speed_balance":return1.0-abs(rotation_norm-dynamic_norm)iffeature=="coordinated_rotation":returnmath.sqrt(dynamic_norm*rotation_norm)*closereturnfeature_value(row,feature)defscore_to_label(score:float)->str:ifscore>=85:return"S"ifscore>=70:return"A"ifscore>=50:return"B"ifscore>=30:return"C"return"D"defwrite_scored_rows(rows:list[dict[str,str]],model:dict,output:Path)->None:ifnotrows:returnfieldnames:list[str]=[]forrowinrows:fornameinrow.keys():ifname!="score"andnamenotinfieldnames:fieldnames.append(name)fornamein["learned_score","learned_label"]:ifnamenotinfieldnames:fieldnames.append(name)output.parent.mkdir(parents=True,exist_ok=True)withoutput.open("w",newline="",encoding="utf-8")asfp:writer=csv.DictWriter(fp,fieldnames=fieldnames)writer.writeheader()forrowinrows:learned_score=predict(row,model)row=dict(row)row.pop("score",None)row["learned_score"]=f"{learned_score:.1f}"row["learned_label"]=score_to_label(learned_score)writer.writerow(row)defprint_model(model:dict)->None:print("Learned feature correlations and weights:")forfeatureinMODEL_FEATURES:stats=model["features"][feature]weight=model["weights"][feature]print(f" {feature:>18}: corr={stats['correlation']: .3f} "f"weight={weight: .3f} range={stats['min']:.3f}..{stats['max']:.3f}")defexpand_paths(paths:list[Path])->list[Path]:expanded:list[Path]=[]forpathinpaths:text=str(path)ifany(charintextforcharin"*?[]"):matches=sorted(Path().glob(text))expanded.extend(matches)else:expanded.append(path)returnexpandeddefmain()->int:parser=argparse.ArgumentParser(description="Calibrate a personal punch score from user_score labels.")parser.add_argument("paths",nargs="*",type=Path,help="Labeled summary CSV files. Defaults to data/*_summary.csv",)parser.add_argument("--model",type=Path,default=Path("data/subjective_score_config.json"),help="Output model JSON path",)parser.add_argument("--scored-output",type=Path,default=Path("data/labeled_punches_scored.csv"),help="Output CSV with learned_score added",)args=parser.parse_args()paths=expand_paths(args.paths)ifargs.pathselsesorted(Path("data").glob("*_summary.csv"))ifnotpaths:print("No summary CSV files found.")return1labeled_rows=read_rows(paths,labeled_only=True)iflen(labeled_rows)<10:print("Need at least 10 labeled punches to calibrate a rough score.")return1model=build_model(labeled_rows)args.model.parent.mkdir(parents=True,exist_ok=True)args.model.write_text(json.dumps(model,indent=2),encoding="utf-8")write_scored_rows(labeled_rows,model,args.scored_output)print(f"Labeled punches used: {len(labeled_rows)}")print_model(model)print(f"Saved model to {args.model}")print(f"Saved scored rows to {args.scored_output}")iflen(labeled_rows)<40:print("Tip: this is still a rough calibration. Add 40-100 labeled punches.")return0if__name__=="__main__":raiseSystemExit(main())
Real-Time Personalized Punch Scoring
Python
Applies the calibrated model to incoming IMU data for real-time punch evaluation.
#!/usr/bin/env python3"""Realtime punch evaluation using the current subjective score calibration."""from__future__importannotationsimportargparseimportjsonimportsysimporttimefrompathlibimportPathtry:importserialexceptImportErrorasexc:raiseSystemExit("pyserial is required. Install it with: python -m pip install pyserial")fromexcfromcalibrate_subjective_scoreimportbuild_model,predict,read_rows,score_to_labelfromread_imuimport(Sample,analyze_punches,auto_detect_port,detect_peaks,parse_sample,print_sample_rate_warning,save_csv,save_plot,)G_TO_MS2=9.80665defload_or_build_model(model_path:Path,summary_glob:str)->dict|None:ifmodel_path.exists():returnjson.loads(model_path.read_text(encoding="utf-8-sig"))rows=read_rows(sorted(Path("data").glob(summary_glob)),labeled_only=True)iflen(rows)<10:returnNonemodel=build_model(rows)model_path.parent.mkdir(parents=True,exist_ok=True)model_path.write_text(json.dumps(model,indent=2),encoding="utf-8")returnmodeldefanalysis_row(analysis)->dict[str,str]:return{"score":f"{analysis.score}","dynamic_peak_g":f"{analysis.peak.value}","accel_rise_g":f"{analysis.accel_rise_g}","rotation_peak_dps":f"{analysis.rotation_peak_dps}","timing_offset_sec":f"{analysis.timing_offset_sec}","decel_offset_sec":f"{analysis.decel_offset_sec}","arrival_offset_sec":f"{analysis.arrival_offset_sec}",}deflearned_score_for(analysis,model:dict|None)->tuple[float,str]:ifmodelisNone:returnanalysis.score,score_to_label(analysis.score)learned_score=predict(analysis_row(analysis),model)returnlearned_score,score_to_label(learned_score)defsend_display_score(ser:serial.Serial,score:float,label:str,accel_ms2:float)->None:command=f"DISPLAY_SCORE,{score:.1f},{label},{accel_ms2:.1f}\n"ser.write(command.encode("ascii"))deflatest_finalized_analyses(samples:list[Sample],args:argparse.Namespace,finalized_before_sec:float,)->list:finalized=[sampleforsampleinsamplesifsample.time_sec<=finalized_before_sec]iflen(finalized)<3:return[]peaks=detect_peaks(finalized,args.peak_field,args.peak_threshold,args.peak_min_interval_sec,args.max_peaks,args.peak_neighbor_gap_sec,)returnanalyze_punches(finalized,peaks,args.rotation_field)defrecent_samples(samples:list[Sample],latest_time_sec:float,window_sec:float)->list[Sample]:start_sec=max(0.0,latest_time_sec-window_sec)return[sampleforsampleinsamplesifsample.time_sec>=start_sec]defmain()->int:parser=argparse.ArgumentParser(description="Read IMU serial CSV and print realtime personal punch scores.")parser.add_argument("-p","--port",help="Serial port, for example COM10")parser.add_argument("-b","--baud",type=int,default=921600)parser.add_argument("-o","--output-dir",default="data")parser.add_argument("--model",type=Path,default=Path("data/subjective_score_config.json"))parser.add_argument("--summary-glob",default="*_summary.csv")parser.add_argument("--peak-field",choices=["dynamic_g","mag_g"],default="dynamic_g")parser.add_argument("--rotation-field",choices=["gx_dps","gy_dps","gz_dps","rotation_dps"],default="rotation_dps")parser.add_argument("--peak-threshold",type=float,default=1.5)parser.add_argument("--peak-min-interval-sec",type=float,default=0.6)parser.add_argument("--peak-neighbor-gap-sec",type=float,default=0.25)parser.add_argument("--max-peaks",type=int,default=100)parser.add_argument("--settle-sec",type=float,default=0.35)parser.add_argument("--analysis-window-sec",type=float,default=8.0)parser.add_argument("--expected-hz",type=float,default=100.0)parser.add_argument("--no-data-warning-sec",type=float,default=5.0)parser.add_argument("--save",action="store_true",help="Save CSV/PNG when stopped")args=parser.parse_args()model=load_or_build_model(args.model,args.summary_glob)ifmodelisNone:print("No subjective model yet. Realtime output will use the old sensor score.\n""Add labels to summary CSV files, then run calibrate_subjective_score.py.")else:print(f"Using subjective model: {args.model}")port=args.portorauto_detect_port()ifnotport:print("No serial port found. Specify one with --port COMx.",file=sys.stderr)return2output_dir=Path(args.output_dir)samples:list[Sample]=[]first_device_ms:int|None=Noneprinted_peak_times:set[float]=set()last_status_time=0.0last_sample_wall_time=time.monotonic()print(f"Opening {port} at {args.baud} bps.")print("Realtime scoring. Press Ctrl+C to stop.")print(f"Analysis window: latest {args.analysis_window_sec:g}s")print("No. time[s] learned label old dynamic[g] rise[g] ""rotation[dps] peak_offset[s] decel_offset[s] arrival_offset[s]")try:withserial.Serial(port,args.baud,timeout=1)asser:whileTrue:raw_line=ser.readline()ifnotraw_line:now=time.monotonic()ifnow-last_sample_wall_time>=args.no_data_warning_sec:print("\nNo valid samples recently. Check the M5Stack connection, ""Serial Monitor, and firmware.",flush=True,)last_sample_wall_time=nowcontinueline=raw_line.decode("utf-8",errors="replace")sample,first_device_ms=parse_sample(line,first_device_ms)ifsampleisNone:text=line.strip()iftext.startswith("#"):print(text)continuesamples.append(sample)last_sample_wall_time=time.monotonic()latest_time=sample.time_secfinalized_before=latest_time-args.settle_secanalysis_samples=recent_samples(samples,latest_time,args.analysis_window_sec)analyses=latest_finalized_analyses(analysis_samples,args,finalized_before)foranalysisinanalyses:peak_time=round(analysis.peak.sample.time_sec,3)ifpeak_timeinprinted_peak_times:continueprinted_peak_times.add(peak_time)learned_score,learned_label=learned_score_for(analysis,model)peak_accel_ms2=analysis.peak.value*G_TO_MS2send_display_score(ser,learned_score,learned_label,peak_accel_ms2)print(f"{len(printed_peak_times):>2} "f"{analysis.peak.sample.time_sec:>7.3f} "f"{learned_score:>7.1f} "f"{learned_label:>5} "f"{analysis.score:>5.1f} "f"{analysis.peak.value:>10.3f} "f"{analysis.accel_rise_g:>7.3f} "f"{analysis.rotation_peak_dps:>13.1f} "f"{analysis.timing_offset_sec:>14.3f} "f"{analysis.decel_offset_sec:>15.3f} "f"{analysis.arrival_offset_sec:>17.3f}")now=time.monotonic()ifnow-last_status_time>=1.0:print(f"\rSamples: {len(samples)} t={latest_time:7.3f}s",end="",flush=True,)last_status_time=nowexceptKeyboardInterrupt:print()ifnotsamples:print("No samples were recorded.")return1print_sample_rate_warning(samples,args.expected_hz)ifargs.save:stamp=time.strftime("%Y%m%d_%H%M%S")csv_path=output_dir/f"realtime_{stamp}.csv"image_path=output_dir/f"realtime_{stamp}.png"save_csv(samples,csv_path)save_plot(samples,image_path,args.peak_field,args.peak_threshold,args.peak_min_interval_sec,args.max_peaks,args.rotation_field,args.peak_neighbor_gap_sec,)print(f"Saved {csv_path}")print(f"Saved {image_path}")return0if__name__=="__main__":raiseSystemExit(main())
PlatformIO Configuration
INI
PlatformIO build configuration and required M5Stack libraries.
; PlatformIO Project Configuration File;; Build options: build flags, source filter; Upload options: custom upload port, speed and extra flags; Library options: dependencies, extra library storages; Advanced options: extra scripting;; Please visit documentation for the other options and examples; https://docs.platformio.org/page/projectconf.html[platformio]data_dir=spiffs_data[env:m5stack-core2]platform=espressif32board=m5stack-core2framework=arduinomonitor_speed=921600lib_deps=m5stack/M5Unifiedhttps://github.com/m5stack/M5Unit-IMU-Pro-Mini.githttps://github.com/adafruit/Adafruit_BMP280_Library.git
Comments