feat(profiler/behave_shell): emit motor.keystroke_cadence
BEHAVE-EXTRACTOR.md Phase B Step B.1.
* SessionContext gains typing_bursts: tuple[tuple[float, ...], ...]
built by _split_typing_bursts(iats) — splits at gaps > IKI_THINK_MAX_S
(1.5s) and drops bursts of fewer than 3 IATs. Mirrors prototype's
_split_into_bursts at BEHAVE/prototype_extractors/shell/extract.py:275.
* _features/motor.py:keystroke_cadence(ctx) emits one Observation
in {steady, bursty, hunt_and_peck, machine}. Median CV across
typing bursts; mean IKI < IKI_MACHINE_MAX_S paired with CV <
CV_MACHINE_MAX → machine. Confidence 0.85/0.70/0.65/0.60 per the
prototype's calibration history.
* < MIN_INPUTS_FOR_CADENCE inputs or zero typing bursts → skip
emission. v0.1 emits only the burst-CV variant; the prototype's
NAIVE session-CV variant is parked for v0.2.
* Calibration grid widened (PHASE_A_PRIMITIVES → PHASE_AB_PRIMITIVES)
to include motor.keystroke_cadence. Grid green across all five
shards.
Tests: too-few-inputs → no emit, all-think-pauses → no burst → no
emit, uniform IATs → steady, sub-5ms → machine, mixed-pace → bursty,
extreme bimodal → hunt_and_peck.
This commit is contained in:
@@ -20,6 +20,7 @@ from decnet.profiler.behave_shell._parse import (
|
|||||||
hash_token,
|
hash_token,
|
||||||
)
|
)
|
||||||
from decnet.profiler.behave_shell._thresholds import (
|
from decnet.profiler.behave_shell._thresholds import (
|
||||||
|
IKI_THINK_MAX_S,
|
||||||
PASTE_BURST_MAX_IAT_S,
|
PASTE_BURST_MAX_IAT_S,
|
||||||
PASTE_MIN_CHARS_PER_EVENT,
|
PASTE_MIN_CHARS_PER_EVENT,
|
||||||
)
|
)
|
||||||
@@ -47,6 +48,9 @@ class SessionContext:
|
|||||||
inter_cmd_iats: tuple[float, ...] = field(default_factory=tuple)
|
inter_cmd_iats: tuple[float, ...] = field(default_factory=tuple)
|
||||||
output_per_cmd: tuple[int, ...] = field(default_factory=tuple)
|
output_per_cmd: tuple[int, ...] = field(default_factory=tuple)
|
||||||
|
|
||||||
|
# Step B.1 derivations — typing bursts (IATs split at think-pauses)
|
||||||
|
typing_bursts: tuple[tuple[float, ...], ...] = field(default_factory=tuple)
|
||||||
|
|
||||||
|
|
||||||
def _detect_paste_bursts(
|
def _detect_paste_bursts(
|
||||||
inputs: list[AsciinemaEvent],
|
inputs: list[AsciinemaEvent],
|
||||||
@@ -102,6 +106,22 @@ def _detect_paste_bursts(
|
|||||||
return tuple(bursts), paste_count
|
return tuple(bursts), paste_count
|
||||||
|
|
||||||
|
|
||||||
|
def _split_typing_bursts(iats: tuple[float, ...]) -> tuple[tuple[float, ...], ...]:
|
||||||
|
"""Split a flat IAT sequence at gaps > IKI_THINK_MAX_S.
|
||||||
|
|
||||||
|
Drops bursts of fewer than 3 IATs — too short to compute a stable
|
||||||
|
CV. Mirrors BEHAVE prototype's ``_split_into_bursts``.
|
||||||
|
"""
|
||||||
|
bursts: list[list[float]] = [[]]
|
||||||
|
for x in iats:
|
||||||
|
if x > IKI_THINK_MAX_S:
|
||||||
|
if bursts[-1]:
|
||||||
|
bursts.append([])
|
||||||
|
else:
|
||||||
|
bursts[-1].append(x)
|
||||||
|
return tuple(tuple(b) for b in bursts if len(b) >= 3)
|
||||||
|
|
||||||
|
|
||||||
def _segment_commands(inputs: list[AsciinemaEvent]) -> tuple[Command, ...]:
|
def _segment_commands(inputs: list[AsciinemaEvent]) -> tuple[Command, ...]:
|
||||||
"""Walk input events, splitting on ``\\r`` / ``\\n`` into commands.
|
"""Walk input events, splitting on ``\\r`` / ``\\n`` into commands.
|
||||||
|
|
||||||
@@ -179,6 +199,7 @@ def build_session_context(
|
|||||||
max(0.0, inputs[i][0] - inputs[i - 1][0]) for i in range(1, len(inputs))
|
max(0.0, inputs[i][0] - inputs[i - 1][0]) for i in range(1, len(inputs))
|
||||||
)
|
)
|
||||||
paste_bursts, paste_count = _detect_paste_bursts(inputs)
|
paste_bursts, paste_count = _detect_paste_bursts(inputs)
|
||||||
|
typing_bursts = _split_typing_bursts(iats)
|
||||||
commands = _segment_commands(inputs)
|
commands = _segment_commands(inputs)
|
||||||
inter_cmd_iats = tuple(
|
inter_cmd_iats = tuple(
|
||||||
max(0.0, commands[i + 1].start_ts - commands[i].end_ts)
|
max(0.0, commands[i + 1].start_ts - commands[i].end_ts)
|
||||||
@@ -204,4 +225,5 @@ def build_session_context(
|
|||||||
commands=commands,
|
commands=commands,
|
||||||
inter_cmd_iats=inter_cmd_iats,
|
inter_cmd_iats=inter_cmd_iats,
|
||||||
output_per_cmd=output_per_cmd,
|
output_per_cmd=output_per_cmd,
|
||||||
|
typing_bursts=typing_bursts,
|
||||||
)
|
)
|
||||||
|
|||||||
@@ -19,6 +19,7 @@ from decnet.profiler.behave_shell._features.cognitive import (
|
|||||||
)
|
)
|
||||||
from decnet.profiler.behave_shell._features.motor import (
|
from decnet.profiler.behave_shell._features.motor import (
|
||||||
input_modality,
|
input_modality,
|
||||||
|
keystroke_cadence,
|
||||||
paste_burst_rate,
|
paste_burst_rate,
|
||||||
)
|
)
|
||||||
|
|
||||||
@@ -27,6 +28,7 @@ FeatureFn = Callable[[SessionContext], Iterable[Observation]]
|
|||||||
FEATURES: tuple[FeatureFn, ...] = (
|
FEATURES: tuple[FeatureFn, ...] = (
|
||||||
input_modality,
|
input_modality,
|
||||||
paste_burst_rate,
|
paste_burst_rate,
|
||||||
|
keystroke_cadence,
|
||||||
inter_command_latency_class,
|
inter_command_latency_class,
|
||||||
command_branch_diversity,
|
command_branch_diversity,
|
||||||
feedback_loop_engagement,
|
feedback_loop_engagement,
|
||||||
|
|||||||
@@ -2,9 +2,12 @@
|
|||||||
|
|
||||||
Step 2: ``motor.input_modality`` — typed / pasted / mixed.
|
Step 2: ``motor.input_modality`` — typed / pasted / mixed.
|
||||||
Step 3: ``motor.paste_burst_rate`` — none / occasional / habitual.
|
Step 3: ``motor.paste_burst_rate`` — none / occasional / habitual.
|
||||||
|
Step B.1: ``motor.keystroke_cadence`` — steady / bursty / hunt_and_peck / machine.
|
||||||
"""
|
"""
|
||||||
from __future__ import annotations
|
from __future__ import annotations
|
||||||
|
|
||||||
|
import statistics
|
||||||
|
from itertools import chain
|
||||||
from typing import Iterator
|
from typing import Iterator
|
||||||
|
|
||||||
from decnet_behave_core.spec.envelope import Observation
|
from decnet_behave_core.spec.envelope import Observation
|
||||||
@@ -12,6 +15,11 @@ from decnet_behave_core.spec.envelope import Observation
|
|||||||
from decnet.profiler.behave_shell._ctx import SessionContext
|
from decnet.profiler.behave_shell._ctx import SessionContext
|
||||||
from decnet.profiler.behave_shell._features._emit import make_observation
|
from decnet.profiler.behave_shell._features._emit import make_observation
|
||||||
from decnet.profiler.behave_shell._thresholds import (
|
from decnet.profiler.behave_shell._thresholds import (
|
||||||
|
CV_BURSTY_MAX,
|
||||||
|
CV_MACHINE_MAX,
|
||||||
|
CV_STEADY_MAX,
|
||||||
|
IKI_MACHINE_MAX_S,
|
||||||
|
MIN_INPUTS_FOR_CADENCE,
|
||||||
MODALITY_PASTED_MIN,
|
MODALITY_PASTED_MIN,
|
||||||
MODALITY_TYPED_MAX,
|
MODALITY_TYPED_MAX,
|
||||||
PASTE_RATE_HABITUAL_MIN,
|
PASTE_RATE_HABITUAL_MIN,
|
||||||
@@ -76,3 +84,45 @@ def paste_burst_rate(ctx: SessionContext) -> Iterator[Observation]:
|
|||||||
value=level,
|
value=level,
|
||||||
confidence=confidence,
|
confidence=confidence,
|
||||||
)
|
)
|
||||||
|
|
||||||
|
|
||||||
|
def keystroke_cadence(ctx: SessionContext) -> Iterator[Observation]:
|
||||||
|
"""Emit ``motor.keystroke_cadence`` ∈ {steady, bursty, hunt_and_peck, machine}.
|
||||||
|
|
||||||
|
Median CV of within-typing-burst IATs (bursts split at gaps >
|
||||||
|
``IKI_THINK_MAX_S`` so think-pauses between commands don't
|
||||||
|
inflate the variance). Pasted-only sessions and sessions below
|
||||||
|
``MIN_INPUTS_FOR_CADENCE`` skip emission — no honest cadence
|
||||||
|
available.
|
||||||
|
|
||||||
|
v0.1 emits only the burst-CV variant. The prototype's NAIVE
|
||||||
|
session-CV variant (lower confidence, second emission per
|
||||||
|
primitive) is parked for v0.2.
|
||||||
|
"""
|
||||||
|
if len(ctx.input_events) < MIN_INPUTS_FOR_CADENCE:
|
||||||
|
return
|
||||||
|
if not ctx.typing_bursts:
|
||||||
|
return
|
||||||
|
burst_cvs: list[float] = []
|
||||||
|
for b in ctx.typing_bursts:
|
||||||
|
m = statistics.fmean(b)
|
||||||
|
if m > 0:
|
||||||
|
burst_cvs.append(statistics.pstdev(b) / m)
|
||||||
|
if not burst_cvs:
|
||||||
|
return
|
||||||
|
cv = statistics.median(burst_cvs)
|
||||||
|
mean_iki = statistics.fmean(chain.from_iterable(ctx.typing_bursts))
|
||||||
|
if mean_iki < IKI_MACHINE_MAX_S and cv < CV_MACHINE_MAX:
|
||||||
|
value, confidence = "machine", 0.85
|
||||||
|
elif cv < CV_STEADY_MAX:
|
||||||
|
value, confidence = "steady", 0.70
|
||||||
|
elif cv < CV_BURSTY_MAX:
|
||||||
|
value, confidence = "bursty", 0.65
|
||||||
|
else:
|
||||||
|
value, confidence = "hunt_and_peck", 0.60
|
||||||
|
yield make_observation(
|
||||||
|
ctx,
|
||||||
|
primitive="motor.keystroke_cadence",
|
||||||
|
value=value,
|
||||||
|
confidence=confidence,
|
||||||
|
)
|
||||||
|
|||||||
@@ -75,3 +75,18 @@ FEEDBACK_MIN_PAIRS: int = 5
|
|||||||
# via Hartigan dip is filed for v0.2).
|
# via Hartigan dip is filed for v0.2).
|
||||||
PAUSE_CV_METRONOMIC_MAX: float = 0.40
|
PAUSE_CV_METRONOMIC_MAX: float = 0.40
|
||||||
PAUSE_CV_BIMODAL_MIN: float = 1.50
|
PAUSE_CV_BIMODAL_MIN: float = 1.50
|
||||||
|
|
||||||
|
# ── motor.keystroke_cadence (Step B.1) ──────────────────────────────────────
|
||||||
|
# Typing bursts split at gaps > IKI_THINK_MAX_S so think-pauses between
|
||||||
|
# commands don't inflate the within-burst CV. Mirrors the prototype's
|
||||||
|
# _split_into_bursts (BEHAVE/prototype_extractors/shell/extract.py:275-286).
|
||||||
|
IKI_THINK_MAX_S: float = 1.50
|
||||||
|
# Sub-human floor for the "machine" classification — only paired with a
|
||||||
|
# pathologically uniform CV, since real humans never produce sub-5ms IATs
|
||||||
|
# in a sustained burst.
|
||||||
|
IKI_MACHINE_MAX_S: float = 0.005
|
||||||
|
CV_MACHINE_MAX: float = 0.05
|
||||||
|
CV_STEADY_MAX: float = 0.50
|
||||||
|
CV_BURSTY_MAX: float = 1.50
|
||||||
|
# Need this many input events before we'll claim a cadence at all.
|
||||||
|
MIN_INPUTS_FOR_CADENCE: int = 5
|
||||||
|
|||||||
@@ -31,13 +31,16 @@ from decnet.profiler.behave_shell import extract_session
|
|||||||
from decnet.profiler.behave_shell._parse import parse_shard_line
|
from decnet.profiler.behave_shell._parse import parse_shard_line
|
||||||
|
|
||||||
|
|
||||||
PHASE_A_PRIMITIVES: frozenset[str] = frozenset({
|
PHASE_AB_PRIMITIVES: frozenset[str] = frozenset({
|
||||||
|
# Phase A — calibration floor
|
||||||
"motor.input_modality",
|
"motor.input_modality",
|
||||||
"motor.paste_burst_rate",
|
"motor.paste_burst_rate",
|
||||||
"cognitive.inter_command_latency_class",
|
"cognitive.inter_command_latency_class",
|
||||||
"cognitive.command_branch_diversity",
|
"cognitive.command_branch_diversity",
|
||||||
"cognitive.feedback_loop_engagement",
|
"cognitive.feedback_loop_engagement",
|
||||||
"cognitive.inter_command_consistency",
|
"cognitive.inter_command_consistency",
|
||||||
|
# Phase B — motor.* completion (lands one primitive per commit)
|
||||||
|
"motor.keystroke_cadence",
|
||||||
})
|
})
|
||||||
|
|
||||||
|
|
||||||
@@ -105,7 +108,7 @@ def test_shard_emits_all_phase_a_primitives(
|
|||||||
obs = _all_observations(path)
|
obs = _all_observations(path)
|
||||||
assert obs, f"{class_label}: extractor produced zero observations"
|
assert obs, f"{class_label}: extractor produced zero observations"
|
||||||
seen = {o.primitive for o in obs}
|
seen = {o.primitive for o in obs}
|
||||||
missing = PHASE_A_PRIMITIVES - seen
|
missing = PHASE_AB_PRIMITIVES - seen
|
||||||
assert not missing, (
|
assert not missing, (
|
||||||
f"{class_label} ({shard_file}) missing primitives: "
|
f"{class_label} ({shard_file}) missing primitives: "
|
||||||
f"{sorted(missing)}"
|
f"{sorted(missing)}"
|
||||||
@@ -142,7 +145,7 @@ def test_shards_are_discriminative_across_classes(
|
|||||||
# At least one primitive should produce different majority values
|
# At least one primitive should produce different majority values
|
||||||
# across the present classes.
|
# across the present classes.
|
||||||
discriminative_primitives: list[str] = []
|
discriminative_primitives: list[str] = []
|
||||||
for prim in PHASE_A_PRIMITIVES:
|
for prim in PHASE_AB_PRIMITIVES:
|
||||||
values = {by_class[c].get(prim) for c in by_class if prim in by_class[c]}
|
values = {by_class[c].get(prim) for c in by_class if prim in by_class[c]}
|
||||||
if len(values) >= 2:
|
if len(values) >= 2:
|
||||||
discriminative_primitives.append(prim)
|
discriminative_primitives.append(prim)
|
||||||
|
|||||||
85
tests/profiler/behave_shell/test_motor_keystroke_cadence.py
Normal file
85
tests/profiler/behave_shell/test_motor_keystroke_cadence.py
Normal file
@@ -0,0 +1,85 @@
|
|||||||
|
"""Step B.1: ``motor.keystroke_cadence``."""
|
||||||
|
from __future__ import annotations
|
||||||
|
|
||||||
|
import random
|
||||||
|
|
||||||
|
from decnet.profiler.behave_shell import extract_session
|
||||||
|
from decnet.profiler.behave_shell._parse import AsciinemaEvent
|
||||||
|
|
||||||
|
|
||||||
|
def _of(observations: list, primitive: str):
|
||||||
|
obs = [o for o in observations if o.primitive == primitive]
|
||||||
|
assert len(obs) == 1, f"expected exactly one {primitive}, got {len(obs)}"
|
||||||
|
return obs[0]
|
||||||
|
|
||||||
|
|
||||||
|
def _typed_events(iats: list[float], terminator: bool = True) -> list[AsciinemaEvent]:
|
||||||
|
"""Build a typed input stream where consecutive single-char events are
|
||||||
|
separated by ``iats``."""
|
||||||
|
events: list[AsciinemaEvent] = []
|
||||||
|
t = 0.0
|
||||||
|
events.append((t, "i", "a"))
|
||||||
|
for x in iats:
|
||||||
|
t += x
|
||||||
|
events.append((t, "i", "b"))
|
||||||
|
if terminator:
|
||||||
|
events.append((t + 0.1, "i", "\r"))
|
||||||
|
return events
|
||||||
|
|
||||||
|
|
||||||
|
def test_too_few_inputs_no_emission() -> None:
|
||||||
|
out = list(extract_session(_typed_events([0.1, 0.1]), sid="cad-low"))
|
||||||
|
assert [o for o in out if o.primitive == "motor.keystroke_cadence"] == []
|
||||||
|
|
||||||
|
|
||||||
|
def test_huge_think_pauses_yield_no_typing_bursts() -> None:
|
||||||
|
# Two events 5s apart → no IAT under IKI_THINK_MAX_S, and only 1
|
||||||
|
# IAT total — below the 3-IAT-per-burst minimum. No burst, no emit.
|
||||||
|
events: list[AsciinemaEvent] = [
|
||||||
|
(0.0, "i", "a"),
|
||||||
|
(5.0, "i", "b"),
|
||||||
|
(10.0, "i", "c"),
|
||||||
|
(15.0, "i", "d"),
|
||||||
|
(20.0, "i", "e"),
|
||||||
|
]
|
||||||
|
out = list(extract_session(events, sid="cad-no-bursts"))
|
||||||
|
assert [o for o in out if o.primitive == "motor.keystroke_cadence"] == []
|
||||||
|
|
||||||
|
|
||||||
|
def test_uniform_iats_emit_steady() -> None:
|
||||||
|
iats = [0.15] * 12
|
||||||
|
out = list(extract_session(_typed_events(iats), sid="cad-steady"))
|
||||||
|
obs = _of(out, "motor.keystroke_cadence")
|
||||||
|
assert obs.value == "steady"
|
||||||
|
assert obs.confidence == 0.70
|
||||||
|
|
||||||
|
|
||||||
|
def test_machine_iats_emit_machine() -> None:
|
||||||
|
# Sub-5ms IATs with near-zero CV — no terminator IAT to inflate the
|
||||||
|
# variance away from machine
|
||||||
|
iats = [0.002] * 20
|
||||||
|
out = list(extract_session(_typed_events(iats, terminator=False), sid="cad-machine"))
|
||||||
|
obs = _of(out, "motor.keystroke_cadence")
|
||||||
|
assert obs.value == "machine"
|
||||||
|
assert obs.confidence == 0.85
|
||||||
|
|
||||||
|
|
||||||
|
def test_bursty_iats_emit_bursty() -> None:
|
||||||
|
# Mean ~0.15 with moderate variance, CV between 0.5 and 1.5
|
||||||
|
rng = random.Random(42)
|
||||||
|
iats = []
|
||||||
|
for _ in range(20):
|
||||||
|
# Mostly fast, occasionally slow → CV in the bursty band
|
||||||
|
iats.append(rng.choice([0.05, 0.05, 0.05, 0.30, 0.50]))
|
||||||
|
out = list(extract_session(_typed_events(iats), sid="cad-bursty"))
|
||||||
|
obs = _of(out, "motor.keystroke_cadence")
|
||||||
|
assert obs.value == "bursty"
|
||||||
|
|
||||||
|
|
||||||
|
def test_hunt_and_peck_iats_emit_hunt_and_peck() -> None:
|
||||||
|
# CV >= 1.5: extreme bimodal (very-fast + very-slow within burst).
|
||||||
|
# Most IATs are tiny; a few are ~10x the mean — drives stdev/mean above 1.5.
|
||||||
|
iats = [0.01] * 15 + [1.4] * 5
|
||||||
|
out = list(extract_session(_typed_events(iats, terminator=False), sid="cad-hp"))
|
||||||
|
obs = _of(out, "motor.keystroke_cadence")
|
||||||
|
assert obs.value == "hunt_and_peck"
|
||||||
Reference in New Issue
Block a user