From 68422dd3049834425080c4752fe7a6ca1e8fc0fd Mon Sep 17 00:00:00 2001 From: noisedestroyers Date: Fri, 10 Apr 2026 17:28:29 -0400 Subject: [PATCH] bookmark - Refactor --- src/impakt/io/csv.py | 29 ++ src/impakt/io/tdms.py | 37 ++ .../plot/__pycache__/engine.cpython-312.pyc | Bin 7672 -> 9208 bytes .../plot/__pycache__/spec.cpython-312.pyc | Bin 7555 -> 7720 bytes src/impakt/plot/engine.py | 168 +++++---- src/impakt/plot/spec.py | 5 + src/impakt/plugin/registry.py | 9 +- .../__pycache__/euro_ncap.cpython-312.pyc | Bin 6719 -> 8192 bytes .../protocol/__pycache__/iihs.cpython-312.pyc | Bin 5073 -> 6250 bytes src/impakt/protocol/euro_ncap.py | 63 +++- src/impakt/protocol/iihs.py | 55 ++- .../protocol/thresholds/euro_ncap_2024.yaml | 76 ++++ src/impakt/protocol/thresholds/iihs_2024.yaml | 40 +++ .../script/__pycache__/api.cpython-312.pyc | Bin 15746 -> 20163 bytes .../script/__pycache__/cli.cpython-312.pyc | Bin 8320 -> 0 bytes src/impakt/script/api.py | 150 ++++++-- src/impakt/web/app.py | 38 +- .../web/callbacks/criteria_callbacks.py | 23 +- src/impakt/web/callbacks/plot_callbacks.py | 237 ++++++------- src/impakt/web/components/channel_tree.py | 221 ------------ src/impakt/web/components/header.py | 2 +- src/impakt/web/state.py | 333 ++++++------------ tests/test_criteria/test_chest_femur_tibia.py | 93 +++++ tests/test_plot/test_engine.py | 81 +++++ tests/test_protocol/test_iihs_usncap.py | 68 ++++ tests/test_scripting_api.py | 126 +++++++ tests/test_template.py | 132 +++++++ tests/test_transform/test_math_resultant.py | 75 ++++ tests/test_web/test_state.py | 66 ++-- 29 files changed, 1371 insertions(+), 756 deletions(-) create mode 100644 src/impakt/io/csv.py create mode 100644 src/impakt/io/tdms.py create mode 100644 src/impakt/protocol/thresholds/euro_ncap_2024.yaml create mode 100644 src/impakt/protocol/thresholds/iihs_2024.yaml delete mode 100644 src/impakt/script/__pycache__/cli.cpython-312.pyc delete mode 100644 src/impakt/web/components/channel_tree.py create mode 100644 tests/test_criteria/test_chest_femur_tibia.py create mode 100644 tests/test_plot/test_engine.py create mode 100644 tests/test_protocol/test_iihs_usncap.py create mode 100644 tests/test_scripting_api.py create mode 100644 tests/test_template.py create mode 100644 tests/test_transform/test_math_resultant.py diff --git a/src/impakt/io/csv.py b/src/impakt/io/csv.py new file mode 100644 index 0000000..4dd5e8f --- /dev/null +++ b/src/impakt/io/csv.py @@ -0,0 +1,29 @@ +"""CSV data reader (stub). + +Placeholder for a future CSV reader plugin. +Install or implement to enable reading CSV crash test data. +""" + +from __future__ import annotations + +from pathlib import Path + +from impakt.channel.model import TestData, TestMetadata + + +class CSVReader: + """Reader for CSV time-series data (not yet implemented).""" + + @property + def format_name(self) -> str: + return "CSV" + + def supports(self, path: Path) -> bool: + # Only claim support for .csv files with crash test structure + return False # Disabled until implemented + + def metadata(self, path: Path) -> TestMetadata: + raise NotImplementedError("CSV reader is not yet implemented. See BRAINSTORM.md.") + + def read(self, path: Path) -> TestData: + raise NotImplementedError("CSV reader is not yet implemented. See BRAINSTORM.md.") diff --git a/src/impakt/io/tdms.py b/src/impakt/io/tdms.py new file mode 100644 index 0000000..e0ce04d --- /dev/null +++ b/src/impakt/io/tdms.py @@ -0,0 +1,37 @@ +"""NI TDMS data reader (stub). + +Placeholder for a future TDMS reader plugin. +Install nptdms to enable: pip install impakt[tdms] +""" + +from __future__ import annotations + +from pathlib import Path + +from impakt.channel.model import TestData, TestMetadata + + +class TDMSReader: + """Reader for NI TDMS files (not yet implemented). + + Requires the optional ``nptdms`` dependency:: + + pip install impakt[tdms] + """ + + @property + def format_name(self) -> str: + return "NI TDMS" + + def supports(self, path: Path) -> bool: + return False # Disabled until implemented + + def metadata(self, path: Path) -> TestMetadata: + raise NotImplementedError( + "TDMS reader is not yet implemented. Install nptdms: pip install impakt[tdms]" + ) + + def read(self, path: Path) -> TestData: + raise NotImplementedError( + "TDMS reader is not yet implemented. Install nptdms: pip install impakt[tdms]" + ) diff --git a/src/impakt/plot/__pycache__/engine.cpython-312.pyc b/src/impakt/plot/__pycache__/engine.cpython-312.pyc index 4644c3e465dbdffeba3f369406af7bd720d7127f..ca36e3edc080ee4575075e69bfabe907a58371ad 100644 GIT binary patch delta 4504 zcmai2YitzP6`t9h*_U_M_P+gCd+-CiYh$o61RG3v7^e^*BoLC6fEn+Mch`(}+&klE z*D{In!y%;LOj|-i1hkdv$X03zRV7lJqP9^Z^+)X#mCYzpXjQ3l|8x-`rT==)T|0JY zqK{gcPzV&HNoKTChQZrb|+KNMscX8NNGldj__Svabw z`uLg`#9w$TcnKQ8t z;e~=r2%(@iqY4E*O+rScs)WaEH3@n;nbL&;kw9d%CP5nz6$QYEqYz|8?3ER*S=cv3 zG!*~>^0OF6^jZ+o+cY5s;Z$Tv=p7T}QE)9ID#F2I1=tA-rgQl!h0)3|?1Mkpjh|Zf z+O&;XOTXQ?!39LgB3a>&gFjxfNp{I`)gn2s+LNs0lKAurEt1@lM=FxMSGl?3S-zl` zWwMrIOqT6;tQcAyyO>h}hMBNt4K*AxIIGX;S(!BcYzY~^u{#1?%q8|zkYQd1tP|V> zpXK|##!e853m(^JJ<@t=myuzDwo!3vbV*U#1(VOv$*YAU6TQuCZ+ zcKOAb+q810*BTpbRWJX4L5YQTh|dL`wmu zA=*NgSKK(bfZsY%LXrQC4;n{oA*06D#O^lWiQP+S7gRv7zGMx`EmE7bRoW(Tb5E?i zv4Cy+tjpl-)qDeJJAhRaY=@k6%$b z)7EYBgW{k}k_7^?z*3!L-ywm{dmY{<%OI1K(oXILCOH5Wb+1?m$I&^tp zP=T$9;M;BTqu8=devA&bLPIxs1@=qreJk-%nUS=lr%X-~wN%VTu!)f{PH?zqm|R92 z0BWcXj}dolddc+Ajp-{pF%TX1lVq>h(Av7WY3r7zZS75)o43W5u&Iq(Uu@|dw>Gzd zYub|Yn`O7hVs;WTe&yIc8GvxAx~QksOb>eALjx9koJ#{Z9-vlRwluZ0Y;I}=%V-uX zwswBg)b^NZ8Et`YtH~vaIA*d4Y+Ijc)z!ge$0rFkA+iNY8m@v@K(5={+hZ2uhX;b)JZ3YCXdZg>w64fz5zR*7fz*J2>7*fraEE2n zg}E!nLO1?_JJVjLedSyw54R+#R*mn zHAD7O_8Q&B$IgAR_n8<(CaIdX?sRg9$fjeZs7=q%paiwlqljbbkZy!r>y2(#1M4&3 z+;Ec9(^wcFAOcM%^g_gF05LeGJyAHGnA}h%t(#6!l6vs8Ae;7M2~md{HJzh933`ms z3e&?_nWWyxyMk5Z2m+^=i!dfV%fRvFCJjbe(LT1+I91fR{|qv)lt}`J>83V%&cX|n z)W~+^wf^as<9y$u(0RYu5e((6uBa~`Vj}CV z*Dr(`a$%t`Mys#)&X8M{#qf@NHRGvz$dq`z`5Gn=o;sK}s@?BRg6dG@@(a^1Ftu6#k<0-j#`1QkAYp%twfY!F_ z*~Rk7w!gZ!=AUHBTk~Bk<1V?l>-?@C>-wPwl;A$w4%-=d<@p{Fgn`o?IxB ztE$UYx8v@>)VvWN6{9fMVEE)gCGm;|YZ|g67{q zasdgV)Oa9>Z5iB~#_mAGdk1G~fdDwK&qxKT0mW3{ou{c@JLpa614@irqxK?57c_p| z+e^S7_pScgZC+mdnK2M-biN9+M?lYraU)nYNyGLQDx$H&<1m>9Vsb~+47g=l2Q$g# z6AH6C<{;D|+?%OarWJnsQrAqrPf1(`7Bj9*06%U4VE^t`~l!QYrEy*sFGh zT8^WKSjO9oC(7D5NJhf=Zh5^Y1q)-=$!R1#KuntmrJj*s z%_rsSO0cp`2kZtzau4{IGX7H@-LV~oai<`tz_h|~jAZfg29gBwEW>bq7;e>YpQqWT zgq$;;t?-x50rLy^YAzNto_F6Is<5+9p~Dd485qVRm_YjPgsh;XHT~qi!twD`0sfCc zrhtInVa^DiFk(S+)%bnobL+!FLf`14Ec#4z@179hBOe_X}!OY!Q{PM=VztSn1nd`&P znZ87h0gd$%Cbf&A=x4h)e40J#{~q;zD@Dz_YE;RCgIr(0*U+rdB_1E~2e@8rmE6ysyTT+OZgm_}1r96h`iCY|UNj#$Aa> zgs%CFzQ%+a7~6vHx}*5-?udVtSs`(PQuBP|X?53pA5)SGr#;C9_7irMTk$xf4WDT0 z7JY+bDH65;XXIr>2ftZv2bQ02s_~LL&(xYEPoBv$Wknpekb0`$1u%Dc1R&9 zyb%VeMT$tRQdEkq##cDU%xYp~ofb5{%*WkmD=|3=A&e#^UP`UDHDX1>V#i>__hJM; z<~zAQdj>g+!0V1p5?8u3MhKP)t)@4ENQB4nn@w^07#!Rt<>R2-vPqelpfc?^=KpS* zQw^QJa>LAx~9Q(U2`> zW;glvAfTwl7LYjm51@=iADnXyyqdRY-EtXPCF4dB{3NiyJ0h0LnjspBR?b#5U4e$> zuc(@#&StL^6+^~5n)?qHoUJ}Nl;Nx<4JmTjnPYio#d5BwAw=#TWqD@SdlAyh_$XFY_}9&&!ORwmDQK2^ zOes{6jQc}deBNA0Sp$hV+44-}M8lAg#Tv?#46!prf=bRn zgpL&I0!YB(Y_+TymTxATt0G-Px@B9LhBL4N)e4y?XH{`ls~Ttrej19lT1~Kztm7U6 zI-@Y|4Uf_RTnumTJqm&lz%c;0%&3ik#m!`=MO6=wHa5fW#lR1Jf@Gts)Zc=ko62?N z!%Np(_*X5TEO4hyr*5Az`?f8QKiGca{?R{;|9<@bfd}Vu54xqbxIf(TZ&xr7thZ6A zJ@sMAAG$g6!H5}6E?ro>aDD9B(S=)#+1s0XNU$8*XPZmGMdt^@TL$>1R}LD zy4K`w4b~%+FSSO6e0;t4-p`|dVRl2hp7RP@tCfn3GN_q!$AIRfw#kX2(1N^-;sCqw zFB7T4w+<UQt!c2Wa0zw2MqSoe3cOgvY?b0f2FU*9cggy|S~yagcTcIDEsqlF6R8 zq5lrRy8vKFX9)3;#SzE|rV~LlZSGuDab+@(Q^$ z2$a(M$fFm5nMY|Bs8#qLZ_RtZx|2FqYf#!{Th?a)v=72^YyETGrlo zhC_C+&qGiz0jK~FdmUZ^*#3IRr$~1$0#P_tE|6P4(6I3lD8rk2l_Csoj^HzGoh06u z+kU*)Ud%%ji^i0iVP9HHBl^xVn-8{vN78Y68NZd@?wce!r^&4I_=9vu)aC<%X*Ma~ zEk>8|Z_??Lw)5~>**maYx`FI%SX^G!L<3zWLl$QTCqthjRNt5YZdmN3rm40Ts^Y%( zy>EcqXcz!CjVc5_rx3W~u+I-I0D*iuWUU`iU$Hx%v2V~^T?KezvR+Qp5a z$$P8q2E&P+G|6b)!oCHEYnyon9Ls7b+ep_A-zJDu(2$@ix*=$JVRSv2Z=CVW?m=Np zL1fs;%R=LA61GA{e~DrWI)6a}SB-i(R{ z52+-OYz+g1X@V@~io*I993Kh6AirBpwpLa&vNKo_Vu2Gxm~pMAeSAacY#$*KbO_)u zz!3tLTO>KEETLogzR(qdlw^4uhR_N8AEABd#lwNXfaOAR9$f$f-v_YQW~RwgKfOnX zzs)wGZ9o_CnXX9qheY@@^65M|9>U;`d(|!vJr0MR@uKUbno9ub?W3sKC-Lt3F}&}% zKTUM(rwiFFU>own^eC#~-*%s+58znO@gt)kx0wS=qIrN9l(M|TF=1afx3&-PrQLTM`}oa!MX=pUdr9F K<|8e8ng0Uku?$WC diff --git a/src/impakt/plot/__pycache__/spec.cpython-312.pyc b/src/impakt/plot/__pycache__/spec.cpython-312.pyc index f694a6ffb7780ac2adf97433b21255a05e816313..49848d9aac80bc28b6337ace6ead9c71fd7ca4d7 100644 GIT binary patch delta 261 zcmZp+USY#`nwOW00SKO@-OOAnypgY0f{B4~@?42)dMVOT!YRV3>?txfK=~BeD3KI7 zAWt65V~-L|Q2_E3LA*+KO{LA{k}-_FZnxNz^K%OllS`^ND-=reGSf0sQ>wU%ONta~ z6q54ub8c~Fxu<>sfPR*9x$CYMA5<#ZG>^GafYD*W>EQg5;4CKjb<=4pyep2#IS z`IVG6cQw$Cq8boUJK0cLk#YUxVCg=U%Pb-v*ce2m8r(ZVF0)A9kWp{&?5MiTqJBd~ tqrvS0kKz>;jmb}>uQ0Yw-YK(LT#Qlk0|Suw%*?>V{gDaCDv|*j2>`EDPjCPL delta 99 zcmZ2s(`?OmnwOW00SMgnZf0&2+Q`={!E}vb@?42)ES2n<(wi4b#xOEz3QwNMB|14? z+I#X&X<5d;$=9X(n6I#iOzxDq!q_x9PIj}nAfxC91|ad7nSqJ>BNLESBm*=R0NO_! AL;wH) diff --git a/src/impakt/plot/engine.py b/src/impakt/plot/engine.py index c8fdbbd..aa6f555 100644 --- a/src/impakt/plot/engine.py +++ b/src/impakt/plot/engine.py @@ -2,6 +2,9 @@ Renders PlotSpec objects into interactive Plotly figures with support for corridors, dual X-cursors, and export. + +The PlotEngine is the **single rendering path** — both the scripting +API and the web UI construct PlotSpec objects and delegate here. """ from __future__ import annotations @@ -30,12 +33,39 @@ DEFAULT_COLORS = [ class PlotEngine: - """Renders PlotSpec into Plotly figures.""" + """Renders PlotSpec into Plotly figures. + + Supports two modes: + - **Default**: Full interactive plot with hover tooltips and legend. + - **Compact** (``spec.compact=True``): Web UI mode with no legend, no + hover tooltips, tight margins, smaller axis labels. Cursor tracking + is handled by external JS. + """ def render(self, spec: PlotSpec) -> go.Figure: """Render a PlotSpec into an interactive Plotly figure.""" fig = go.Figure() + if not spec.channels: + fig.update_layout( + template="plotly_white", + annotations=[ + { + "text": "Select channels to plot", + "xref": "paper", + "yref": "paper", + "x": 0.5, + "y": 0.5, + "showarrow": False, + "font": {"size": 14, "color": "#bbb"}, + } + ], + xaxis={"visible": False}, + yaxis={"visible": False}, + margin={"l": 40, "r": 20, "t": 30, "b": 40}, + ) + return fig + # Add corridor fills first (behind data traces) for corridor in spec.corridors: self._add_corridor(fig, corridor) @@ -54,53 +84,92 @@ class PlotEngine: color = style.color or DEFAULT_COLORS[i % len(DEFAULT_COLORS)] label = style.label or ch_ref.label - fig.add_trace( - go.Scatter( - x=ch.time, - y=ch.data, - mode="lines", - name=label, - line=dict( - color=color, - width=style.line_width, - dash=style.line_dash, - ), - opacity=style.opacity, - hovertemplate=f"{label}
t=%{{x:.6f}}s
%{{y:.4f}} {ch.unit}", + trace_kwargs: dict[str, Any] = { + "x": ch.time, + "y": ch.data, + "mode": "lines", + "name": label, + "line": dict( + color=color, + width=style.line_width, + dash=style.line_dash, + ), + "opacity": style.opacity, + } + + if not spec.compact: + trace_kwargs["hovertemplate"] = ( + f"{label}
t=%{{x:.6f}}s
%{{y:.4f}} {ch.unit}" ) - ) + + fig.add_trace(go.Scatter(**trace_kwargs)) # Add cursor lines if spec.x_cursors: x1, x2 = spec.x_cursors - for x_val, label in [(x1, "x1"), (x2, "x2")]: + if spec.compact: + # Color-coded cursor lines for compact mode fig.add_vline( - x=x_val, + x=x1, line_dash="dash", - line_color="gray", + line_color="rgba(220,53,69,0.6)", line_width=1, - annotation_text=f"{label}={x_val:.6f}s", - annotation_position="top", + annotation_text=f"X1={x1:.4f}s", + annotation_font_size=9, + annotation_font_color="rgba(220,53,69,0.8)", ) + fig.add_vline( + x=x2, + line_dash="dash", + line_color="rgba(13,110,253,0.6)", + line_width=1, + annotation_text=f"X2={x2:.4f}s", + annotation_font_size=9, + annotation_font_color="rgba(13,110,253,0.8)", + ) + else: + for x_val, lbl in [(x1, "x1"), (x2, "x2")]: + fig.add_vline( + x=x_val, + line_dash="dash", + line_color="gray", + line_width=1, + annotation_text=f"{lbl}={x_val:.6f}s", + annotation_position="top", + ) # Layout - fig.update_layout( - title=spec.title, - xaxis_title=spec.x_label, - yaxis_title=spec.y_label, - showlegend=spec.show_legend, - height=spec.height, - width=spec.width, - template="plotly_white", - hovermode="x unified", - legend=dict( - orientation="h", - yanchor="bottom", - y=-0.3, - xanchor="center", - x=0.5, - ), - ) + if spec.compact: + margin = spec.margin or {"l": 45, "r": 8, "t": 4, "b": 28} + fig.update_layout( + xaxis_title=dict(text=spec.x_label, font=dict(size=10, color="#999")), + yaxis_title=dict(text=spec.y_label, font=dict(size=10, color="#999")), + template="plotly_white", + hovermode=False, + showlegend=False, + margin=margin, + ) + else: + margin = spec.margin or {"l": 60, "r": 20, "t": 40 if spec.title else 10, "b": 60} + hovermode = spec.hovermode if spec.hovermode is not None else "x unified" + fig.update_layout( + title=spec.title, + xaxis_title=spec.x_label, + yaxis_title=spec.y_label, + showlegend=spec.show_legend, + height=spec.height, + width=spec.width, + template="plotly_white", + hovermode=hovermode, + legend=dict( + orientation="h", + yanchor="bottom", + y=-0.3, + xanchor="center", + x=0.5, + ), + margin=margin, + ) if spec.show_grid: fig.update_xaxes(showgrid=True, gridwidth=1, gridcolor="rgba(128,128,128,0.2)") @@ -117,7 +186,6 @@ class PlotEngine: """Add a corridor (tolerance band) to the figure.""" style = corridor.style - # Upper bound fig.add_trace( go.Scatter( x=corridor.time, @@ -128,8 +196,6 @@ class PlotEngine: showlegend=False, ) ) - - # Lower bound with fill to upper fig.add_trace( go.Scatter( x=corridor.time, @@ -144,16 +210,7 @@ class PlotEngine: ) def to_image(self, spec: PlotSpec, format: str = "png", scale: float = 2.0) -> bytes: - """Render to a static image. - - Args: - spec: Plot specification. - format: Image format ('png', 'svg', 'pdf', 'jpeg'). - scale: Resolution multiplier. - - Returns: - Image bytes. - """ + """Render to a static image.""" fig = self.render(spec) return fig.to_image(format=format, scale=scale) @@ -168,16 +225,7 @@ def cursor_values( x1: float, x2: float, ) -> CursorValues: - """Compute interpolated values at two X-axis positions. - - Args: - spec_or_channels: PlotSpec or list of Channels. - x1: First cursor position (time). - x2: Second cursor position (time). - - Returns: - CursorValues with interpolated values for each channel. - """ + """Compute interpolated values at two X-axis positions.""" channels: list[tuple[str, Channel]] = [] if isinstance(spec_or_channels, PlotSpec): diff --git a/src/impakt/plot/spec.py b/src/impakt/plot/spec.py index f8772c1..abb7039 100644 --- a/src/impakt/plot/spec.py +++ b/src/impakt/plot/spec.py @@ -168,3 +168,8 @@ class PlotSpec: show_grid: bool = True height: int = 500 width: int = 900 + # Web UI compact mode: disables hover tooltips, removes legend, + # uses tight margins, and renders axis labels in a smaller font. + compact: bool = False + hovermode: str | bool = "x unified" + margin: dict[str, int] | None = None diff --git a/src/impakt/plugin/registry.py b/src/impakt/plugin/registry.py index 72f2115..d73da9b 100644 --- a/src/impakt/plugin/registry.py +++ b/src/impakt/plugin/registry.py @@ -42,8 +42,15 @@ class PluginRegistry: self._plugins: list[ImpaktPlugin] = [] def register_reader(self, reader: Any) -> None: - """Register a data reader.""" + """Register a data reader. + + Forwards to the IO reader registry so plugin readers are + discoverable by Session.open() and the web UI. + """ self._readers.append(reader) + from impakt.io.reader import register_reader + + register_reader(reader) logger.info("Plugin reader registered: %s", getattr(reader, "format_name", reader)) def register_transform(self, name: str, transform_cls: type) -> None: diff --git a/src/impakt/protocol/__pycache__/euro_ncap.cpython-312.pyc b/src/impakt/protocol/__pycache__/euro_ncap.cpython-312.pyc index 48736701f65ec38f38e252a22fc7a634e579a41b..18deb8cd9ac64995bd58ced58ee3e3de907d1e32 100644 GIT binary patch delta 3429 zcmZ`*Yit}>6~41OyF0twpvk7O-R5P}HnAP&9mjdtl+-C1CZ1V)?R{}) zoW$F;c8gNUsS?#@fCOmO@&^SKP#}LIAq3(l56P)h#Ti_Ys(?U6uq8#6B7Shr-G}W^ zuH|#joY%egyXTy{zuZ~u_5Rl5aS<5C_=CjzE%V;6CCJTh=?-y3;S_t)nR1P|QtlCV zs&1r?oBvv5z9-sLb3Arm5-))2*+a}zQWaFX($(xSM* zGpN)l9>uHp6hHh0l%P`orZ7^kxYP!v;RT4W%02}4AqzP~lvv|sqBOm0DX%cn=p@Rz zsv+>LO8ByEq-i=5Yy0VAm9z0DWqthi%a)9V6O7;S%eDwO*5CfI?T;MiHm*6gb7A9G zj<9V5h`Wss97BO9aHdNir>ZudNh;c7gLC!m9<7{@E$ablC4_tQACO)rIujyMxIFnD zQ7ms+@x(Knan0LW&@2fuL=>ySCoMCUyk){sv#&9EZo*mns4cl+nKb61=6Ks9LIt~d~|>f&oO z!vEDcpVSn}0U@00NmWhbUZbgB*}_Dq%CF3rI@J?@-sYb0DX$N?W_{8r?#T0b8)n

>rZ$A31H>)k_IY*GwKGnNCfPt5Om+W;&@VE0V5W(oI)FOQbbjPLHZ4uOvow z)0WWHlxEuEV5*z^`AjBhTE|p9<}e+VotjRJC}j&!8nhEu2Ul5_O*^FgR1%!hatgv- zkdvB*xhj?i4Wns@?LlG@RuH0lL3a%PwSNF9l1~K^5Z^g^^Wx3Kt*M0#d+z%86pt*q z8}7TC@4B1sxz{cE0<#6@N4)Q$+xN!aD|>Im?nm42M%!;)T!^+WMtc_By~X{@?$Dg_ z=Kg!`^=R&Vc;Zy?;8J7s`o#Rix06>~v;3@5F?rv;{!e}|zR@+GTJ&#PW_Q3uxAAi_G z`@oOxL(&3dnD!fU&TPSqtj?hA&urf*9qbzz*w@#8Y)$vFf)D(2kDdbtx>v2Ug_->f!Mj#rlCyM2)^GT4x(yDwr0?H}0+6si_+t%Rb zcP=glI*UgiI08$)$bt}Aa(L&Sde0GF_H9`Zwk)|l#Um8kyYVyE4+@ys42BE9Tqk*G z-4bx+p(99a*9L$T$x^s&mcQy*Zr*ycYq5F9 zEPqc3FZt^4``X^~wJir*mV$fkjNN&uM7Tp-WI3{BAskzXeX(Ta0?$9PTI;OEIdnegY-7-(P3I_u_DF%H_{N$1R63#4 zHe~JDOd_poR5ZE*PjTJGxj>8)joHBZ-a$~)ok+US(3Z?xgrvsTiUpw?P->U6Sv9Rt zynoYuKBG)a*o;|0P$)w} zV@NRR>C;HIA;EXVbfn}%H<}g z(EO)lUCjk*a92g&y8Th``y3ZFUTgSycnIRavu4eX>Z@-`HO_=Ka7)H3p?!WVxWONC z^e~$8wA=W7D6&5XA9T7biNo|bijUV>O{d$${^}i_X3x=AC^JR?!TlmS?n06?wl#L2 zI|ZyMWN9X=QhmCNFl!-`sHlUdQBRDfRDC?7P!-*9$E*pklTv)M7K$+idS*0} z>^LuLDk}uaHpWt)l?$sBL6Yz%lq5mP$y812tR1bq0q{delGEu7-drZFNfK*jE10jk z!Ok$QFT9Y-C{sywFU3}Z{Xz=>g&xFl|FSqZ{@+2uZTvF{e_+HT=i5IKNqy7YC8Ohh zaP!^Z=9}B@26q>IC3}<$E_wVVE6fjV2OimAF6}jb6KQ@{;w%d!ve10?5pi;sQrO9P bnI+8366VjsD#wLBAxKNZMpJYXk9YoGwc{sK delta 1907 zcmZ8hZ){Ul6u&NzL-cWCCR?bd=dWi8lJHav6DGjrKAbJ={sPfSx{ z%1#2r%>NC{&i5k72U@gY8!}|I?PJ zERGiC^F>x_@| z>bhI@_B-y-f_vlj?DdnM6~6Ad8TdZ(P2_f8>UPh-{GLK=8$2_Dh`R6H4~Mxyd0W!h~`v@LS=-qt*b9=0ke_&6%@^ol~m1`m(NR-!UfU8P`DGCDN`y*MM6y= z*au}Ymab-{anQ7+OA;NDifSe+rBykts2QqI3&hw71Wp+9P$-9HsckbupF`+LM0)@b z0gJPKfOIiakTly$2U(xJeRC6l%!SM=>5{6zv6XUO)eOAdnt==YQE39g^#a+$9(lw3 zV*7njJTkY__W|a&`i}n#-*#?|Va-q)QXjAQ0rqKN4W4B`1fJangW3Nvrt1JI>dnv$ z{1!o0wM>he*e@5Qm_SD$6TAc*b&%~3M*X@5j?Og_|7phIyqr<#2)h{Eer6kJEF~)Q zf~jNG>OiaMNia0BKlHp_uEsDVhqRF_a;$q&= z3A+vk+>pScjVX!i&VARQ%Z1bwN3lOV6=BiG@DA@ES zh=a16r*4pSWpkZtsdJt>*IMWF+Q5SmGgZ3{{5<3opz{(f-#9&9bW?lX70#O~*M7?ymA1Clv$6_`YDq;zKvWZTHx} z;X$DiW|tzt&LkEl8|RyjEFmivDn4YiPamongaNL$%{RQTgaj=7Z6g?a{z5=2N7+K8 IU4%CL2ZN!{wg3PC diff --git a/src/impakt/protocol/__pycache__/iihs.cpython-312.pyc b/src/impakt/protocol/__pycache__/iihs.cpython-312.pyc index fe686f9cab13aa0fdeb7d03f28bd460bf352eaa3..d62f7c9c55a2057fbe5897d511976330da0039c2 100644 GIT binary patch delta 2902 zcmZ`4TTC3+_0G=BzTeC8Sg=_e8$7956YLUdz8fvA(5hDkc5s*QTUD{iu4+?1PfJv)XeX z=l#4}J%8`>|H0>TBl`Vj;iVF7+VHnB*EV|2b}>j|B)jU$xRdUTC+W#}lU@edkj%-x zPZ(f8X2=^YNRFHbp=3acNv?ZbGAOwvkL0C4pX8SUAK8;3$tj1W;4o#NG$Be8aiLfL zq;@Gx@lCRK@ij}s0U$B-{^WrxgyunPgd!5(Lcl!;>U+g~RN<0sHuM&fn(ra05$ZVk59I*v?t3X?!Tg39_Yt?TfMtC_QziXhW}H1al&;34 zDhUG0BE;2hf#JpW{w9!Ld^8t+W|6*JRFcE|bHDm3I(YomU6-uaXdR@^Rm*?_46w!nC zIF#pRkl@XAPH8DAJ~=rt6W>oqd=6`w`1eQOo=Om2Q5JMD$0t@&85KjCr*r(WG1prd z6d2-^vuRCIvh#$S*EET^Q|Yw4WTa+QnK&~kJg;O^DhVto^9wQ-6V1qJPffSoRl(B#7?VkS*1ML zR7NJ;?UbtPP%`njxycU~0+x2sa^Ld(IK_qOU;iBiMf7E`^U3(ro~LckkC%=PZUqO6 zlVwlSj;DRg)BeJ9s2m8bx&FijzVHMdj@%!4EbPR3w_?4|@04P_+p#ydJr{~&yPoK} z^wHQ0&k>8dJTzRqTyBm%THILt#k_joy~eFcHJU$qj(i!U)Q<-?)a_u`uC*WDYCf?Y z6sm`zt>*6SV9#!_Y5n@6cQ@WG9X|EEX*)Ww9ektW;ye6nm)FmHFje7EOIIa`yy4>8 zd(;nPp0)k0Kq4ZPfY{&He@47KIyH51^wQM>?MPd(+KxkXL))>@s`*ACLn~rio|!>; zI{0p#KnUB7+FU@iGod>L+ce=oi(tbLPb^R%EZFd2I6%+qz!8t*V{{=+BF$tq1^d;B zVSt?mQmTR`VmB0j(y|>AidSAbLgheP$=OzR_}9BXceL&Xj+UH9%N}2G5_ePL&0n(* z8OCLf@;7_1l}n!9#^I|f^oTfR$Re$UWf>1p+_1^me{RFw=41Pl0#t`JlWx<^Bb29; zNTj>!XF!OoFVEClj^Y~u$fHHBPO-o0%|>HP(aFZZWHX#~JUy`=^L1&3A2TaR;HGok z)7pV<)t0kTd``pFKCV^$P(0^M#4jz#>09w!&_Y8;$LADP))N9xoMMgA3Mlc2`ynRo z8xVnUj5Unutf6@$oL-RCIV({Ia^w_6C94>snDapmCF;jCWz(l9D57#}=Nk9Gx7&W| zxnsM1aE*K6Y%K>OJAuy61D(6!j&gYLvzE_X6~tU*+IHJclv;(7aJIrSp>umI8}U~V z>+@Gw9tGnx6Tf#JkDjA)pP^?Az5rUF7-<4co2xsh!Ubfth`tC!Dfyu5{K$EF#$$k~ zk-`~DF6Xsut_5NVC}8vhZ{*?X{ne7Q19xw3y1H1V-`w$@WDc90FVXD+et0hN2!I%X zLlh8R)$YjHsvO*JCVjCtA-lvgt4S*&mgng-!dHMAQjEs|fNg|dURsi|;IsT8&NQY| z4J#@9j=AYOGK>R27z6;%h*i1x2=KAD;iE{~S z)L>(!>H~xl{{X-x0HDX}Ck%PA!25KgPf~CXz2w@y;T*ib5;lJn9I;!(=gi**e{vec zQ0*2INTd53borl}tEvKuUzxd3SHbeD=8$D})d6_W1uyOg5$ynAXjFG+8oTc{b}$cc z2LQ`iE64PWM}oa-#ruG=JcYgEQI%QD!O3o{ zEYX(|^X6;e*jskEYEm*wsar;ZM$^#JnwpqR=`x0rsTK&>K>JjLo)tx6UrJ#)Ygipw z8=s~Yjuv1bR%_0x$y0Fb1MoV4YXBNC{Z@vEb2y_(%c?wrHA)BVq5qo#+I$T24;#mD z|MViJ>nqgyXVm-^>iY|t*g_M34L7gfH4{7Go~>}t(>G3Tg$Ig(J$r-+SNf1&D7W^N zTYAf_iL&QtIdbA_pJR-%l~AMH9zm!o61S?o_0+1WEn3x#DwPX8^kf=1xBtIRQeZ57p84jR|84)> zTkVYhq^eN@Yjggg^=r#wJWJOtcD}ZIGE#_4Mhj6&jF8b}gy9o0qK0xaR8WkV88g%Y zcx1#)#i9k(Xa=SQn8j*x@sS*l5Mzzex*!+gX7t>;#d^9wdQL+xFD@ei*Bf8t@MI)` z-^?|d>a*lRQX&V5kzOE1=C0Hb7gFG%JwjGK##*CoAy`PyW^*0<1Nt@Zm-h2dLx0gZ zensl(5~)LnhDY^d!|xs#9X@(^XjJd(?b|72#JOFDHSV2qJ=UY~52fzJ8PWWb`hGEd zjnXiG5;>g_%@bB>!eqMT>K7d5nr;w6;w(6jP;^%GOns(kPn&EFup0NGpGixEE6P{L zBR~WQp-Z2`<_j`U=ZOxx#+s+za%P0gP}bw2QGXe9p57paIS(buk5cpXKbeMe(Pd_ zn#QvH+gQhmgK%&U16(Gx?E0(n$7)SU+}vJOa(^T`YKgAfZMP%$Uir@Z;pp!}C#ypz z9}E>944$g&JzZ%zvlNh$@j8*z_|p*S=)S$Xy0-tS{98E7+W9thOL{$+y+iI$wiTUV zanNf?-P&>#39Q0p@~Alt@>}W~Z*G7y>&Ba9DV{q4=3-cw{Br!GsE7gJ+vK&2A6=}3 zJ6Jmq;}X9orS9*DM*}p;2U>e*C%0PjU17Lm%?MinM9}igNmm3)rYC~7bHQZ(kgz2G zzO{WYgc4=UF=lmUmY@-A2foCvvmOKtQv|1{rc9Po{nSKwoPkYd71=JHNo=r$Tyh*k zM2q9&=9Gu+6XD4sD_P~D&D!vN2--7kdm;e+5&<}{KA>G}tM>e2SyYEjoMo8lIn}Sk z%JIbgPZA}%3v)jLWyYf9WWQu00KX&j;!U%%kdHWJGbgbT5U{rzszJUjrG<+*n-Da zgapFt2&-jzsoEkuFzFc6wz-#`0v-N3+-Csw0HyRFNv86%7@?aUlgwk%`zIN$lHq!s z#B;T5Z>?=xEt{_?8~HbFp{@f|s*qGAvu}wsQK_CHL8TJY>H%qlN|&1}tp}Hg1bmeL W*k;E@D2+ZLi1qjR{_JKMoA4h9SdPO0 diff --git a/src/impakt/protocol/euro_ncap.py b/src/impakt/protocol/euro_ncap.py index 19dd79c..265504b 100644 --- a/src/impakt/protocol/euro_ncap.py +++ b/src/impakt/protocol/euro_ncap.py @@ -9,6 +9,7 @@ Threshold values are versioned — this module supports multiple protocol years. from __future__ import annotations +from pathlib import Path from typing import Any from impakt.criteria.base import CriterionResult @@ -64,9 +65,44 @@ def _points_from_color(color: Color, max_points: float) -> float: return max_points * color_fractions[color] -# Threshold sets by year -# Format: {criterion: (green, yellow, orange, brown, red, higher_is_worse, max_points)} -THRESHOLDS_2024: dict[str, tuple[float, float, float, float, float, bool, float]] = { +# --------------------------------------------------------------------------- +# Threshold loading — from YAML files or hardcoded fallback +# --------------------------------------------------------------------------- + +_THRESHOLDS_DIR = Path(__file__).parent / "thresholds" + + +def _load_yaml_thresholds( + version: str, +) -> dict[str, tuple[float, float, float, float, float, bool, float]]: + """Load thresholds from a YAML file for a given version.""" + yaml_path = _THRESHOLDS_DIR / f"euro_ncap_{version}.yaml" + if not yaml_path.exists(): + return {} + + import yaml + + data = yaml.safe_load(yaml_path.read_text(encoding="utf-8")) + if not isinstance(data, dict): + return {} + + result: dict[str, tuple[float, float, float, float, float, bool, float]] = {} + for name, vals in data.items(): + if isinstance(vals, dict): + result[name] = ( + float(vals["green"]), + float(vals["yellow"]), + float(vals["orange"]), + float(vals["brown"]), + float(vals["red"]), + bool(vals.get("higher_is_worse", True)), + float(vals.get("max_points", 0)), + ) + return result + + +# Hardcoded fallback (used if YAML files are missing) +_THRESHOLDS_2024_FALLBACK: dict[str, tuple[float, float, float, float, float, bool, float]] = { "HIC15": (500.0, 620.0, 700.0, 850.0, 1000.0, True, 4.0), "3ms Clip": (42.0, 48.0, 54.0, 57.0, 60.0, True, 4.0), "Chest Deflection": (22.0, 34.0, 42.0, 50.0, 63.0, True, 4.0), @@ -77,9 +113,17 @@ THRESHOLDS_2024: dict[str, tuple[float, float, float, float, float, bool, float] "Viscous Criterion": (0.32, 0.56, 0.8, 0.9, 1.0, True, 2.0), } -THRESHOLDS: dict[str, dict[str, tuple[float, float, float, float, float, bool, float]]] = { - "2024": THRESHOLDS_2024, -} + +def _get_thresholds( + version: str, +) -> dict[str, tuple[float, float, float, float, float, bool, float]]: + """Get thresholds for a version, preferring YAML over hardcoded.""" + thresholds = _load_yaml_thresholds(version) + if thresholds: + return thresholds + if version == "2024": + return _THRESHOLDS_2024_FALLBACK + return {} class EuroNCAP: @@ -87,11 +131,12 @@ class EuroNCAP: def __init__(self, version: str = "2024") -> None: self._version = version - if version not in THRESHOLDS: + self._thresholds = _get_thresholds(version) + if not self._thresholds: raise ValueError( - f"Unknown Euro NCAP version: {version}. Available: {list(THRESHOLDS.keys())}" + f"No thresholds found for Euro NCAP version: {version}. " + f"Check {_THRESHOLDS_DIR} for YAML files." ) - self._thresholds = THRESHOLDS[version] @property def protocol_name(self) -> str: diff --git a/src/impakt/protocol/iihs.py b/src/impakt/protocol/iihs.py index 0ce26ec..48b33ab 100644 --- a/src/impakt/protocol/iihs.py +++ b/src/impakt/protocol/iihs.py @@ -6,27 +6,55 @@ The overall rating is determined by the worst sub-rating. from __future__ import annotations +from pathlib import Path from typing import Any from impakt.criteria.base import CriterionResult from impakt.protocol.base import BodyRegionScore, ProtocolResult, Rating +_THRESHOLDS_DIR = Path(__file__).parent / "thresholds" -# Thresholds: (good_limit, acceptable_limit, marginal_limit) -# Values above marginal_limit = Poor -# higher_is_worse indicates that higher values are worse -IIHS_THRESHOLDS_2024: dict[str, tuple[float, float, float, bool]] = { + +def _load_iihs_yaml(version: str) -> dict[str, tuple[float, float, float, bool]]: + """Load IIHS thresholds from YAML.""" + yaml_path = _THRESHOLDS_DIR / f"iihs_{version}.yaml" + if not yaml_path.exists(): + return {} + import yaml + + data = yaml.safe_load(yaml_path.read_text(encoding="utf-8")) + if not isinstance(data, dict): + return {} + result: dict[str, tuple[float, float, float, bool]] = {} + for name, vals in data.items(): + if isinstance(vals, dict): + result[name] = ( + float(vals["good"]), + float(vals["acceptable"]), + float(vals["marginal"]), + bool(vals.get("higher_is_worse", True)), + ) + return result + + +# Hardcoded fallback +_IIHS_2024_FALLBACK: dict[str, tuple[float, float, float, bool]] = { "HIC15": (250.0, 500.0, 700.0, True), - "Chest Deflection": (38.0, 50.0, 63.0, True), # mm - "Femur Load Left": (3.8, 6.2, 10.0, True), # kN - "Femur Load Right": (3.8, 6.2, 10.0, True), # kN + "Chest Deflection": (38.0, 50.0, 63.0, True), + "Femur Load Left": (3.8, 6.2, 10.0, True), + "Femur Load Right": (3.8, 6.2, 10.0, True), "Nij": (0.52, 0.78, 1.0, True), "Tibia Index": (0.5, 0.8, 1.3, True), } -IIHS_THRESHOLDS: dict[str, dict[str, tuple[float, float, float, bool]]] = { - "2024": IIHS_THRESHOLDS_2024, -} + +def _get_iihs_thresholds(version: str) -> dict[str, tuple[float, float, float, bool]]: + thresholds = _load_iihs_yaml(version) + if thresholds: + return thresholds + if version == "2024": + return _IIHS_2024_FALLBACK + return {} def _rate_value( @@ -70,11 +98,12 @@ class IIHS: def __init__(self, version: str = "2024") -> None: self._version = version - if version not in IIHS_THRESHOLDS: + self._thresholds = _get_iihs_thresholds(version) + if not self._thresholds: raise ValueError( - f"Unknown IIHS version: {version}. Available: {list(IIHS_THRESHOLDS.keys())}" + f"No thresholds found for IIHS version: {version}. " + f"Check {_THRESHOLDS_DIR} for YAML files." ) - self._thresholds = IIHS_THRESHOLDS[version] @property def protocol_name(self) -> str: diff --git a/src/impakt/protocol/thresholds/euro_ncap_2024.yaml b/src/impakt/protocol/thresholds/euro_ncap_2024.yaml new file mode 100644 index 0000000..fb54bcd --- /dev/null +++ b/src/impakt/protocol/thresholds/euro_ncap_2024.yaml @@ -0,0 +1,76 @@ +# Euro NCAP 2024 Adult Occupant Frontal Impact Thresholds +# +# Each criterion: [green, yellow, orange, brown, red, higher_is_worse, max_points] +# Sliding-scale: values at or below green = full points, at or above red = zero. + +HIC15: + green: 500.0 + yellow: 620.0 + orange: 700.0 + brown: 850.0 + red: 1000.0 + higher_is_worse: true + max_points: 4.0 + +3ms Clip: + green: 42.0 + yellow: 48.0 + orange: 54.0 + brown: 57.0 + red: 60.0 + higher_is_worse: true + max_points: 4.0 + +Chest Deflection: + green: 22.0 + yellow: 34.0 + orange: 42.0 + brown: 50.0 + red: 63.0 + higher_is_worse: true + max_points: 4.0 + +Nij: + green: 0.5 + yellow: 0.65 + orange: 0.8 + brown: 0.9 + red: 1.0 + higher_is_worse: true + max_points: 2.0 + +Femur Load Left: + green: 3.8 + yellow: 5.4 + orange: 7.0 + brown: 8.5 + red: 10.0 + higher_is_worse: true + max_points: 2.0 + +Femur Load Right: + green: 3.8 + yellow: 5.4 + orange: 7.0 + brown: 8.5 + red: 10.0 + higher_is_worse: true + max_points: 2.0 + +Tibia Index: + green: 0.4 + yellow: 0.7 + orange: 1.0 + brown: 1.15 + red: 1.3 + higher_is_worse: true + max_points: 2.0 + +Viscous Criterion: + green: 0.32 + yellow: 0.56 + orange: 0.8 + brown: 0.9 + red: 1.0 + higher_is_worse: true + max_points: 2.0 diff --git a/src/impakt/protocol/thresholds/iihs_2024.yaml b/src/impakt/protocol/thresholds/iihs_2024.yaml new file mode 100644 index 0000000..3c28d17 --- /dev/null +++ b/src/impakt/protocol/thresholds/iihs_2024.yaml @@ -0,0 +1,40 @@ +# IIHS 2024 Crashworthiness Thresholds +# +# Each criterion: [good, acceptable, marginal, higher_is_worse] +# Values above marginal = Poor. + +HIC15: + good: 250.0 + acceptable: 500.0 + marginal: 700.0 + higher_is_worse: true + +Chest Deflection: + good: 38.0 + acceptable: 50.0 + marginal: 63.0 + higher_is_worse: true + +Femur Load Left: + good: 3.8 + acceptable: 6.2 + marginal: 10.0 + higher_is_worse: true + +Femur Load Right: + good: 3.8 + acceptable: 6.2 + marginal: 10.0 + higher_is_worse: true + +Nij: + good: 0.52 + acceptable: 0.78 + marginal: 1.0 + higher_is_worse: true + +Tibia Index: + good: 0.5 + acceptable: 0.8 + marginal: 1.3 + higher_is_worse: true diff --git a/src/impakt/script/__pycache__/api.cpython-312.pyc b/src/impakt/script/__pycache__/api.cpython-312.pyc index 9a1ec10e01814435469584a3971c280307ec7184..4abac9faca885049d14e0fb3c818c3f462e85961 100644 GIT binary patch delta 8702 zcmbtZeRNyJm4EZ3Crh#{*^)owD1QEmrNmb37}CTE33k39h2W4l5L06dy=O;>ESY(7 z5(@>Tkh3HdmXD#3QfSk{38YC+vm02l?7{~2>`B?~DK-vu&_b7N;q;vLbdN*A(uF^E z@13V7$ugY&u`ls`^X}Zab7$tx@7@_baNj>aPn^FhDX|Oi)bIWveB`>naXv+)@j+1# z*o)-pUG_dLIHGoS+3hks!!$N3heyYPd*k?u(YU-_)wFOds@KO=EiT_+@B*d!xK5GOkABa%guj91TZD9hIpXAE!}G4$A9y2cuCna#Jv>L{vLlUEiDvWaR=AgTY96B&*6F=g1-KvkUP0F)d2XIF`%Bo!cn@Qk1e{X`M^MTPX66tUl|EL6OT`LNV4 zsTO6CA{{oL!bOFrzX&*WiY1R@Rc(r_6ygGB$QEK{luTl%YEywxC917oUz_X4F|Spe@^r9-KO z_reL2e|fqn`!?tGCl%3 zDj=_iy{js6EE-aQ$ira@+bpxJI@tIFTX#_G%}>$}C8G_HAd+b56*pXRxGyy5an&QRmlOjKH zFsHSOeb?#{4_a7nMJ4OCN>!eDt&5UHdn$PA=86h-_X6w6q%@dyu%xtIz&~)uYNIK* zqZwTc`B|?biAUH)YpC6DEzNMJoCk%#KLXkP=`W$3ppmVzsS@qL(hIgmvXWh}1s3jt zPwf>Tumq)o%`s#3PFcM$^`E#rOtQaJwsuKRtMGQKt!I_#?Ue*5`-!8Exr=KnyP;LW zWh^enfWfe`k9CxI*v{g;Ej_3ix2f(LbOVZ)8sejBJcz+F!R5ay9w9v}pMsaX<3K1>Yv9U-vgb}JRqY30fR-dM7^3B_CGt|hiBA8o&v*28C zkD=j_-8}7zMpKaZ$dGC2c~>Uh#iF?xzdf6PDdA9Djtv_@5fB*{TW(t`uh$c+SoC(D z^yC^F*j#}j64E7}l4VmV0Lw+G49xCV@8PLdED9N+7BB)Y9e~;6OAa30gk@ypFJrp2 zP;H{xSgOLB)7Tav@R0DiL$KXUF1xO4I=Ki+WXOUB}CCLv3#s0kM6ygm$4-kKGume8UXHsNivbn>}%|n7R4aU)Z zjD#f5)4FMDMC-}|DTKMKRF~YF$%SC&CA=DUk9m3QO4eMQD&sPrZa8E(I(t| z(Y?{w{%GE8C42!HHm>|l(**ldpQ(V~LNRqq*WkWM?S+#i7#@?K!wxyp;48z&t%UMi_LR`Xs7 zRIdE`%F6dj>OSgpTtC0|6ZgVP?xy2+9^Wz<*g08x$5oT)an1^2iIeYc%yiC1-Gq=X zHs;uoh*_Gu#7Q4Dp(vctN=r&|YT6`j+ekr!f z9Pa24Fi)CbzB1=ZI?l?f%Ims-O>21r1oDZr1zu`Yp=DUn%{EkZkZo*V)gpNu8T*i6 zj7^)VI;_##@VggDoV`-Doc!+W@2gUzsW1XP3KF^Krv>+^7Q!B``D)n^R#-F0=%WYm z(`Ez-ElvEe?2W+GJL+byEa(x73H$Sc7m6Q8wv42Qov8JRF2W{j`%5+f$3W%V*tZsT zif+Op3wNbN`V0ym!+JAMK@xT?PUR#|Tl7jVRG!QT-m5F(jcj>kJZL}%$WIjWj3+5? zg#eLhm#{_Pz;G969x$XlI{*e)lh>W%(}jxl?zW;TGwnl0mYXQy3NjhVz`S~jIj*6) z3d0|E)@s~DY{rZs+jSsU4<+`hw;?kRi3PcufupDq5pr%2Vx;v)u?3u1ZHUb7f$(P)yT!3Pv%#qIV%dWaB_WDVDy61Ra`Uq8Fx5Vex4oCk4Khb>KlyqDU~&(pn&{ z9Tu*LPRTke3$^m3x$2r(k~(Ldf~D*;(JpN!S4~iy<;B@*&Rrl?WtdQ!UCAc>wd{_% z#U-LthkWO(2qm_!Zb_p>s=rz)O1`UBQSu{c)JqPl)N!r6R&ril%l^JDk~$Bz6kOW4 z6cH{b%&Mf6DyF>q5({w3;3_Bo0!M-i9lG@(0zX~ z2zd%_bdWRWLPsgb>zsb4QlV7BHICHp0yX^lX2R*OqPTB@tuUwTO}S@i?$w;W{&=n` z$lG&wc2RXdRL)8HhlW~th*qPJ$YN#zFv)NK;L?#5)vvrIdtnmVojVN#Lc<|nM|*p_ zKaZEa^ULGKX?WBZ9|T;zp*_&Seo?=g2<(sbJvQ)jobfK5M@l}dX?fW^UDI~N{HPPoJ+;j1sAvD!P-pWKdIAIi|2}D0EH)A5XxvlFIefnq zg79tZiN-3?PuR~Iqq-P|z4!h2KUZWf0oN*)eTwsJGeQ; zS`e^F9fXu>zE!V6--((zE)mF+?*?6&ljz4T&tnIV9-MLcrr5H@RqVIzHa6>8W21GT z0+#axyV%@BUSt2$Tq|xQXDb(PBzK|X)BCaBNT&K5u;WaKvq#4QQ6)%ea3aG>OT&>^5MlWos(3T22TS0SL-;c1!94GT zqk0*AK84~xLh@rI$B`g>)1M;w84~Wj+&8($ok1R-m9tpFrJ}DP;Yf@iOkY5PrzCw= z5Fa>5;GLuP5AcQHTO?Vpx&bo->~((|xx#L&t78A_uVY*5Y{t9EpFvgvIBEfM%~Ax! zO{7L$J!)o;n<`Vgp-=#&_AHbZ&xKOOfk4_{1b3Px#;tMY?$`;CD)n0x7r<{3lx%== zZU9WF->#!%fANIHUrt|w?h?l`zf8dI3+(kJ>u#QhVe)5LxLL(}rE&3#+Ycn4Ga7H~2s z1SJiGjLV}FEV#4+u$gx{h^^hcHEy8f(d3M)Y0A|!X>H;Y8xo<;c+geA%gI#*3VD|l zaA=KV9e?03D<;Wl6FXP#Wj8PIvj1MXBy|9){8HLDOk`&OftqKz+$-pJL5v%1?5+u9 zU=}=gck%qhy~CJCIHnz#Kvf|e4bIRZ;Zy7KPh1Osuu5FXnZTbLl+ZvA*3UCxAS9LGS-on6%d(HsOiaAM=aDsnMOqyOKe@)o=@(MoK ziniM===6SxjkeVUEK<) z4;=1E$ChJ@&n%j9Y&pLI$c&?R%F#Qyt#4-A-Ba7{o^}kvxuw(&=N6lt4{Ph3g+uV* z3wMC62sBpC84HXcX+CW}C9>DcYxx+yAGm9&ME@1-`!y22YjaDJIqzBa!~IZfKm9eU zT(<5ubTF-xTm7lE<8oc|6FXmCbJTuBI--3D%dqoI#re*Q-pw=KTc^CY{!)2&^qtXZ z?{}tM1C!Q)`Ria|ZGoy9Q^Bmm?MYHRZOXIs^JVLa#8$O0Ts-qNhMIMRzJygDAemwN z+FMfzG<9*tRPH1@UT!&8anZYO#=CLKyYakoVe~DqZpXB%Z_?WL|J(+rpf`N>6zOk) zy;JVs)7;sy$tTe?u<1ug3Z^*|Uw+H$*>fG8sY7VE;oU3HyYD=+^rCm|jCaG7cfB4?Vd;ImZK3~;#s5Hpdy-Fy9*7XOId=EqoQB{F1Cbd5 zk%LH!;0F|o^D~i1&Dp5aeAQx>ZYQ%&VOUtafm(^}x? zqxSFNJ5gdB;{tdTeVkpo{$vN9ct8QBAe1*k#5U3pe!kDO!UO)(Xa9D^0J)v#EkoD{ z1Ap3{g=7t|7c1eWG!r7=${#Ybs7C(?Y*-$&H7#LoqtPgo^@HdOZ0|~M`ycT80VMna zqZ&%!Kzww6LT*l;`_CWPxs_|kbL{h#&u-*G_o5K@9BfLT#r9@>aGb%b%*Zuu-Ku9a z4qZKO`6qU9RecIONt9-f`-ZcPqDC%O;Je&P+t)c*+5y*n@o*?}c>NSL;7aJ{%qv(b z@T}hnLXkj$3J6F8BeDkSk&ZF@GLlU#!`+JZK^f lR|KSA+*GuNG+q&ad~rWnM*7GV0qGYCX(#s21R&f^{|}-K$hiOj delta 4924 zcmbVPeQ;FO6@T~bdz;S4Z*gA>lPEqA7iUlYosyT~d1r|(!c4!r~Ty;R}QvIqETDMxE zD$wSsE>%?BPl#3_`=`!O^9Id9k6NjEfGZ!&&4;!?ov9W;>s6~%FSLbw#XumP)1g@s ziLQ7oJddy!%tzR5Irms>K6u(ACf2hcfA|B7pO881AJ!K3%(R01O$42#0~${sIZ9Z# zuw3jX?2~EpMcv9O906iu?T*4?dxq;;+i&hO?{KpHj>1BBhMIA1e@Q0hIcHUi1Zh0gx z!cr~=o9$8Dlf{>_HJ&LX$T~fD0Q=jXS>$&1m1mGNv)%d4)_{rnaFSYIq@ zU_kiP;byUd*F3A?Eoslp-q{sXyV&*Kqcd(qRuRBAZe|-YIx>vgXuP+lo4E^zW=u9{ zEiyVMFedI~e=a;Zdot(E$eEwvjBF3bVp=q$Q%&n-r;GgT)OGgZ=cm~Ssb#+{t0xWY z%d(=5^+MdxG3wLXDC6eTXzBj7< zTdej`Tp!&8Fl0(w#yJyoGjiqZ4MoFSwP?VdabV0}N*1yQ{B>j{`@Mgf57}rCVL1v& zRO?L8R(xN9u#tV^pGWQ+o?DS36_ZzfCvaTzdLq0_J0`FXX8gy6#w0D{tE-}6cD}OM z%#|)=-&8h>Cj_>9<}E=Os4D1a0!l!n6*$n!QdRZhDS`c_YI91W z(}9+>C3NcR^Q{Z(+vz@JDg_u4jsAM*gD|ICv-*~6JV|S`E1};CTDFXHCOzxNec#ce zML>6ZJf;OC!w;yD>+{eSR4CCkNHNIlbfBOx>m5gOkPA#;Uj)j?K9)P{I+H{C9JW!k|nDB#U^?Bm`#+c#w?;7K=2zqJA1i0 zntA{fPO4^hRfL3(wafukExVSyoT}y0_N7eFb-_?F-S2)_MH8s@KvXZ%r31J)OHe2B^@YA!B zJw3Ntd`n;>bMq}ZXNqg@l^%4m(YzA&GtV?Oqqf{SMCkpvIgM;xZKe3G!1mPcDB%=- zTqmFSR(7G@!(4Te_`blrbuojLhsk8trO4{QBJcrD@CX3Re9Kf+o|g13NMw^m9_1o9 z^rvhQW0IXnnk=!COZ3LN^z>DU+r1mD6w3pg!TVWWL&dbm;U($4!N|<3!Z9_f)o{0k z!ult|@cM>Uax0JRtte%o9@OPV;GWn9y>zasB_gz|M~4do<+dou5Cwqh#J#;42^#?t zCIHx1Bjn>?W4;`X>lXy(Sa33B_#4@>$|oL~$tH$ERO_Lk(BF|4%^46rRy+^49Bez# zc19^5v6SVF*(aG^wO5)>9Kx!<2^iLiYFx(9&{eG#=Is z;U|%o@PY!vd{I|ZAmypZN$&z4dN;!N5e5-3uh1VLJdD8OheyVv_{v>|8?n$x6g)@V zgCp0HLkqql{0{yJyc}YBFw2%PmnhdUPvac+aAOf0Xe{%NnTZ^@Y|a75BeLI6(!%U> ziX$69Zc-Mo8d*TKVGan&Rzc0pzBaJ|H==j#(gN&nix$}EL%>NN1YnC6&&sQ_o5lk~_&K72;mI%3(%SJlq; z+r6Zdk!AjT?tfFCbWA#m_bPZM*xkQuU8O|v22GD4T=TVIU)sDIE}!w;^cef+vSmwU z*u`1Jbgsz1k`7RUxwS3`HDP8<6uR`(Q1&E%UXt-oJ_C=(=gm9AS*H_Q7 zqv_M(eL&!G;dy-Jk%qHM-8p5!sIuUcblUMUSi54xvf}@V0Uv&=Ew15 ziYb7;!d`5yPvxN{nFyVKK`A`6=B!eCPMJTd%s;Nb)PJJ?jM6${Y2{v{zW_nVTRi45 zg&0WKYA&E4TxZ#aq>oj;K5c|h?5hfzhJL>NMl5g=ox1S5x{1)%siG|4G?FO>3I zdDg7<@n*|}D;v&7e~l7mik|(qoik;4xSvlW4@_X_2i1Jd+<!E`L+F5 zI692*GQz6}c(M#{%)}lhEe>BoCxL6lylR5HY9q35%q+<_5V(9TYbF;>W;sGW^cIhI z<}bczs^_G(bq48Et7}X~%4L@MGm;V!=JU(6Gtt)LE&kI zhBh$&iV8CIkySJHSOd5pa>T2mr!~uk<}6`~l&Q2t1tLVc)D?0-xcH zH~zGhGx36 zp*(J-^}FhK+MdwEdRJupo8V2f3AeJ(NP`aR`p>#v(g3<$} z8%i%<#d$9G5QIeA2%frM>EvVR7X1e@1v=N@;A)RsmJ4!m?}+ z?DvZ!DKbgUIV_+M=Xl^OLD`bFIXugz#8Djmz*T~>C+%~tO(4+Z+?=NjMH1UU;VrHP zIP;jcgR9|c72}ZcOq`Fadr*JBxPonIH*k&Ijt4s{Xj`wRiEA!%nnFqRLhG@n)xzzn zpyfz9UWk@8R4Z4@wQblNYq)k;=k6C+C&TTj;LVwIz7TKLPBMJQZK}xH59&A`!~#K4TpK_1MUL%;T7Qow~3sqD%30UByqp^ z!cugSo~y1LE7Up4c=}$&XO1Ol69+^sN9ILPl8)d}>S&;jfD}tP0ML{(rV;>A~jKs%dk|IQB{5BoMrPz$_;dnVL32`hv zbt+sCx@z6<6;=@0X_04&T{6Z1SVc#(ycpLVGA~`>Ro_`Q5e>uqvph4;Pcv_h`$@?I z90GZe!s;$8iy0+W!5}9{d{~J|3#vO1Q20n(WEEbu1r*hHvDm^0Y&#h&guP=mEdqt|Mh3&g)dM-(W$d0s)SnPR!_T8Rp|dij)A3 zUx^y)GDet4jN@TtPBx4;pvbupy%baHF3iT}3v5Stg~fZT)8SdrHM}URRc8g*@KVzS zY_baIsHo~26GW&m({OObE?9&j@x1CaNU}16TRP2OVZ{Vqd;1%SD5$u(1ROS;1z-y- zQyc}Z@IJv=DY0-&gq6DZgcJ)!!)#oq5^|w}1YuT|@CAp(P~q8_5awm62HGke2*eUf zJc0Kxrq-M{>dfTm7-nx)cNVUM>O0Lxc?q~-*y1fwYu+$!3+zGRj4lFgTH@bLzyWf` zR!#V;b()oC;;ba|diAh0lZe2boy1K^uVy(eBqyegQAyU_xLs)IUR(!E175eo1=roi z5bl&EtR!`JC>#L-_@3!5qmB(146XPw&7ajmX^#x>Nwiq4e#lG%&eu`Ru7_Q-OF_6lqrK)?O1$CvlC<~*&L ziz}YJi=*FCo*T()$@Kgas(qcVOCNbkH?6spO&5wwWZ6-Xd0nL>xf-)>%?kKm+nl-%#4%4y+n}zT@CnW4N+#_}B z0x!!TxxpE6hb11QY6#Ss+FMlH1x3k7iD(p5F~cg%tfIt+`uoLLm=$MZvN8ms)=|)1 z2|8zH`-(GRIEsm@sm*#!x zk@9%qvxP^OHTS7iG_r=M8>g?GPIuhyz16#P;=z1GaMU)?7PI^v=%&pa8PC{ho{{0u zwrqu&Yu(x>4OOg#2JT)E54dw}1KqxL7Jznt?AX8`vlVI%Kb%cz1AiWvk=XC}}1SJ-O1%W%Jqi(oH|9}y8%4Cb-Tz%b?1!1u^&}PW9<0pWJJa~8jIk91m1^NfwS}QmnT4N8xq(; z{9?J=l%ty#hyRLPH^346xck8`$M62#t zr91J+``6#cBks(>yC?6Q)LQmy)qNni@zBDxg&S9|U&X_&%$~dcJN~=DJ3+14pY>%A zJ{-I^`0&KN6I$=k@}A}4FUCI~|Kj}T=d~Z4)vC{}(i51#{I&T!)%=8N#w4f=d6&9F z<(qnQO}*OA{nE_gxhV?j4Ct7fXFilO1OJpvM>TinD(b>ruYPzT zv-58Io%WSx|4L)8*05i5_gShLd}RzZ^N;^uYKE{vdNU`32c ztz23u6KTgbuokK$>H$~_gCOOpI00l0!+AeKTvZZ22Nm#jFI-<(tJ#sS*_W%?w^Gx26Ro#*EM5F`>XWJ8yz_PY zp${f*j-;E{y}tCRw4CPCy{q1~_0Ij-@rUQ{onPsD`RmRTx!OHxd-|RA+Qy76b2!tP zo?ETmv)yE%r6O9`O%@r1D_2%@-H7+p8i7oTwMO?S4Xw>*Pk}M@i#lMyZdhrqI-mdyz3-n zyAMCBBM!avxb=6f>4DqFZyjGkpL##>uGR;Bf8;CbkKTXxu8h2SlU}cD$~5J=2l92t zZk}4}?R(gLuRGs6^rUwvedzYUt$}=f*OU6LFKz!u{VR2Iw6Iv|_WZ5+jGT@v9bCGY z-M4fm%RaI_raq&V?b+1waEZAiTHWMR@9V%!{oq!|s`UNr_iz*&9V>_s*hM2M5J4$> z#F0V~XUY+wQt(j_oUT%Q9}tTPUX_K+FkdogkCImoemu2}Bz9O!Pmt@*4aRMmR)ypS zIYC#*3BN$K99sNJ0vUF_u$3uN0!U z*ntPxGA}@wTeSy+!Ndrxvu7+JiVT}jVi18=3g`@cq+}L&g#m|MG@b*`84>)1%p39Z zVjLa_4o{Bz{Z-|6uGbWud^0*MfmA0X1!q`{NLO$+kF%0dTQFrQ!(58|z zEtz4UWe6wpT^8PCC!PXlDjt6oGQ-&X4>b7{O>Q+gzjG4~_s!Qb)ZLmpHQLUetTU@T zOx;UmW6Kvb_wXt@g<Pc|hKJ>A<)um{|p3T>fawk>+JF!Y;N6b)*=no+-qz;jBuV z&6Yl7T!ElEd{hZC5S87!-$D3pfsKDCaQve}q)T1UHnJOdzysMLv5{|`4ux>!HWZSk zU>L0S9w^jWEJlAUE7DJxBr_3I23|_c<53B@OhXA?~a7OCFyo~p&-C#MPXV(xc?B&jL^(JT$e!6rlKh6~`EO$HUx=DN5$-<|P2V|> ylC96fW None: self._data = test_data self._session_mgr = session_mgr or ( @@ -49,12 +73,29 @@ class Session: ) self._template: TemplateSpec | None = None + @classmethod + def _discover_plugins(cls) -> None: + """Discover and register plugins. Called once on first Session.open().""" + if cls._plugins_discovered: + return + cls._plugins_discovered = True + try: + from impakt.plugin.registry import discover_all + + discover_all() + except Exception as e: + logger.debug("Plugin discovery failed (non-fatal): %s", e) + @classmethod def open(cls, path: str | Path) -> Session: """Open a crash test from a path. Auto-detects the format using the reader registry. + Discovers plugins on first call. """ + # Discover plugins (idempotent — safe to call multiple times) + cls._discover_plugins() + path = Path(path).resolve() registry = get_registry() test_data = registry.read(path) @@ -87,6 +128,11 @@ class Session: """Underlying TestData object.""" return self._data + @property + def path(self) -> Path | None: + """Path to the test data directory.""" + return self._data.path + @property def channel_names(self) -> list[str]: return self._data.channel_names @@ -98,7 +144,10 @@ class Session: # ----- Channel access ----- def channel(self, name: str) -> ChannelHandle: - """Get a channel by name, wrapped with transform convenience methods.""" + """Get a channel by name, wrapped with transform convenience methods. + + Returns a ChannelHandle with a fluent ``.transform`` interface. + """ ch = self._data.get(name) return ChannelHandle(ch) @@ -118,6 +167,47 @@ class Session: """Hierarchical channel tree for UI display.""" return self._data.channel_tree() + # ----- Criteria & Protocol ----- + + def compute_criteria(self) -> dict[str, CriterionResult]: + """Auto-detect channels and compute all applicable injury criteria. + + Uses ISO channel naming to find the right channels for each criterion. + Returns a dict of criterion name -> CriterionResult. + """ + from impakt.web.components.criteria import auto_compute_criteria + + return auto_compute_criteria(self._data) + + def evaluate(self, protocol: str = "euro_ncap", version: str = "") -> ProtocolResult: + """Compute criteria and score against a rating protocol. + + Args: + protocol: Protocol name ('euro_ncap', 'us_ncap', 'iihs'). + version: Protocol version (optional, uses latest if empty). + + Returns: + ProtocolResult with stars/rating and per-region scores. + """ + criteria = self.compute_criteria() + + if protocol == "euro_ncap": + from impakt.protocol.euro_ncap import evaluate + + return evaluate(criteria, version=version or "2024") + elif protocol == "us_ncap": + from impakt.protocol.us_ncap import evaluate + + return evaluate(criteria, version=version or "2023") + elif protocol == "iihs": + from impakt.protocol.iihs import evaluate + + return evaluate(criteria, version=version or "2024") + else: + raise ValueError( + f"Unknown protocol: {protocol}. Use 'euro_ncap', 'us_ncap', or 'iihs'." + ) + # ----- Template ----- def apply_template(self, name_or_spec: str | TemplateSpec) -> None: @@ -188,15 +278,16 @@ class Session: class ChannelHandle: """Wrapper around a Channel providing fluent transform access. - Example: + Each transform method on ``.transform`` returns a new ``ChannelHandle``, + enabling chaining:: + ch = session.channel("11HEAD0000ACXA") - filtered = ch.transform.cfc(1000) - aligned = ch.transform.x_align(method="threshold", threshold_value=5.0) + result = ch.transform.cfc(1000).transform.y_align().transform.trim(t_end=0.1) """ def __init__(self, channel: Channel) -> None: self._channel = channel - self.transform = TransformProxy(channel) + self.transform = TransformProxy(self) @property def raw(self) -> Channel: @@ -207,6 +298,10 @@ class ChannelHandle: def name(self) -> str: return self._channel.name + @property + def code(self): + return self._channel.code + @property def data(self) -> np.ndarray: return self._channel.data @@ -219,6 +314,14 @@ class ChannelHandle: def unit(self) -> str: return self._channel.unit + @property + def peak(self) -> float: + return self._channel.peak + + @property + def sample_rate(self) -> float: + return self._channel.sample_rate + def value_at(self, t: float) -> float: return self._channel.value_at(t) @@ -238,44 +341,51 @@ class ChannelHandle: class TransformProxy: """Fluent transform interface for a channel. - Each method returns a new Channel (non-destructive). + Each method returns a new ``ChannelHandle`` wrapping the transformed + channel, enabling chaining. """ - def __init__(self, channel: Channel) -> None: - self._channel = channel + def __init__(self, handle: ChannelHandle) -> None: + self._handle = handle - def cfc(self, cfc_class: int) -> Channel: + @property + def _channel(self) -> Channel: + return self._handle._channel + + def cfc(self, cfc_class: int) -> ChannelHandle: """Apply CFC filter.""" from impakt.transform.cfc import CFCFilter - return CFCFilter(cfc_class=cfc_class).apply(self._channel) + return ChannelHandle(CFCFilter(cfc_class=cfc_class).apply(self._channel)) def x_align( self, method: str = "manual", reference_time: float = 0.0, **kwargs: Any - ) -> Channel: + ) -> ChannelHandle: """Apply time-zero alignment.""" from impakt.transform.align import XAlign - return XAlign(method=method, reference_time=reference_time, **kwargs).apply(self._channel) + return ChannelHandle( + XAlign(method=method, reference_time=reference_time, **kwargs).apply(self._channel) + ) - def y_align(self, window: tuple[float, float] | None = None) -> Channel: + def y_align(self, window: tuple[float, float] | None = None) -> ChannelHandle: """Apply Y-axis zero correction.""" from impakt.transform.align import YAlign start, end = window if window else (None, None) - return YAlign(window_start=start, window_end=end).apply(self._channel) + return ChannelHandle(YAlign(window_start=start, window_end=end).apply(self._channel)) - def trim(self, t_start: float | None = None, t_end: float | None = None) -> Channel: + def trim(self, t_start: float | None = None, t_end: float | None = None) -> ChannelHandle: """Trim to a time range.""" from impakt.transform.resample import Trim - return Trim(t_start=t_start, t_end=t_end).apply(self._channel) + return ChannelHandle(Trim(t_start=t_start, t_end=t_end).apply(self._channel)) - def resample(self, target_rate: float) -> Channel: + def resample(self, target_rate: float) -> ChannelHandle: """Resample to a new rate.""" from impakt.transform.resample import Resample - return Resample(target_rate=target_rate).apply(self._channel) + return ChannelHandle(Resample(target_rate=target_rate).apply(self._channel)) class Template: diff --git a/src/impakt/web/app.py b/src/impakt/web/app.py index a724bc6..a6af628 100644 --- a/src/impakt/web/app.py +++ b/src/impakt/web/app.py @@ -1,7 +1,7 @@ """Dash web application factory. Creates the Dash app with all layout components and callbacks registered. -The AppState holds server-side data; Dash stores hold lightweight UI state. +The AppState holds Session objects server-side; Dash stores hold lightweight UI state. """ from __future__ import annotations @@ -13,14 +13,12 @@ import dash import dash_bootstrap_components as dbc from impakt.channel.model import TestData +from impakt.script.api import Session from impakt.template.library import TemplateLibrary from impakt.web.callbacks import register_callbacks from impakt.web.layout import build_layout from impakt.web.state import AppState -if TYPE_CHECKING: - from impakt.script.api import Session - def create_app( session_or_data: Session | TestData | None = None, @@ -37,21 +35,15 @@ def create_app( Returns: Configured Dash app ready to run. """ - # Build or use provided AppState if app_state is None: app_state = AppState() if session_or_data is not None: - if hasattr(session_or_data, "data"): - test_data: TestData = session_or_data.data # type: ignore[union-attr] - else: - test_data = session_or_data # type: ignore[assignment] - - # Create a LoadedTest from TestData directly - from impakt.web.state import LoadedTest - - loaded = LoadedTest(test_data) - app_state._tests[test_data.test_id] = loaded - app_state._test_order.append(test_data.test_id) + if isinstance(session_or_data, Session): + app_state.add_session(session_or_data) + elif isinstance(session_or_data, TestData): + # Wrap raw TestData in a Session + session = Session(session_or_data) + app_state.add_session(session) # Discover templates if template_names is None: @@ -64,14 +56,13 @@ def create_app( # Title title = "Impakt" if app_state.primary_test: - title = f"Impakt — {app_state.primary_test.test_id}" + title = f"Impakt \u2014 {app_state.primary_test.test_id}" app = dash.Dash( __name__, external_stylesheets=[dbc.themes.FLATLY], title=title, suppress_callback_exceptions=True, - # Prevent browser from caching old layouts serve_locally=True, meta_tags=[ {"http-equiv": "Cache-Control", "content": "no-cache, no-store, must-revalidate"}, @@ -92,16 +83,9 @@ def serve( port: int = 8050, debug: bool = False, ) -> None: - """Convenience function to create and run the web UI. - - Args: - session_or_data: Session or TestData to visualize, or None for empty app. - template: Template name to pre-apply. - port: Server port. - debug: Enable Dash debug mode. - """ + """Convenience function to create and run the web UI.""" if template and session_or_data and hasattr(session_or_data, "apply_template"): - session_or_data.apply_template(template) # type: ignore[union-attr] + session_or_data.apply_template(template) app = create_app(session_or_data) print(f"Impakt running at http://localhost:{port}") diff --git a/src/impakt/web/callbacks/criteria_callbacks.py b/src/impakt/web/callbacks/criteria_callbacks.py index 9456bef..9d1fecc 100644 --- a/src/impakt/web/callbacks/criteria_callbacks.py +++ b/src/impakt/web/callbacks/criteria_callbacks.py @@ -1,9 +1,7 @@ """Criteria computation callbacks. -Handles: -- Compute All button -> runs auto_compute_criteria -- Protocol scoring -- Results display +Uses Session.compute_criteria() and Session.evaluate() from the +scripting API — the same path used by programmatic scripts. """ from __future__ import annotations @@ -14,11 +12,7 @@ import dash from dash import Input, Output, State, html from dash.exceptions import PreventUpdate -from impakt.web.components.criteria import ( - auto_compute_criteria, - build_criteria_results_display, - score_protocol, -) +from impakt.web.components.criteria import build_criteria_results_display from impakt.web.state import AppState @@ -41,8 +35,8 @@ def register_criteria_callbacks(app: dash.Dash, app_state: AppState) -> None: if primary is None: return html.Div("No test loaded.", className="text-danger small") - # Compute criteria - criteria = auto_compute_criteria(primary.data) + # Use Session.compute_criteria() — single code path for script + web + criteria = primary.compute_criteria() if not criteria: return html.Div( @@ -51,7 +45,10 @@ def register_criteria_callbacks(app: dash.Dash, app_state: AppState) -> None: className="text-warning small", ) - # Score against protocol - protocol_result = score_protocol(criteria, protocol) + # Score against protocol — use Session.evaluate() + try: + protocol_result = primary.evaluate(protocol) + except Exception: + protocol_result = None return build_criteria_results_display(criteria, protocol_result) diff --git a/src/impakt/web/callbacks/plot_callbacks.py b/src/impakt/web/callbacks/plot_callbacks.py index 9add249..643e69e 100644 --- a/src/impakt/web/callbacks/plot_callbacks.py +++ b/src/impakt/web/callbacks/plot_callbacks.py @@ -1,10 +1,11 @@ """Plot rendering callbacks. -Handles: -- Updating plot figures when channels/transforms change -- Cursor line rendering (X1/X2 vertical lines) -- Hover data is NOT shown as a Plotly tooltip — instead the cursor grid - picks it up via the hoverData callback +Builds a PlotSpec from the UI state and delegates to PlotEngine.render(). +This is the single rendering path — the same PlotEngine used by the +scripting API renders the web UI plots. + +Transform application uses TransformChain, making the pipeline +serializable and reproducible. """ from __future__ import annotations @@ -15,16 +16,55 @@ from typing import Any import dash import numpy as np import plotly.graph_objects as go -from dash import ALL, Input, Output, State, ctx +from dash import Input, Output, State from dash.exceptions import PreventUpdate from impakt.channel.model import Channel -from impakt.plot.engine import DEFAULT_COLORS +from impakt.plot.engine import DEFAULT_COLORS, PlotEngine +from impakt.plot.spec import ChannelRef, Corridor, CorridorStyle, PlotSpec, PlotStyle from impakt.transform.align import XAlign, YAlign +from impakt.transform.base import TransformChain from impakt.transform.cfc import CFCFilter from impakt.transform.resultant import resultant_from_channels from impakt.web.state import AppState +# Module-level engine instance (stateless, safe to reuse) +_engine = PlotEngine() + + +def _build_transform_chain( + cfc_value: str, + y_align: bool, + x_align_method: str, + x_align_value: float | None, + per_channel_cfc: str | None = None, +) -> TransformChain: + """Build a TransformChain from the current UI state. + + Per-channel CFC override takes precedence over the global CFC setting. + """ + chain = TransformChain() + + # CFC filter + effective_cfc = per_channel_cfc if per_channel_cfc else cfc_value + if effective_cfc and effective_cfc != "none": + try: + chain = chain.append(CFCFilter(cfc_class=int(effective_cfc))) + except (ValueError, Exception): + pass + + # Y-align + if y_align: + chain = chain.append(YAlign()) + + # X-align + if x_align_method == "manual" and x_align_value is not None: + chain = chain.append(XAlign(method="manual", reference_time=x_align_value)) + elif x_align_method == "threshold" and x_align_value is not None: + chain = chain.append(XAlign(method="threshold", threshold_value=x_align_value)) + + return chain + def _resolve_channels( selected_keys: list[str], @@ -35,12 +75,15 @@ def _resolve_channels( x_align_value: float | None, show_resultant: bool, ) -> list[tuple[str, Channel]]: - """Resolve selected channel keys to Channel objects with transforms applied. + """Resolve selected channel keys to transformed Channel objects. - Per-channel overrides (from app_state.channel_overrides) take precedence - over the global CFC setting. + Uses TransformChain for each channel. Per-channel overrides from + app_state.channel_overrides take precedence over the global CFC. + + Returns list of (label, transformed_channel) tuples. """ channels: list[tuple[str, Channel]] = [] + multi_test = len(app_state.tests) > 1 for key in selected_keys: if "::" in key: @@ -55,29 +98,22 @@ def _resolve_channels( if ch is None: continue - # Determine CFC: per-channel override takes precedence over global + # Build per-channel transform chain override = app_state.channel_overrides.get(key, {}) - ch_cfc = override.get("cfc", "") - effective_cfc = ch_cfc if ch_cfc else cfc_value + per_ch_cfc = override.get("cfc", "") + chain = _build_transform_chain( + cfc_value, + y_align, + x_align_method, + x_align_value, + per_channel_cfc=per_ch_cfc, + ) - if effective_cfc and effective_cfc != "none": - try: - ch = CFCFilter(cfc_class=int(effective_cfc)).apply(ch) - except (ValueError, Exception): - pass - - # Apply Y-align - if y_align: - ch = YAlign().apply(ch) - - # Apply X-align - if x_align_method == "manual" and x_align_value is not None: - ch = XAlign(method="manual", reference_time=x_align_value).apply(ch) - elif x_align_method == "threshold" and x_align_value is not None: - ch = XAlign(method="threshold", threshold_value=x_align_value).apply(ch) + # Apply the chain + if len(chain) > 0: + ch = chain.apply(ch) # Build label - multi_test = len(app_state.tests) > 1 label = ch.code.short_label if ch.code.is_valid else ch.name if multi_test: label = f"[{test_id}] {label}" @@ -99,7 +135,7 @@ def _resolve_channels( res_label = ( f"{res.code.location_label} Resultant" if res.code.is_valid else "Resultant" ) - if len(app_state.tests) > 1: + if multi_test: res_label = f"[{comps[0].source_test_id}] {res_label}" channels.append((res_label, res)) except Exception: @@ -108,123 +144,61 @@ def _resolve_channels( return channels -def _build_figure( +def _build_plot_spec( channels: list[tuple[str, Channel]], cursor_x1: float | None, cursor_x2: float | None, - cfc_value: str, corridors: list[dict] | None = None, -) -> go.Figure: - """Build a Plotly figure from resolved channels.""" - fig = go.Figure() +) -> PlotSpec: + """Build a PlotSpec from resolved channels and UI state. - if not channels: - fig.update_layout( - template="plotly_white", - annotations=[ - { - "text": "Select channels from the grid to plot", - "xref": "paper", - "yref": "paper", - "x": 0.5, - "y": 0.5, - "showarrow": False, - "font": {"size": 14, "color": "#bbb"}, - } - ], - xaxis={"visible": False}, - yaxis={"visible": False}, - margin={"l": 40, "r": 20, "t": 30, "b": 40}, - ) - return fig - - # Add traces — hovermode is disabled at the layout level; cursor tracking - # is handled by our own JS mousemove handler (cursor_tracker.js). + Channels are already transformed — they are wrapped in ChannelRef + objects with no additional transform chain (transforms were applied + during resolution). + """ + # Build ChannelRef objects + refs: list[ChannelRef] = [] for i, (label, ch) in enumerate(channels): color = DEFAULT_COLORS[i % len(DEFAULT_COLORS)] - fig.add_trace( - go.Scatter( - x=ch.time.tolist(), - y=ch.data.tolist(), - mode="lines", - name=label, - line=dict(color=color, width=1.5), + refs.append( + ChannelRef( + channel=ch, + style=PlotStyle(label=label, color=color), ) ) - # Add corridor fills + # Build Corridor objects from raw dicts + corridor_objs: list[Corridor] = [] if corridors: - for corridor in corridors: - if not corridor.get("visible", True): + for c in corridors: + if not c.get("visible", True): continue - c_time = corridor["time"] - c_upper = corridor["upper"] - c_lower = corridor["lower"] - c_name = corridor.get("name", "Corridor") - - # Upper bound - fig.add_trace( - go.Scatter( - x=c_time, - y=c_upper, - mode="lines", - line=dict(color="rgba(100,100,255,0.4)", width=1, dash="dash"), - showlegend=False, - name=f"{c_name} upper", - ) - ) - # Lower bound with fill to upper - fig.add_trace( - go.Scatter( - x=c_time, - y=c_lower, - mode="lines", - line=dict(color="rgba(100,100,255,0.4)", width=1, dash="dash"), - fill="tonexty", - fillcolor="rgba(100,100,255,0.1)", - showlegend=True, - name=c_name, + corridor_objs.append( + Corridor( + name=c.get("name", "Corridor"), + time=np.array(c["time"]), + lower=np.array(c["lower"]), + upper=np.array(c["upper"]), + style=CorridorStyle(), ) ) - # Add X1/X2 cursor lines - if cursor_x1 is not None: - fig.add_vline( - x=cursor_x1, - line_dash="dash", - line_color="rgba(220,53,69,0.6)", - line_width=1, - annotation_text=f"X1={cursor_x1:.4f}s", - annotation_font_size=9, - annotation_font_color="rgba(220,53,69,0.8)", - ) - if cursor_x2 is not None: - fig.add_vline( - x=cursor_x2, - line_dash="dash", - line_color="rgba(13,110,253,0.6)", - line_width=1, - annotation_text=f"X2={cursor_x2:.4f}s", - annotation_font_size=9, - annotation_font_color="rgba(13,110,253,0.8)", - ) + # Cursor positions + x_cursors = None + if cursor_x1 is not None and cursor_x2 is not None: + x_cursors = (cursor_x1, cursor_x2) - # Layout — hovermode is disabled; cursor tracking is handled entirely - # by our JS (cursor_tracker.js) which reads pixel positions from mousemove - # events and converts to data coordinates via Plotly's axis internals. + # Y-axis label from first channel y_label = channels[0][1].unit if channels else "" - fig.update_layout( - xaxis_title=dict(text="Time (s)", font=dict(size=10, color="#999")), - yaxis_title=dict(text=y_label, font=dict(size=10, color="#999")), - template="plotly_white", - hovermode=False, - showlegend=False, - margin=dict(l=45, r=8, t=4, b=28), + return PlotSpec( + channels=refs, + corridors=corridor_objs, + x_cursors=x_cursors, + y_label=y_label, + compact=True, # Web UI always uses compact mode ) - return fig - def register_plot_callbacks(app: dash.Dash, app_state: AppState) -> None: """Register all plot-related callbacks.""" @@ -271,10 +245,5 @@ def register_plot_callbacks(app: dash.Dash, app_state: AppState) -> None: show_resultant, ) - return _build_figure( - channels, - cursor_x1, - cursor_x2, - cfc_value, - corridors=corridors_data, - ) + spec = _build_plot_spec(channels, cursor_x1, cursor_x2, corridors_data) + return _engine.render(spec) diff --git a/src/impakt/web/components/channel_tree.py b/src/impakt/web/components/channel_tree.py deleted file mode 100644 index e4b68db..0000000 --- a/src/impakt/web/components/channel_tree.py +++ /dev/null @@ -1,221 +0,0 @@ -"""Collapsible channel tree component. - -Renders channels in a hierarchical accordion: - Test Object > Body Region > Measurement Type > [channels] - -Supports search filtering, select-all per group, and channel preview -(peak, unit, sample rate) on hover/expand. -""" - -from __future__ import annotations - -from typing import Any - -import dash_bootstrap_components as dbc -from dash import dcc, html - -from impakt.web.state import AppState - - -def _make_channel_item(ch_info: dict[str, str], show_test_prefix: bool = False) -> html.Div: - """Build a single channel item with checkbox and preview info.""" - label = ch_info["label"] - key = ch_info["key"] - peak = ch_info.get("peak", "") - unit = ch_info.get("unit", "") - - preview = f" ({peak} {unit})" if peak else "" - - return html.Div( - [ - dbc.Checkbox( - id={"type": "channel-check", "index": key}, - label=html.Span( - [ - html.Span(label, style={"fontSize": "12px"}), - html.Span( - preview, - style={"fontSize": "10px", "color": "#999", "marginLeft": "4px"}, - ), - ] - ), - value=False, - style={"marginBottom": "1px"}, - ), - ], - className="channel-item", - ) - - -def _make_group_header( - group_label: str, - group_id: str, - channel_keys: list[str], -) -> html.Div: - """Build a measurement group header with select-all button.""" - return html.Div( - [ - html.Span(group_label, style={"fontSize": "12px", "fontWeight": "500"}), - dbc.Button( - "All", - id={"type": "select-group-btn", "index": group_id}, - color="link", - size="sm", - style={"fontSize": "10px", "padding": "0 4px", "textDecoration": "none"}, - ), - # Hidden store for this group's channel keys - dcc.Store( - id={"type": "group-channels", "index": group_id}, - data=channel_keys, - ), - ], - className="d-flex align-items-center justify-content-between", - ) - - -def build_channel_tree(app_state: AppState) -> dbc.Card: - """Build the full channel tree panel with search and accordion.""" - if app_state.is_empty: - return dbc.Card( - [ - dbc.CardHeader("Channels", className="fw-bold py-2"), - dbc.CardBody( - html.Div("No test loaded", className="text-muted small text-center py-3"), - ), - ] - ) - - multi_test = len(app_state.tests) > 1 - full_tree = app_state.build_channel_tree() - - # Build accordion items - accordion_items = [] - group_counter = 0 - - for test_id, test_tree in full_tree.items(): - test_label = "" - loaded = app_state.get_test(test_id) - if loaded and multi_test: - test_label = f"[{test_id}] " - - for obj_label, locations in test_tree.items(): - # Top-level accordion: test object (Driver, Vehicle Structure, etc.) - obj_content = [] - - for loc_label, measurements in locations.items(): - for meas_label, channels in measurements.items(): - group_id = f"grp_{group_counter}" - group_counter += 1 - - channel_keys = [ch["key"] for ch in channels] - header_label = ( - f"{loc_label} — {meas_label}" if loc_label != meas_label else meas_label - ) - - obj_content.append(_make_group_header(header_label, group_id, channel_keys)) - for ch in channels: - obj_content.append(_make_channel_item(ch, show_test_prefix=multi_test)) - - obj_content.append(html.Hr(style={"margin": "4px 0"})) - - if obj_content: - # Remove trailing hr - if obj_content and isinstance(obj_content[-1], html.Hr): - obj_content = obj_content[:-1] - - display_label = f"{test_label}{obj_label}" if test_label else obj_label - accordion_items.append( - dbc.AccordionItem( - html.Div(obj_content), - title=display_label, - style={"fontSize": "12px"}, - ) - ) - - return dbc.Card( - [ - dbc.CardHeader( - [ - html.Span("Channels", className="fw-bold"), - html.Span( - f" ({app_state.total_channels})", - style={"fontSize": "11px", "color": "#999"}, - ), - ], - className="py-2", - ), - dbc.CardBody( - [ - # Search - dbc.Input( - id="channel-search", - placeholder="Search channels...", - type="text", - size="sm", - className="mb-2", - debounce=True, - ), - # Selected channels display - html.Div(id="selected-channels-badges", className="mb-2"), - # Tree - html.Div( - dbc.Accordion( - accordion_items, - flush=True, - always_open=True, - start_collapsed=True, - id="channel-accordion", - ), - style={"maxHeight": "450px", "overflowY": "auto"}, - id="channel-tree-container", - ), - ], - className="py-2", - ), - ] - ) - - -def build_selected_channels_badges(selected_keys: list[str], app_state: AppState) -> list: - """Build badge pills showing currently selected channels.""" - if not selected_keys: - return [ - html.Span("No channels selected", className="text-muted", style={"fontSize": "11px"}) - ] - - badges = [] - for key in selected_keys[:20]: # Limit display to 20 - # Extract label - if "::" in key: - test_id, ch_name = key.split("::", 1) - ch = app_state.get_channel(test_id, ch_name) - if ch and ch.code.is_valid: - label = ch.code.short_label - else: - label = ch_name - if len(app_state.tests) > 1: - label = f"{test_id}: {label}" - else: - label = key - - badges.append( - dbc.Badge( - label, - id={"type": "channel-badge", "index": key}, - color="primary", - pill=True, - className="me-1 mb-1", - style={"fontSize": "10px", "cursor": "pointer"}, - ) - ) - - if len(selected_keys) > 20: - badges.append( - html.Span( - f"+{len(selected_keys) - 20} more", - className="text-muted", - style={"fontSize": "10px"}, - ) - ) - - return badges diff --git a/src/impakt/web/components/header.py b/src/impakt/web/components/header.py index 479e978..e7a36b8 100644 --- a/src/impakt/web/components/header.py +++ b/src/impakt/web/components/header.py @@ -129,7 +129,7 @@ def build_test_info_panel(app_state: AppState) -> dbc.Card: f"{meta.impact.speed_kmh:.1f}" if meta.impact.speed_kmh else "—", style={"fontSize": "12px"}, ), - html.Td(str(loaded.channel_count), style={"fontSize": "12px"}), + html.Td(str(len(loaded)), style={"fontSize": "12px"}), html.Td( dbc.Button( "x", diff --git a/src/impakt/web/state.py b/src/impakt/web/state.py index cf80de2..c7b92a1 100644 --- a/src/impakt/web/state.py +++ b/src/impakt/web/state.py @@ -1,14 +1,12 @@ """Application state management. -AppState is the central data store for the web UI. It holds all loaded tests, -manages channel transforms, and provides the data that callbacks need to -render plots, compute criteria, and generate reports. +AppState is the central data store for the web UI. It holds Session +objects (from the scripting API) and delegates to them for channel +access, transforms, criteria, and protocol scoring. -AppState is NOT a Dash Store — it lives server-side in Python memory. The Dash -stores hold lightweight references (test IDs, selected channels, plot config) +AppState is NOT a Dash Store — it lives server-side in Python memory. +Dash stores hold lightweight references (test IDs, selected channels) that callbacks use to look up data from AppState. - -This design keeps large numpy arrays out of the browser and avoids serialization. """ from __future__ import annotations @@ -17,220 +15,95 @@ import logging from pathlib import Path from typing import Any -from impakt.channel.model import Channel, ChannelGroup, TestData -from impakt.io.reader import get_registry +from impakt.channel.model import Channel, TestData +from impakt.script.api import Session from impakt.template.library import TemplateLibrary -from impakt.template.model import PlotDefinition, SessionState, TemplateSpec -from impakt.template.session import SessionManager +from impakt.template.model import PlotDefinition, TemplateSpec from impakt.transform.align import YAlign from impakt.transform.cfc import CFCFilter logger = logging.getLogger(__name__) -class LoadedTest: - """A single loaded test with its metadata and channel access.""" - - def __init__(self, test_data: TestData, color_offset: int = 0) -> None: - self.data = test_data - self.color_offset = color_offset - - @property - def test_id(self) -> str: - return self.data.test_id - - @property - def label(self) -> str: - """Short label for UI display.""" - meta = self.data.metadata - parts = [meta.test_number] - if meta.vehicle.make: - parts.append(f"{meta.vehicle.make} {meta.vehicle.model}".strip()) - return " — ".join(parts) if len(parts) > 1 else parts[0] - - @property - def channel_count(self) -> int: - return len(self.data) - - def __len__(self) -> int: - return len(self.data) - - class AppState: """Server-side application state. - Manages multiple loaded tests and provides channel resolution - for the UI callbacks. + Manages multiple loaded test sessions and provides channel resolution + for the UI callbacks. All test data access goes through Session objects. """ def __init__(self) -> None: - self._tests: dict[str, LoadedTest] = {} - self._test_order: list[str] = [] - self._color_counter: int = 0 + self._sessions: dict[str, Session] = {} + self._session_order: list[str] = [] self._template_library = TemplateLibrary() self._active_template: TemplateSpec | None = None - self._session_managers: dict[str, SessionManager] = {} - # Per-channel transform overrides: {channel_key: {"cfc": "600", "y_align": True}} + # Per-channel transform overrides: {channel_key: {"cfc": "600"}} self.channel_overrides: dict[str, dict[str, Any]] = {} - # Active corridors: list of {name, time, lower, upper, visible} + # Active corridors self.corridors: list[dict[str, Any]] = [] - def load_test(self, path: str | Path) -> LoadedTest: + def load_test(self, path: str | Path) -> Session: """Load a test from a path and add it to the state. - If a test with the same ID is already loaded, replaces it. - - Returns: - The loaded test. - - Raises: - ValueError: If the path is not a valid test directory. + Returns the Session object. """ - path = Path(path).resolve() - registry = get_registry() - test_data = registry.read(path) + session = Session.open(path) + test_id = session.test_id - loaded = LoadedTest(test_data, color_offset=self._color_counter) - self._color_counter += len(test_data) + self._sessions[test_id] = session + if test_id not in self._session_order: + self._session_order.append(test_id) - test_id = test_data.test_id - if test_id in self._tests: - self._tests[test_id] = loaded - else: - self._tests[test_id] = loaded - self._test_order.append(test_id) + logger.info("Loaded test '%s' from %s (%d channels)", test_id, path, len(session)) + return session - logger.info("Loaded test '%s' from %s (%d channels)", test_id, path, len(test_data)) - return loaded + def add_session(self, session: Session) -> None: + """Add an already-created Session to the state.""" + test_id = session.test_id + self._sessions[test_id] = session + if test_id not in self._session_order: + self._session_order.append(test_id) def remove_test(self, test_id: str) -> None: """Remove a loaded test.""" - if test_id in self._tests: - del self._tests[test_id] - self._test_order.remove(test_id) + self._sessions.pop(test_id, None) + if test_id in self._session_order: + self._session_order.remove(test_id) @property - def tests(self) -> list[LoadedTest]: - """All loaded tests in load order.""" - return [self._tests[tid] for tid in self._test_order if tid in self._tests] + def sessions(self) -> list[Session]: + """All loaded sessions in load order.""" + return [self._sessions[tid] for tid in self._session_order if tid in self._sessions] + + @property + def tests(self) -> list[Session]: + """Alias for sessions (backward compat for components).""" + return self.sessions @property def test_ids(self) -> list[str]: - return list(self._test_order) + return list(self._session_order) @property - def primary_test(self) -> LoadedTest | None: - """The first loaded test (primary for criteria, etc.).""" - if self._test_order: - return self._tests.get(self._test_order[0]) + def primary_test(self) -> Session | None: + """The first loaded session (primary for criteria, etc.).""" + if self._session_order: + return self._sessions.get(self._session_order[0]) return None - def get_test(self, test_id: str) -> LoadedTest | None: - return self._tests.get(test_id) + def get_test(self, test_id: str) -> Session | None: + return self._sessions.get(test_id) def get_channel(self, test_id: str, channel_name: str) -> Channel | None: """Resolve a channel from a test ID and channel name.""" - test = self._tests.get(test_id) - if test is None: + session = self._sessions.get(test_id) + if session is None: return None try: - return test.data.get(channel_name) + return session.data.get(channel_name) except KeyError: return None - def resolve_channel( - self, - channel_key: str, - cfc_class: int | None = None, - y_align: bool = False, - ) -> Channel | None: - """Resolve a channel key (test_id::channel_name) and apply transforms. - - Channel keys use the format 'test_id::channel_name'. If no '::' - separator, assumes the primary test. - """ - if "::" in channel_key: - test_id, ch_name = channel_key.split("::", 1) - else: - if self.primary_test is None: - return None - test_id = self.primary_test.test_id - ch_name = channel_key - - ch = self.get_channel(test_id, ch_name) - if ch is None: - return None - - # Apply transforms - if cfc_class is not None: - try: - ch = CFCFilter(cfc_class=cfc_class).apply(ch) - except (ValueError, Exception): - pass - - if y_align: - ch = YAlign().apply(ch) - - return ch - - def build_channel_tree( - self, - ) -> dict[str, dict[str, dict[str, dict[str, list[dict[str, str]]]]]]: - """Build a hierarchical channel tree across all loaded tests. - - Returns: - {test_id: {object_label: {location_label: {measurement_label: [{name, label, key}]}}}} - """ - result: dict[str, dict[str, dict[str, dict[str, list[dict[str, str]]]]]] = {} - - for loaded in self.tests: - tree = loaded.data.channel_tree() - test_tree: dict[str, dict[str, dict[str, list[dict[str, str]]]]] = {} - - for obj, locations in sorted(tree.items()): - obj_tree: dict[str, dict[str, list[dict[str, str]]]] = {} - for loc, measurements in sorted(locations.items()): - loc_tree: dict[str, list[dict[str, str]]] = {} - for meas, channels in sorted(measurements.items()): - ch_list = [] - for ch in channels: - label = ch.code.short_label if ch.code.is_valid else ch.name - ch_list.append( - { - "name": ch.name, - "label": label, - "key": f"{loaded.test_id}::{ch.name}", - "unit": ch.unit, - "peak": f"{ch.peak:.2f}", - "samples": str(ch.n_samples), - "rate": f"{ch.sample_rate:.0f}", - } - ) - loc_tree[meas] = ch_list - obj_tree[loc] = loc_tree - test_tree[obj] = obj_tree - result[loaded.test_id] = test_tree - - return result - - def flat_channel_list(self) -> list[dict[str, str]]: - """Flat list of all channels across all tests, for dropdown/checklist options.""" - items: list[dict[str, str]] = [] - multi = len(self._tests) > 1 - - for loaded in self.tests: - prefix = f"[{loaded.test_id}] " if multi else "" - for ch in sorted(loaded.data, key=lambda c: c.name): - label = ch.code.short_label if ch.code.is_valid else ch.name - items.append( - { - "label": f"{prefix}{label}", - "value": f"{loaded.test_id}::{ch.name}", - } - ) - - return items - # ----- Template & Session ----- @property @@ -245,52 +118,37 @@ class AppState: def template_names(self) -> list[str]: return self._template_library.list() - def apply_template( - self, - name: str, - selected_keys: list[str] | None = None, - ) -> tuple[list[str], dict[str, str]]: + def apply_template(self, name: str) -> tuple[list[str], dict[str, str]]: """Apply a template by name. - Resolves the template's channel patterns against the primary test, - sets the active template, and returns the resolved channel keys and - transform settings. - - Returns: - (selected_channel_keys, transform_settings) + Resolves channel patterns against the primary test. + Returns (selected_channel_keys, transform_settings). """ template = self._template_library.get(name) self._active_template = template - # Persist to session if primary test has a path primary = self.primary_test - if primary and primary.data.path: - mgr = self._get_session_manager(primary.data.path) - mgr.apply_template(template) + if primary: + primary.apply_template(template) - # Resolve channel patterns from the template + # Resolve channel patterns resolved_keys: list[str] = [] if primary: for plot_def in template.plots: for pattern in plot_def.channel_patterns: - matches = primary.data.find(pattern) + matches = primary.find(pattern) for ch in matches: key = f"{primary.test_id}::{ch.name}" if key not in resolved_keys: resolved_keys.append(key) - # Transform settings transforms: dict[str, str] = {} if template.default_cfc is not None: transforms["cfc"] = str(template.default_cfc) else: transforms["cfc"] = "none" - logger.info( - "Applied template '%s' — resolved %d channels", - name, - len(resolved_keys), - ) + logger.info("Applied template '%s' — %d channels", name, len(resolved_keys)) return resolved_keys, transforms def save_as_template( @@ -303,30 +161,19 @@ class AppState: x2: float | None, protocol: str = "", ) -> TemplateSpec: - """Capture the current UI state as a new template. - - Converts selected channels back to patterns and stores the - current filter/cursor/protocol settings. - """ - # Convert selected keys to channel patterns + """Capture current UI state as a new template.""" patterns: list[str] = [] for key in selected_keys: - if "::" in key: - _, ch_name = key.split("::", 1) - else: - ch_name = key - # Use the raw channel name as a pattern (exact match) + ch_name = key.split("::", 1)[-1] if "::" in key else key if ch_name not in patterns: patterns.append(ch_name) - # Build plot definition plot = PlotDefinition( title=name, channel_patterns=patterns, x_cursors=(x1, x2) if x1 is not None and x2 is not None else None, ) - # Build transforms list transforms: list[dict[str, Any]] = [] if cfc_value and cfc_value != "none": transforms.append({"type": "cfc_filter", "cfc_class": int(cfc_value)}) @@ -343,17 +190,18 @@ class AppState: self._template_library.save(template) self._active_template = template - - logger.info("Saved template '%s' with %d channel patterns", name, len(patterns)) + logger.info("Saved template '%s' with %d patterns", name, len(patterns)) return template def save_session(self, selected_keys: list[str], cfc_value: str, **overrides: Any) -> None: """Auto-save current state to the session for the primary test.""" primary = self.primary_test - if not primary or not primary.data.path: + if not primary or not primary.path: return - mgr = self._get_session_manager(primary.data.path) + from impakt.template.session import SessionManager + + mgr = SessionManager(primary.path) mgr.state.overrides = { "selected_channels": selected_keys, "cfc": cfc_value, @@ -362,12 +210,14 @@ class AppState: mgr.save() def load_session_state(self) -> dict[str, Any] | None: - """Load saved session state for the primary test, if any.""" + """Load saved session state for the primary test.""" primary = self.primary_test - if not primary or not primary.data.path: + if not primary or not primary.path: return None - mgr = self._get_session_manager(primary.data.path) + from impakt.template.session import SessionManager + + mgr = SessionManager(primary.path) if not mgr.has_session: return None @@ -377,21 +227,48 @@ class AppState: "template": mgr.state.template_name, } - def _get_session_manager(self, path: Path) -> SessionManager: - key = str(path) - if key not in self._session_managers: - self._session_managers[key] = SessionManager(path) - return self._session_managers[key] + # ----- Channel tree / grid helpers ----- + + def build_channel_tree(self) -> dict[str, Any]: + """Build hierarchical channel tree across all loaded tests.""" + result: dict[str, Any] = {} + for session in self.sessions: + tree = session.channel_tree() + test_tree: dict[str, Any] = {} + for obj, locations in sorted(tree.items()): + obj_tree: dict[str, Any] = {} + for loc, measurements in sorted(locations.items()): + loc_tree: dict[str, Any] = {} + for meas, channels in sorted(measurements.items()): + ch_list = [] + for ch in channels: + label = ch.code.short_label if ch.code.is_valid else ch.name + ch_list.append( + { + "name": ch.name, + "label": label, + "key": f"{session.test_id}::{ch.name}", + "unit": ch.unit, + "peak": f"{ch.peak:.2f}", + "samples": str(ch.n_samples), + "rate": f"{ch.sample_rate:.0f}", + } + ) + loc_tree[meas] = ch_list + obj_tree[loc] = loc_tree + test_tree[obj] = obj_tree + result[session.test_id] = test_tree + return result @property def is_empty(self) -> bool: - return len(self._tests) == 0 + return len(self._sessions) == 0 @property def total_channels(self) -> int: - return sum(t.channel_count for t in self.tests) + return sum(len(s) for s in self.sessions) def __repr__(self) -> str: - test_info = ", ".join(f"{t.test_id}({t.channel_count}ch)" for t in self.tests) + test_info = ", ".join(f"{s.test_id}({len(s)}ch)" for s in self.sessions) tmpl = f", template={self._active_template.name}" if self._active_template else "" return f"AppState([{test_info}]{tmpl})" diff --git a/tests/test_criteria/test_chest_femur_tibia.py b/tests/test_criteria/test_chest_femur_tibia.py new file mode 100644 index 0000000..9450997 --- /dev/null +++ b/tests/test_criteria/test_chest_femur_tibia.py @@ -0,0 +1,93 @@ +"""Tests for chest deflection, femur load, tibia index, 3ms clip, viscous criterion.""" + +import numpy as np +import pytest + +from impakt.criteria import chest_deflection, clip_3ms, femur_load, tibia_index, viscous_criterion + + +class TestChestDeflection: + def test_basic(self, chest_deflection_channel): + result = chest_deflection(channel=chest_deflection_channel) + assert result.criterion == "Chest Deflection" + assert result.unit == "mm" + assert 30.0 < result.value < 40.0 # ~35mm in fixture + + def test_peak_time(self, chest_deflection_channel): + result = chest_deflection(channel=chest_deflection_channel) + assert result.time_of_peak is not None + assert 0.0 < result.time_of_peak < 0.1 + + +class TestClip3ms: + def test_basic(self, head_accel_x): + result = clip_3ms(head_accel_x) + assert result.criterion == "3ms Clip" + assert result.value > 0 + + def test_body_region(self, head_accel_x): + result = clip_3ms(head_accel_x) + assert result.body_region == "Chest" + + +class TestFemurLoad: + def test_single_channel(self, femur_left_channel): + result = femur_load(channel=femur_left_channel, side="left") + assert result.criterion == "Femur Load Left" + assert result.unit == "kN" + assert result.value > 0 + + def test_peak_time(self, femur_left_channel): + result = femur_load(channel=femur_left_channel, side="left") + assert result.time_of_peak is not None + + +class TestTibiaIndex: + def test_with_all_components(self, time_array, sample_rate): + from impakt.channel.code import ChannelCode + from impakt.channel.model import Channel + + t = time_array + fz = np.zeros_like(t) + mx = np.zeros_like(t) + my = np.zeros_like(t) + mask = (t >= 0.02) & (t <= 0.08) + fz[mask] = -5000 * np.sin(np.pi * (t[mask] - 0.02) / 0.06) + mx[mask] = 50 * np.sin(np.pi * (t[mask] - 0.02) / 0.06) + my[mask] = 80 * np.sin(np.pi * (t[mask] - 0.02) / 0.06) + + fz_ch = Channel( + name="TIBFZ", + code=ChannelCode.parse("TIBFZ"), + data=fz, + time=t, + unit="N", + sample_rate=sample_rate, + ) + mx_ch = Channel( + name="TIBMX", + code=ChannelCode.parse("TIBMX"), + data=mx, + time=t, + unit="N·m", + sample_rate=sample_rate, + ) + my_ch = Channel( + name="TIBMY", + code=ChannelCode.parse("TIBMY"), + data=my, + time=t, + unit="N·m", + sample_rate=sample_rate, + ) + + result = tibia_index(fz_channel=fz_ch, mx_channel=mx_ch, my_channel=my_ch) + assert result.value > 0 + + +class TestViscousCriterion: + def test_basic(self, chest_deflection_channel): + result = viscous_criterion(channel=chest_deflection_channel) + assert result.criterion == "Viscous Criterion" + assert result.unit == "m/s" + assert result.value >= 0 diff --git a/tests/test_plot/test_engine.py b/tests/test_plot/test_engine.py new file mode 100644 index 0000000..5f37100 --- /dev/null +++ b/tests/test_plot/test_engine.py @@ -0,0 +1,81 @@ +"""Tests for plot engine.""" + +import numpy as np +import pytest + +from impakt.plot.engine import PlotEngine, DEFAULT_COLORS +from impakt.plot.spec import ChannelRef, Corridor, CorridorStyle, PlotSpec, PlotStyle + + +class TestPlotEngine: + def test_render_empty(self): + engine = PlotEngine() + spec = PlotSpec() + fig = engine.render(spec) + assert fig is not None + + def test_render_with_channels(self, head_accel_x, head_accel_y): + engine = PlotEngine() + spec = PlotSpec( + channels=[ + ChannelRef(channel=head_accel_x, style=PlotStyle(label="Head X")), + ChannelRef(channel=head_accel_y, style=PlotStyle(label="Head Y")), + ], + y_label="g", + ) + fig = engine.render(spec) + assert len(fig.data) == 2 + + def test_render_compact_mode(self, head_accel_x): + engine = PlotEngine() + spec = PlotSpec( + channels=[ChannelRef(channel=head_accel_x)], + compact=True, + ) + fig = engine.render(spec) + assert fig.layout.hovermode is False + assert fig.layout.showlegend is False + + def test_render_with_corridors(self, head_accel_x): + engine = PlotEngine() + corridor = Corridor( + name="Test Corridor", + time=np.linspace(0, 0.1, 100), + lower=np.full(100, -50.0), + upper=np.full(100, 50.0), + ) + spec = PlotSpec( + channels=[ChannelRef(channel=head_accel_x)], + corridors=[corridor], + ) + fig = engine.render(spec) + # 1 data trace + 2 corridor traces (upper + lower) + assert len(fig.data) == 3 + + def test_render_with_cursors(self, head_accel_x): + engine = PlotEngine() + spec = PlotSpec( + channels=[ChannelRef(channel=head_accel_x)], + x_cursors=(0.02, 0.06), + ) + fig = engine.render(spec) + # Should have vertical lines (as shapes) + assert len(fig.layout.shapes) >= 2 + + def test_compact_cursor_annotations(self, head_accel_x): + engine = PlotEngine() + spec = PlotSpec( + channels=[ChannelRef(channel=head_accel_x)], + x_cursors=(0.02, 0.06), + compact=True, + ) + fig = engine.render(spec) + # Should have X1/X2 annotations + annotations = [ + a for a in (fig.layout.annotations or []) if "X1" in str(a.text) or "X2" in str(a.text) + ] + assert len(annotations) == 2 + + def test_default_colors(self): + assert len(DEFAULT_COLORS) == 10 + assert all(c.startswith("#") for c in DEFAULT_COLORS) diff --git a/tests/test_protocol/test_iihs_usncap.py b/tests/test_protocol/test_iihs_usncap.py new file mode 100644 index 0000000..4bb9904 --- /dev/null +++ b/tests/test_protocol/test_iihs_usncap.py @@ -0,0 +1,68 @@ +"""Tests for IIHS and US NCAP scoring.""" + +import pytest + +from impakt.criteria.base import CriterionResult +from impakt.protocol.iihs import IIHS, evaluate as iihs_evaluate +from impakt.protocol.us_ncap import USNCAP, evaluate as us_evaluate + + +@pytest.fixture +def sample_criteria(): + return { + "HIC15": CriterionResult(criterion="HIC15", value=400, body_region="Head"), + "Nij": CriterionResult(criterion="Nij", value=0.4, body_region="Neck"), + "Chest Deflection": CriterionResult( + criterion="Chest Deflection", value=35, unit="mm", body_region="Chest" + ), + "Femur Load Left": CriterionResult( + criterion="Femur Load Left", value=4.0, unit="kN", body_region="Femur Left" + ), + "Femur Load Right": CriterionResult( + criterion="Femur Load Right", value=4.5, unit="kN", body_region="Femur Right" + ), + } + + +class TestIIHS: + def test_good_results(self, sample_criteria): + result = iihs_evaluate(sample_criteria) + assert result.protocol == "IIHS" + assert result.overall_rating in ("GOOD", "ACCEPTABLE", "MARGINAL", "POOR") + + def test_region_scores(self, sample_criteria): + result = iihs_evaluate(sample_criteria) + assert len(result.region_scores) > 0 + for rs in result.region_scores: + assert rs.rating is not None + + def test_poor_hic(self): + result = iihs_evaluate( + { + "HIC15": CriterionResult(criterion="HIC15", value=1500, body_region="Head"), + } + ) + assert result.overall_rating == "POOR" + + def test_summary(self, sample_criteria): + result = iihs_evaluate(sample_criteria) + summary = result.summary() + assert "IIHS" in summary + + +class TestUSNCAP: + def test_basic(self, sample_criteria): + result = us_evaluate(sample_criteria) + assert result.protocol == "US NCAP" + assert result.stars is not None + assert 1 <= result.stars <= 5 + + def test_injury_probabilities(self, sample_criteria): + result = us_evaluate(sample_criteria) + assert "combined_injury_probability" in result.details + p = result.details["combined_injury_probability"] + assert 0.0 <= p <= 1.0 + + def test_region_scores(self, sample_criteria): + result = us_evaluate(sample_criteria) + assert len(result.region_scores) > 0 diff --git a/tests/test_scripting_api.py b/tests/test_scripting_api.py new file mode 100644 index 0000000..573ed57 --- /dev/null +++ b/tests/test_scripting_api.py @@ -0,0 +1,126 @@ +"""Tests for the scripting API (Session, ChannelHandle, TransformProxy).""" + +from pathlib import Path + +import pytest + +from impakt import Session, Template + +FIXTURE_DATA = Path(__file__).parent / "fixtures" / "sample_mme" +MME_DATA = Path(__file__).parent / "mme_data" + + +class TestSession: + def test_open(self): + s = Session.open(FIXTURE_DATA) + assert s.test_id == "IMPAKT_SYNTH_001" + assert len(s) == 26 + + def test_channel_access(self): + s = Session.open(FIXTURE_DATA) + ch = s.channel("11HEAD0000ACXA") + assert ch.name == "11HEAD0000ACXA" + assert ch.peak > 0 + + def test_find(self): + s = Session.open(FIXTURE_DATA) + channels = s.find("*HEAD*AC*") + assert len(channels) == 3 + + def test_group(self): + s = Session.open(FIXTURE_DATA) + group = s.group("HEAD0000AC") + assert group.x is not None + + def test_compute_criteria(self): + s = Session.open(FIXTURE_DATA) + criteria = s.compute_criteria() + assert len(criteria) > 0 + + def test_evaluate(self): + s = Session.open(FIXTURE_DATA) + result = s.evaluate("euro_ncap") + assert result.stars is not None + assert result.protocol == "Euro NCAP" + + def test_evaluate_us_ncap(self): + s = Session.open(FIXTURE_DATA) + result = s.evaluate("us_ncap") + assert result.stars is not None + + def test_evaluate_iihs(self): + s = Session.open(FIXTURE_DATA) + result = s.evaluate("iihs") + assert result.overall_rating in ("GOOD", "ACCEPTABLE", "MARGINAL", "POOR") + + def test_evaluate_invalid_protocol(self): + s = Session.open(FIXTURE_DATA) + with pytest.raises(ValueError, match="Unknown protocol"): + s.evaluate("invalid") + + def test_contains(self): + s = Session.open(FIXTURE_DATA) + assert "11HEAD0000ACXA" in s + assert "NONEXISTENT" not in s + + +class TestChannelHandleChaining: + """The fluent API must support chaining — each transform returns ChannelHandle.""" + + def test_single_transform(self): + s = Session.open(FIXTURE_DATA) + ch = s.channel("11HEAD0000ACXA") + filtered = ch.transform.cfc(600) + assert type(filtered).__name__ == "ChannelHandle" + assert filtered.raw.cfc_class == 600 + + def test_double_chain(self): + s = Session.open(FIXTURE_DATA) + result = s.channel("11HEAD0000ACXA").transform.cfc(600).transform.y_align() + assert type(result).__name__ == "ChannelHandle" + assert len(result.raw.transform_history) == 2 + + def test_triple_chain(self): + s = Session.open(FIXTURE_DATA) + result = ( + s.channel("11HEAD0000ACXA") + .transform.cfc(1000) + .transform.y_align() + .transform.trim(t_start=0.0, t_end=0.1) + ) + assert type(result).__name__ == "ChannelHandle" + assert len(result.raw.transform_history) == 3 + + def test_chain_preserves_data(self): + s = Session.open(FIXTURE_DATA) + original = s.channel("11HEAD0000ACXA") + original_peak = original.peak + filtered = original.transform.cfc(600) + # Original should be unchanged — peak should be the same + assert original.peak == original_peak + # Filtered should have different CFC and lower peak (smoothed) + assert filtered.raw.cfc_class == 600 + assert filtered.peak <= original_peak + + +@pytest.mark.skipif(not (MME_DATA / "3239").exists(), reason="Real data not available") +class TestSessionRealData: + def test_open_real(self): + s = Session.open(MME_DATA / "3239") + assert s.test_id == "3239" + assert len(s) == 133 + + def test_full_pipeline(self): + s = Session.open(MME_DATA / "3239") + # Chain: get channel -> filter -> check + ch = s.channel("11HEAD0000H3ACXP").transform.cfc(1000) + assert ch.peak > 100 # Significant head acceleration + + # Compute criteria + criteria = s.compute_criteria() + assert "HIC15" in criteria + + # Evaluate + result = s.evaluate("euro_ncap") + assert result.stars is not None + assert result.stars >= 0 diff --git a/tests/test_template.py b/tests/test_template.py new file mode 100644 index 0000000..756ba58 --- /dev/null +++ b/tests/test_template.py @@ -0,0 +1,132 @@ +"""Tests for template model and session persistence.""" + +from pathlib import Path + +import pytest + +from impakt.template.library import TemplateLibrary +from impakt.template.model import PlotDefinition, SessionState, TemplateSpec +from impakt.template.session import SessionManager + + +class TestTemplateSpec: + def test_yaml_round_trip(self): + template = TemplateSpec( + name="Test Template", + version=2, + description="A test template", + plots=[ + PlotDefinition( + title="Head Acceleration", + channel_patterns=["*HEAD*AC*"], + x_cursors=(0.0, 0.1), + ) + ], + default_cfc=1000, + criteria=["hic15", "nij"], + protocol="euro_ncap", + ) + + yaml_str = template.to_yaml() + restored = TemplateSpec.from_yaml(yaml_str) + + assert restored.name == "Test Template" + assert restored.version == 2 + assert restored.default_cfc == 1000 + assert len(restored.plots) == 1 + assert restored.plots[0].channel_patterns == ["*HEAD*AC*"] + assert restored.criteria == ["hic15", "nij"] + + def test_save_and_load(self, tmp_path): + template = TemplateSpec(name="Saved Test", version=1) + path = tmp_path / "test.yaml" + template.save(path) + assert path.exists() + + loaded = TemplateSpec.load(path) + assert loaded.name == "Saved Test" + + +class TestSessionState: + def test_yaml_round_trip(self): + state = SessionState( + template_name="my_template", + template_version=3, + test_path="/data/test_001", + overrides={"cfc": "600", "selected": ["ch1", "ch2"]}, + ) + yaml_str = state.to_yaml() + restored = SessionState.from_yaml(yaml_str) + + assert restored.template_name == "my_template" + assert restored.template_version == 3 + assert restored.overrides["cfc"] == "600" + + def test_save_and_load(self, tmp_path): + state = SessionState(template_name="test") + path = tmp_path / "session.yaml" + state.save(path) + assert path.exists() + + loaded = SessionState.load(path) + assert loaded.template_name == "test" + + +class TestTemplateLibrary: + def test_empty_library(self, tmp_path): + lib = TemplateLibrary(tmp_path / "templates") + assert lib.list() == [] + assert len(lib) == 0 + + def test_save_and_list(self, tmp_path): + lib = TemplateLibrary(tmp_path / "templates") + template = TemplateSpec(name="My Template") + lib.save(template) + assert "my_template" in lib.list() + assert len(lib) == 1 + + def test_get(self, tmp_path): + lib = TemplateLibrary(tmp_path / "templates") + lib.save(TemplateSpec(name="Getter Test", version=5)) + loaded = lib.get("getter_test") + assert loaded.name == "Getter Test" + assert loaded.version == 5 + + def test_delete(self, tmp_path): + lib = TemplateLibrary(tmp_path / "templates") + lib.save(TemplateSpec(name="To Delete")) + assert lib.delete("to_delete") + assert "to_delete" not in lib.list() + + def test_get_missing_raises(self, tmp_path): + lib = TemplateLibrary(tmp_path / "templates") + with pytest.raises(FileNotFoundError): + lib.get("nonexistent") + + +class TestSessionManager: + def test_create_and_save(self, tmp_path): + mgr = SessionManager(tmp_path) + mgr.state.template_name = "test_tmpl" + mgr.save() + assert mgr.has_session + assert (tmp_path / ".impakt" / "session.yaml").exists() + + def test_load_existing(self, tmp_path): + # Save + mgr1 = SessionManager(tmp_path) + mgr1.state.template_name = "saved_tmpl" + mgr1.state.overrides = {"key": "value"} + mgr1.save() + + # Load + mgr2 = SessionManager(tmp_path) + assert mgr2.state.template_name == "saved_tmpl" + assert mgr2.state.overrides["key"] == "value" + + def test_clear(self, tmp_path): + mgr = SessionManager(tmp_path) + mgr.save() + assert mgr.has_session + mgr.clear() + assert not mgr.has_session diff --git a/tests/test_transform/test_math_resultant.py b/tests/test_transform/test_math_resultant.py new file mode 100644 index 0000000..c395b7f --- /dev/null +++ b/tests/test_transform/test_math_resultant.py @@ -0,0 +1,75 @@ +"""Tests for math expressions and resultant computation.""" + +import numpy as np +import pytest + +from impakt.transform.math_expr import math_expr +from impakt.transform.resultant import resultant_from_channels +from impakt.transform.resample import trim, resample + + +class TestMathExpr: + def test_simple_expression(self, head_accel_x, head_accel_z): + result = math_expr( + expression="sqrt(a**2 + b**2)", + channels={"a": head_accel_x, "b": head_accel_z}, + name="resultant_xz", + unit="g", + ) + assert result.name == "resultant_xz" + assert result.unit == "g" + assert result.peak > 0 + assert len(result.data) == len(head_accel_x.data) + + def test_constant_expression(self, head_accel_x): + result = math_expr( + expression="a * 0 + 42.0", + channels={"a": head_accel_x}, + name="constant", + ) + assert np.allclose(result.data, 42.0) + + def test_invalid_expression(self, head_accel_x): + with pytest.raises(ValueError, match="Error evaluating"): + math_expr( + expression="invalid_func(a)", + channels={"a": head_accel_x}, + ) + + def test_forbidden_expression(self, head_accel_x): + with pytest.raises(ValueError, match="Forbidden"): + math_expr( + expression="__import__('os')", + channels={"a": head_accel_x}, + ) + + +class TestResultant: + def test_from_channels(self, head_accel_x, head_accel_y, head_accel_z): + result = resultant_from_channels(head_accel_x, head_accel_y, head_accel_z) + assert result.code.direction == "R" + # Resultant >= any component + assert result.peak >= head_accel_x.peak + assert result.peak >= head_accel_y.peak + + def test_from_two_channels(self, head_accel_x, head_accel_z): + result = resultant_from_channels(head_accel_x, head_accel_z) + assert result.peak > 0 + + def test_single_channel_raises(self): + with pytest.raises(ValueError, match="At least one"): + resultant_from_channels() + + +class TestTrimResample: + def test_trim(self, head_accel_x): + trimmed = trim(head_accel_x, t_start=0.0, t_end=0.05) + assert trimmed.time[0] >= 0.0 + assert trimmed.time[-1] <= 0.05 + assert len(trimmed.data) < len(head_accel_x.data) + + def test_resample(self, head_accel_x): + resampled = resample(head_accel_x, target_rate=5000.0) + expected_samples = int(head_accel_x.duration * 5000.0) + assert abs(len(resampled.data) - expected_samples) <= 2 + assert resampled.sample_rate == 5000.0 diff --git a/tests/test_web/test_state.py b/tests/test_web/test_state.py index 56d24c7..93c950b 100644 --- a/tests/test_web/test_state.py +++ b/tests/test_web/test_state.py @@ -19,11 +19,11 @@ class TestAppState: def test_load_test(self): state = AppState() - loaded = state.load_test(FIXTURE_DATA) + session = state.load_test(FIXTURE_DATA) assert not state.is_empty - assert loaded.test_id == "IMPAKT_SYNTH_001" - assert loaded.channel_count == 26 - assert state.primary_test is loaded + assert session.test_id == "IMPAKT_SYNTH_001" + assert len(session) == 26 + assert state.primary_test is session def test_load_multiple_tests(self): state = AppState() @@ -32,13 +32,13 @@ class TestAppState: if (MME_DATA / "VW1FGS15").exists(): t2 = state.load_test(MME_DATA / "VW1FGS15") assert len(state.tests) == 2 - assert state.primary_test is t1 # First loaded is primary + assert state.primary_test is t1 assert state.total_channels == 26 + 10 def test_remove_test(self): state = AppState() - loaded = state.load_test(FIXTURE_DATA) - state.remove_test(loaded.test_id) + session = state.load_test(FIXTURE_DATA) + state.remove_test(session.test_id) assert state.is_empty def test_get_channel(self): @@ -54,38 +54,38 @@ class TestAppState: ch = state.get_channel("IMPAKT_SYNTH_001", "NONEXISTENT") assert ch is None - def test_resolve_channel_with_key(self): + def test_get_channel_via_session(self): + """Channels can be accessed through the Session scripting API.""" state = AppState() state.load_test(FIXTURE_DATA) - ch = state.resolve_channel("IMPAKT_SYNTH_001::11HEAD0000ACXA") - assert ch is not None + session = state.primary_test + assert session is not None + ch_handle = session.channel("11HEAD0000ACXA") + assert ch_handle.name == "11HEAD0000ACXA" - def test_resolve_channel_primary_default(self): + def test_session_fluent_transforms(self): + """Fluent transform chaining works through the Session API.""" state = AppState() state.load_test(FIXTURE_DATA) - ch = state.resolve_channel("11HEAD0000ACXA") - assert ch is not None + session = state.primary_test + ch = session.channel("11HEAD0000ACXA") + filtered = ch.transform.cfc(600).transform.y_align() + assert filtered.raw.cfc_class == 600 + assert len(filtered.raw.transform_history) == 2 - def test_resolve_channel_with_cfc(self): + def test_session_compute_criteria(self): + """Session.compute_criteria() auto-detects channels.""" state = AppState() state.load_test(FIXTURE_DATA) - ch = state.resolve_channel("11HEAD0000ACXA", cfc_class=600) - assert ch is not None - assert ch.cfc_class == 600 - - def test_flat_channel_list(self): - state = AppState() - state.load_test(FIXTURE_DATA) - items = state.flat_channel_list() - assert len(items) == 26 - assert all("value" in item and "label" in item for item in items) + criteria = state.primary_test.compute_criteria() + assert len(criteria) > 0 + assert "HIC15" in criteria or "Chest Deflection" in criteria def test_build_channel_tree(self): state = AppState() state.load_test(FIXTURE_DATA) tree = state.build_channel_tree() assert "IMPAKT_SYNTH_001" in tree - # Should have hierarchical structure test_tree = tree["IMPAKT_SYNTH_001"] assert len(test_tree) > 0 @@ -94,15 +94,23 @@ class TestAppState: class TestAppStateRealData: def test_load_real_mme(self): state = AppState() - loaded = state.load_test(MME_DATA / "3239") - assert loaded.test_id == "3239" - assert loaded.channel_count == 133 + session = state.load_test(MME_DATA / "3239") + assert session.test_id == "3239" + assert len(session) == 133 def test_channel_tree_real_data(self): state = AppState() state.load_test(MME_DATA / "3239") tree = state.build_channel_tree() assert "3239" in tree - # Should contain "Driver" in some key test_tree = tree["3239"] assert any("Driver" in k for k in test_tree) + + def test_session_evaluate_real_data(self): + """Full pipeline through Session API on real data.""" + state = AppState() + state.load_test(MME_DATA / "3239") + result = state.primary_test.evaluate("euro_ncap") + assert result.stars is not None + assert result.stars >= 0 + assert len(result.region_scores) > 0