Refactor into standard Python package with uv
- Move all code into src/mouse_control/ following the src-layout convention, with model/collect/train/visualize/inference as separate modules and SimpleNet extracted into model.py - Add hatchling-based pyproject.toml with three CLI entry points (mouse-collect, mouse-train, mouse-visualize) and bundle the trained ONNX model as a package resource under assets/ - Move training data to data/, delete superseded show.py, remove dead imports (numpy, onnxruntime, onnx, onnxsim) and the unused visualize_path() helper - Fix crashes in the collector: guard against destroyed-widget access after the n==100 destroy(), and skip save_to_csv when recording is off or the path is too short to derive 10 keypoints - Switch ONNX export to external_data=False so the model ships as a single self-contained 19KB file - Bytes-load the bundled model via importlib.resources so packaging remains correct under zip-safe distributions - Rewrite README around the new layout, commands, and public API
This commit is contained in:
7
src/mouse_control/__init__.py
Normal file
7
src/mouse_control/__init__.py
Normal file
@@ -0,0 +1,7 @@
|
||||
"""Neural-network mouse trajectory generation that mimics human motion."""
|
||||
|
||||
from mouse_control.inference import load_model, predict
|
||||
from mouse_control.model import SimpleNet
|
||||
|
||||
__version__ = "0.1.0"
|
||||
__all__ = ["SimpleNet", "load_model", "predict", "__version__"]
|
||||
0
src/mouse_control/assets/__init__.py
Normal file
0
src/mouse_control/assets/__init__.py
Normal file
BIN
src/mouse_control/assets/mouse.onnx
Normal file
BIN
src/mouse_control/assets/mouse.onnx
Normal file
Binary file not shown.
133
src/mouse_control/collect.py
Normal file
133
src/mouse_control/collect.py
Normal file
@@ -0,0 +1,133 @@
|
||||
import argparse
|
||||
import csv
|
||||
import random
|
||||
import tkinter as tk
|
||||
from pathlib import Path
|
||||
|
||||
BG_COLOR = "#2b2b2b"
|
||||
BALL_RADIUS = 20
|
||||
TARGET_OFFSET_RANGE = 200
|
||||
N_KEYPOINTS = 10
|
||||
|
||||
|
||||
class TrajectoryCollector:
|
||||
def __init__(self, csv_path: Path, n_trajectories: int = 100) -> None:
|
||||
self.csv_path = Path(csv_path)
|
||||
self.n_target = n_trajectories
|
||||
|
||||
self.root = tk.Tk()
|
||||
self.root.attributes("-fullscreen", True)
|
||||
self.root.configure(bg=BG_COLOR)
|
||||
|
||||
screen_w = self.root.winfo_screenwidth()
|
||||
screen_h = self.root.winfo_screenheight()
|
||||
self.ball1_pos = (screen_w / 2, screen_h / 2)
|
||||
self.ball2_pos = self._new_target()
|
||||
|
||||
self.label = tk.Label(
|
||||
self.root, text="n: 0", font=("Helvetica", 16),
|
||||
bg=BG_COLOR, fg="white",
|
||||
)
|
||||
self.label.pack()
|
||||
self.canvas = tk.Canvas(
|
||||
self.root, width=screen_w, height=screen_h,
|
||||
bg=BG_COLOR, highlightthickness=0,
|
||||
)
|
||||
self.canvas.pack()
|
||||
self._draw_ball(self.ball1_pos, "red")
|
||||
self._draw_ball(self.ball2_pos, "blue", tags="ball2")
|
||||
|
||||
self.recording = False
|
||||
self.mouse_path: list[tuple[int, int]] = []
|
||||
self.n = 0
|
||||
|
||||
self.canvas.bind("<Motion>", self._on_motion)
|
||||
self.canvas.bind("<Button-1>", self._on_click)
|
||||
self.root.bind("<Key>", self._on_key)
|
||||
|
||||
def _new_target(self) -> tuple[float, float]:
|
||||
cx, cy = self.ball1_pos
|
||||
return (
|
||||
cx + random.randint(-TARGET_OFFSET_RANGE, TARGET_OFFSET_RANGE),
|
||||
cy + random.randint(-TARGET_OFFSET_RANGE, TARGET_OFFSET_RANGE),
|
||||
)
|
||||
|
||||
def _draw_ball(self, pos: tuple[float, float], color: str, tags: str = "") -> None:
|
||||
x, y = pos
|
||||
self.canvas.create_oval(
|
||||
x - BALL_RADIUS, y - BALL_RADIUS,
|
||||
x + BALL_RADIUS, y + BALL_RADIUS,
|
||||
fill=color, tags=tags,
|
||||
)
|
||||
|
||||
@staticmethod
|
||||
def _hit(event: tk.Event, pos: tuple[float, float]) -> bool:
|
||||
return (
|
||||
abs(event.x - pos[0]) <= BALL_RADIUS
|
||||
and abs(event.y - pos[1]) <= BALL_RADIUS
|
||||
)
|
||||
|
||||
def _on_motion(self, event: tk.Event) -> None:
|
||||
if self.recording:
|
||||
self.mouse_path.append((event.x, event.y))
|
||||
|
||||
def _on_click(self, event: tk.Event) -> None:
|
||||
if self._hit(event, self.ball1_pos):
|
||||
self.recording = True
|
||||
self.mouse_path = [(event.x, event.y)]
|
||||
elif self._hit(event, self.ball2_pos):
|
||||
if not self.recording or len(self.mouse_path) < N_KEYPOINTS:
|
||||
return
|
||||
self.recording = False
|
||||
self.canvas.delete("ball2")
|
||||
self._save_path(self.mouse_path)
|
||||
self.n += 1
|
||||
if self.n >= self.n_target:
|
||||
self.root.destroy()
|
||||
return
|
||||
self.label.config(text=f"n: {self.n}")
|
||||
self.mouse_path = []
|
||||
self.ball2_pos = self._new_target()
|
||||
self._draw_ball(self.ball2_pos, "blue", tags="ball2")
|
||||
|
||||
def _on_key(self, event: tk.Event) -> None:
|
||||
if event.keysym == "Escape":
|
||||
self.root.destroy()
|
||||
|
||||
def _save_path(self, path: list[tuple[int, int]]) -> None:
|
||||
origin_x, origin_y = path[0]
|
||||
x_rel = [px - origin_x for px, _ in path]
|
||||
y_rel = [-(py - origin_y) for _, py in path]
|
||||
step = max(1, len(path) // N_KEYPOINTS)
|
||||
idx = list(range(0, len(path), step))
|
||||
kx = [x_rel[i] for i in idx]
|
||||
ky = [y_rel[i] for i in idx]
|
||||
with open(self.csv_path, "a", newline="") as f:
|
||||
writer = csv.writer(f, quoting=csv.QUOTE_MINIMAL)
|
||||
row = [f"{kx[-1]},{ky[-1]}"] + [
|
||||
f"{kx[i]},{ky[i]}" for i in range(N_KEYPOINTS)
|
||||
]
|
||||
writer.writerow(row)
|
||||
|
||||
def run(self) -> None:
|
||||
self.root.mainloop()
|
||||
|
||||
|
||||
def main() -> None:
|
||||
parser = argparse.ArgumentParser(
|
||||
description="Collect human mouse trajectories via a fullscreen Tk GUI.",
|
||||
)
|
||||
parser.add_argument(
|
||||
"--output", "-o", type=Path, default=Path("mouse_data.csv"),
|
||||
help="CSV file to append trajectories to (default: mouse_data.csv).",
|
||||
)
|
||||
parser.add_argument(
|
||||
"--count", "-n", type=int, default=100,
|
||||
help="Number of trajectories to collect before exiting (default: 100).",
|
||||
)
|
||||
args = parser.parse_args()
|
||||
TrajectoryCollector(args.output, args.count).run()
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
main()
|
||||
28
src/mouse_control/inference.py
Normal file
28
src/mouse_control/inference.py
Normal file
@@ -0,0 +1,28 @@
|
||||
from importlib.resources import files
|
||||
from pathlib import Path
|
||||
|
||||
import numpy as np
|
||||
import onnxruntime as ort
|
||||
|
||||
|
||||
def load_model(model_path: str | Path | None = None) -> ort.InferenceSession:
|
||||
"""Load the ONNX mouse-trajectory model.
|
||||
|
||||
Resolution order when ``model_path`` is None:
|
||||
1. ``./mouse.onnx`` in the current working directory (freshly trained).
|
||||
2. The model bundled inside the installed package.
|
||||
"""
|
||||
if model_path is not None:
|
||||
return ort.InferenceSession(str(model_path))
|
||||
cwd_model = Path("mouse.onnx")
|
||||
if cwd_model.is_file():
|
||||
return ort.InferenceSession(str(cwd_model))
|
||||
data = files("mouse_control.assets").joinpath("mouse.onnx").read_bytes()
|
||||
return ort.InferenceSession(data)
|
||||
|
||||
|
||||
def predict(session: ort.InferenceSession, dx: float, dy: float) -> np.ndarray:
|
||||
"""Return a (10, 2) array of trajectory points for the given displacement."""
|
||||
inp = np.array([[[float(dx), float(dy)]]], dtype=np.float32)
|
||||
input_name = session.get_inputs()[0].name
|
||||
return session.run(None, {input_name: inp})[0][0]
|
||||
17
src/mouse_control/model.py
Normal file
17
src/mouse_control/model.py
Normal file
@@ -0,0 +1,17 @@
|
||||
import torch
|
||||
import torch.nn as nn
|
||||
|
||||
|
||||
class SimpleNet(nn.Module):
|
||||
def __init__(self) -> None:
|
||||
super().__init__()
|
||||
self.fc1 = nn.Linear(2, 64)
|
||||
self.fc2 = nn.Linear(64, 32)
|
||||
self.fc3 = nn.Linear(32, 20)
|
||||
|
||||
def forward(self, x: torch.Tensor) -> torch.Tensor:
|
||||
x = torch.flatten(x, start_dim=1)
|
||||
x = nn.functional.relu(self.fc1(x))
|
||||
x = nn.functional.relu(self.fc2(x))
|
||||
x = self.fc3(x)
|
||||
return x.view(-1, 10, 2)
|
||||
113
src/mouse_control/train.py
Normal file
113
src/mouse_control/train.py
Normal file
@@ -0,0 +1,113 @@
|
||||
import argparse
|
||||
from pathlib import Path
|
||||
|
||||
import pandas as pd
|
||||
import torch
|
||||
import torch.nn as nn
|
||||
import torch.optim as optim
|
||||
from torch.utils.data import DataLoader, Dataset
|
||||
|
||||
from mouse_control.model import SimpleNet
|
||||
|
||||
|
||||
class TrajectoryDataset(Dataset):
|
||||
def __init__(self, dx_dy: torch.Tensor, labels: torch.Tensor) -> None:
|
||||
self.dx_dy = dx_dy
|
||||
self.labels = labels
|
||||
|
||||
def __len__(self) -> int:
|
||||
return len(self.dx_dy)
|
||||
|
||||
def __getitem__(self, idx: int) -> tuple[torch.Tensor, torch.Tensor]:
|
||||
return self.dx_dy[idx], self.labels[idx]
|
||||
|
||||
|
||||
def _load_csv(csv_path: Path) -> tuple[torch.Tensor, torch.Tensor]:
|
||||
df = pd.read_csv(csv_path, header=None)
|
||||
inputs = df.iloc[:, 0].apply(lambda x: list(map(int, x.split(","))))
|
||||
labels = df.apply(
|
||||
lambda row: [list(map(int, row[i].split(","))) for i in range(1, 11)],
|
||||
axis=1,
|
||||
)
|
||||
inputs_t = torch.Tensor(inputs.tolist()).unsqueeze(1)
|
||||
labels_t = torch.Tensor(labels.tolist())
|
||||
return inputs_t, labels_t
|
||||
|
||||
|
||||
def train(
|
||||
train_csv: Path,
|
||||
test_csv: Path,
|
||||
output: Path,
|
||||
epochs: int = 1000,
|
||||
batch_size: int = 64,
|
||||
lr: float = 1e-3,
|
||||
) -> None:
|
||||
train_x, train_y = _load_csv(train_csv)
|
||||
test_x, test_y = _load_csv(test_csv)
|
||||
|
||||
train_loader = DataLoader(
|
||||
TrajectoryDataset(train_x, train_y), batch_size=batch_size, shuffle=True,
|
||||
)
|
||||
test_loader = DataLoader(
|
||||
TrajectoryDataset(test_x, test_y), batch_size=batch_size, shuffle=False,
|
||||
)
|
||||
|
||||
model = SimpleNet()
|
||||
criterion = nn.MSELoss()
|
||||
optimizer = optim.Adam(model.parameters(), lr=lr)
|
||||
scheduler = optim.lr_scheduler.StepLR(optimizer, step_size=50, gamma=0.9)
|
||||
|
||||
model.train()
|
||||
for epoch in range(epochs):
|
||||
epoch_loss = 0.0
|
||||
for batch_x, batch_y in train_loader:
|
||||
optimizer.zero_grad()
|
||||
output_t = model(batch_x)
|
||||
loss = criterion(output_t, batch_y)
|
||||
loss.backward()
|
||||
optimizer.step()
|
||||
scheduler.step()
|
||||
epoch_loss += loss.item()
|
||||
if (epoch + 1) % 50 == 0 or epoch == 0:
|
||||
print(f"epoch {epoch + 1:>4}/{epochs} train_loss={epoch_loss / len(train_loader):.4f}")
|
||||
|
||||
model.eval()
|
||||
with torch.no_grad():
|
||||
total = 0.0
|
||||
last_batch = None
|
||||
for batch_x, batch_y in test_loader:
|
||||
pred = model(batch_x)
|
||||
total += criterion(pred, batch_y).item()
|
||||
last_batch = (batch_x, pred)
|
||||
print(f"test_loss={total / len(test_loader):.4f}")
|
||||
if last_batch is not None:
|
||||
sample_x, sample_pred = last_batch
|
||||
print("sample input :", sample_x[0].tolist())
|
||||
print("sample output:", sample_pred[0].tolist())
|
||||
|
||||
dummy = torch.randn(1, 1, 2)
|
||||
torch.onnx.export(
|
||||
model, dummy, str(output),
|
||||
input_names=["input"], output_names=["output"],
|
||||
external_data=False,
|
||||
)
|
||||
print(f"exported -> {output}")
|
||||
|
||||
|
||||
def main() -> None:
|
||||
parser = argparse.ArgumentParser(description="Train the mouse trajectory model.")
|
||||
parser.add_argument("--train-csv", type=Path, default=Path("data/mouse_data.csv"))
|
||||
parser.add_argument("--test-csv", type=Path, default=Path("data/mouse_data_test.csv"))
|
||||
parser.add_argument("--output", "-o", type=Path, default=Path("mouse.onnx"))
|
||||
parser.add_argument("--epochs", type=int, default=1000)
|
||||
parser.add_argument("--batch-size", type=int, default=64)
|
||||
parser.add_argument("--lr", type=float, default=1e-3)
|
||||
args = parser.parse_args()
|
||||
train(
|
||||
args.train_csv, args.test_csv, args.output,
|
||||
epochs=args.epochs, batch_size=args.batch_size, lr=args.lr,
|
||||
)
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
main()
|
||||
104
src/mouse_control/visualize.py
Normal file
104
src/mouse_control/visualize.py
Normal file
@@ -0,0 +1,104 @@
|
||||
import argparse
|
||||
from pathlib import Path
|
||||
|
||||
import matplotlib.pyplot as plt
|
||||
import numpy as np
|
||||
from scipy.interpolate import CubicSpline
|
||||
|
||||
from mouse_control.inference import load_model, predict
|
||||
|
||||
|
||||
def _interpolate(xs: np.ndarray, ys: np.ndarray, n_points: int) -> tuple[np.ndarray, np.ndarray]:
|
||||
chord = np.cumsum(np.sqrt(np.diff(xs) ** 2 + np.diff(ys) ** 2))
|
||||
t = np.concatenate([[0.0], chord])
|
||||
if t[-1] == 0:
|
||||
return xs, ys
|
||||
cs_x = CubicSpline(t, xs)
|
||||
cs_y = CubicSpline(t, ys)
|
||||
t_dense = np.linspace(0, t[-1], n_points)
|
||||
return cs_x(t_dense), cs_y(t_dense)
|
||||
|
||||
|
||||
def visualize(
|
||||
model_path: Path | None = None,
|
||||
n_trajectories: int = 10,
|
||||
n_interp: int = 80,
|
||||
seed: int | None = None,
|
||||
save_path: Path | None = Path("trajectories.png"),
|
||||
show: bool = True,
|
||||
) -> None:
|
||||
rng = np.random.default_rng(seed)
|
||||
targets = rng.integers(low=-200, high=201, size=(n_trajectories, 2))
|
||||
|
||||
session = load_model(model_path)
|
||||
cmap = plt.get_cmap("tab10")
|
||||
fig, ax = plt.subplots(figsize=(10, 8))
|
||||
|
||||
for i, (dx, dy) in enumerate(targets):
|
||||
pred = predict(session, dx, dy) # (10, 2)
|
||||
xs = np.concatenate([[0.0], pred[:, 0]])
|
||||
ys = np.concatenate([[0.0], pred[:, 1]])
|
||||
xs_dense, ys_dense = _interpolate(xs, ys, n_interp)
|
||||
|
||||
color = cmap(i % cmap.N)
|
||||
ax.plot(xs_dense, ys_dense, "-", color=color, alpha=0.85, linewidth=1.6)
|
||||
ax.scatter(xs_dense, ys_dense, color=color, s=6, alpha=0.5, zorder=2)
|
||||
ax.scatter(
|
||||
pred[:, 0], pred[:, 1], color=color, s=35, zorder=3,
|
||||
edgecolors="white", linewidths=0.6,
|
||||
)
|
||||
ax.scatter(
|
||||
[dx], [dy], color=color, marker="x", s=140, linewidths=2.5,
|
||||
zorder=4, label=f"target ({dx}, {dy})",
|
||||
)
|
||||
|
||||
ax.scatter([0], [0], color="black", s=90, zorder=5, label="start (0, 0)")
|
||||
ax.axhline(0, color="gray", linewidth=0.5, alpha=0.5)
|
||||
ax.axvline(0, color="gray", linewidth=0.5, alpha=0.5)
|
||||
ax.set_aspect("equal")
|
||||
ax.set_title(
|
||||
f"Mouse trajectories — 10 model points + cubic-spline interpolation ({n_interp} pts)",
|
||||
)
|
||||
ax.set_xlabel("dx")
|
||||
ax.set_ylabel("dy")
|
||||
ax.legend(loc="best", fontsize=8, ncol=2)
|
||||
ax.grid(True, alpha=0.3)
|
||||
plt.tight_layout()
|
||||
|
||||
if save_path is not None:
|
||||
plt.savefig(save_path, dpi=120)
|
||||
print(f"saved -> {save_path}")
|
||||
if show:
|
||||
plt.show()
|
||||
plt.close(fig)
|
||||
|
||||
|
||||
def main() -> None:
|
||||
parser = argparse.ArgumentParser(
|
||||
description="Render predicted mouse trajectories for random targets.",
|
||||
)
|
||||
parser.add_argument(
|
||||
"--model", type=Path, default=None,
|
||||
help="Path to ONNX model (defaults to ./mouse.onnx or bundled model).",
|
||||
)
|
||||
parser.add_argument("--n-trajectories", "-n", type=int, default=10)
|
||||
parser.add_argument("--n-interp", type=int, default=80)
|
||||
parser.add_argument("--seed", type=int, default=None)
|
||||
parser.add_argument("--save", type=Path, default=Path("trajectories.png"))
|
||||
parser.add_argument(
|
||||
"--no-show", action="store_true",
|
||||
help="Skip opening the matplotlib window (still writes --save).",
|
||||
)
|
||||
args = parser.parse_args()
|
||||
visualize(
|
||||
model_path=args.model,
|
||||
n_trajectories=args.n_trajectories,
|
||||
n_interp=args.n_interp,
|
||||
seed=args.seed,
|
||||
save_path=args.save,
|
||||
show=not args.no_show,
|
||||
)
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
main()
|
||||
Reference in New Issue
Block a user