Merge pull request #3 from k23172173/template-branch

.gitignore

@@ -0,0 +1,36 @@
+# Python
+__pycache__/
+*.py[cod]
+*$py.class
+*.so
+.Python
+venv/
+.venv/
+env/
+.env
+*.egg-info/
+.eggs/
+dist/
+build/
+
+# IDE
+.idea/
+.vscode/
+*.swp
+*.swo
+
+# Data and outputs (optional: uncomment if you don’t want to track large files)
+# data_preparation/raw/
+# data_preparation/processed/*.npy
+# evaluation/logs/
+# evaluation/results/
+
+# Model checkpoints (uncomment to ignore .pt files)
+# *.pt
+
+# Project
+docs/
+
+# OS
+.DS_Store
+Thumbs.db

README.md

@@ -1 +1,10 @@
-# GAP_Large_project
+# GAP — FocusGuard
+
+Real-time focus estimation from webcam (head pose + eye behaviour). 
+
+## Layout
+
+- **data_preparation/** — Dataset team (raw data, processed, scripts)
+- **models/** — Face orientation, eye behaviour, fusion, landmarks. Training entry: `models/train.py`
+- **evaluation/** — Metrics, runs, results
+- **ui/** — Live demo + session view

data_preparation/README.md

@@ -0,0 +1,3 @@
+# data_preparation
+
+Dataset team owns layout and scripts here.

evaluation/README.md

@@ -0,0 +1,3 @@
+# evaluation
+
+Metrics, experiment configs, and results live here.

models/README.md

@@ -0,0 +1,8 @@
+# models
+
+- `face_orientation_model/` — S_face
+- `eye_behaviour_model/` — S_eye
+- `attention_score_fusion/` — fusion + smoothing
+- `face_landmarks_pretrained/` — MediaPipe FaceMesh (no training)
+
+`train.py` trains the MLP on feature vectors; `prepare_dataset.py` loads from `data_preparation/processed/` or synthetic.

models/face_orientation_model/best_model.pt

@@ -0,0 +1,3 @@
+version https://git-lfs.github.com/spec/v1
+oid sha256:18c1f2750c7274e72538b94afcc9f0243287a5b2eb8fcce6be6e4ae18ec59cb0
+size 15033

models/prepare_dataset.py

@@ -0,0 +1,91 @@
+import os
+import numpy as np
+import torch
+from torch.utils.data import Dataset, DataLoader, random_split
+
+DATA_DIR = os.path.join(os.path.dirname(__file__), "..", "data_preparation", "processed")
+
+FEATURE_FILES = {
+    "face_orientation": {
+        "features": "face_orientation_features.npy",
+        "labels": "face_orientation_labels.npy",
+    },
+    "eye_behaviour": {
+        "features": "eye_behaviour_features.npy",
+        "labels": "eye_behaviour_labels.npy",
+    },
+}
+
+SYNTHETIC_CONFIG = {
+    "face_orientation": {"num_samples": 500, "num_features": 12, "num_classes": 2},
+    "eye_behaviour": {"num_samples": 500, "num_features": 8, "num_classes": 2},
+}
+
+
+class FeatureVectorDataset(Dataset):
+    def __init__(self, features: np.ndarray, labels: np.ndarray):
+        self.features = torch.tensor(features, dtype=torch.float32)
+        self.labels = torch.tensor(labels, dtype=torch.long)
+
+    def __len__(self):
+        return len(self.labels)
+
+    def __getitem__(self, idx):
+        return self.features[idx], self.labels[idx]
+
+
+def _load_real_data(model_name: str):
+    file_cfg = FEATURE_FILES.get(model_name)
+    if file_cfg is None:
+        return None
+
+    feat_path = os.path.join(DATA_DIR, file_cfg["features"])
+    label_path = os.path.join(DATA_DIR, file_cfg["labels"])
+
+    if os.path.exists(feat_path) and os.path.exists(label_path):
+        features = np.load(feat_path)
+        labels = np.load(label_path)
+        print(f"[DATA] Loaded real data for '{model_name}': {features.shape[0]} samples, {features.shape[1]} features")
+        return features, labels
+
+    return None
+
+
+def _generate_synthetic_data(model_name: str):
+    cfg = SYNTHETIC_CONFIG.get(model_name, SYNTHETIC_CONFIG["face_orientation"])
+    n = cfg["num_samples"]
+    d = cfg["num_features"]
+    c = cfg["num_classes"]
+
+    rng = np.random.RandomState(42)
+    features = rng.randn(n, d).astype(np.float32)
+    labels = rng.randint(0, c, size=n).astype(np.int64)
+
+    print(f"[DATA] Using synthetic data for '{model_name}': {n} samples, {d} features, {c} classes")
+    return features, labels
+
+
+def get_dataloaders(model_name: str, batch_size: int = 32, split_ratios=(0.7, 0.15, 0.15), seed: int = 42):
+    data = _load_real_data(model_name)
+    if data is None:
+        data = _generate_synthetic_data(model_name)
+
+    features, labels = data
+    num_features = features.shape[1]
+    num_classes = int(labels.max()) + 1
+
+    dataset = FeatureVectorDataset(features, labels)
+    total = len(dataset)
+    train_n = int(total * split_ratios[0])
+    val_n = int(total * split_ratios[1])
+    test_n = total - train_n - val_n
+
+    gen = torch.Generator().manual_seed(seed)
+    train_ds, val_ds, test_ds = random_split(dataset, [train_n, val_n, test_n], generator=gen)
+
+    train_loader = DataLoader(train_ds, batch_size=batch_size, shuffle=True)
+    val_loader = DataLoader(val_ds, batch_size=batch_size, shuffle=False)
+    test_loader = DataLoader(test_ds, batch_size=batch_size, shuffle=False)
+
+    print(f"[DATA] Split: train={train_n}, val={val_n}, test={test_n}")
+    return train_loader, val_loader, test_loader, num_features, num_classes

models/train.py

@@ -0,0 +1,186 @@
+# Run from repo root: python -m models.train (or cd models && python train.py)
+
+import json
+import os
+import random
+
+import numpy as np as np
+import torch
+import torch.nn as nn
+import torch.optim as optim
+
+from prepare_dataset import get_dataloaders
+
+CFG = {
+    "model_name": "face_orientation",  # "face_orientation" or "eye_behaviour"
+    "epochs": 30,
+    "batch_size": 32,
+    "lr": 1e-3,
+    "seed": 42,
+    "split_ratios": (0.7, 0.15, 0.15),
+    "checkpoints_dir": {
+        "face_orientation": os.path.join(os.path.dirname(__file__), "face_orientation_model"),
+        "eye_behaviour": os.path.join(os.path.dirname(__file__), "eye_behaviour_model"),
+    },
+    "logs_dir": os.path.join(os.path.dirname(__file__), "..", "evaluation", "logs"),
+}
+
+
+def set_seed(seed: int):
+    random.seed(seed)
+    np.random.seed(seed)
+    torch.manual_seed(seed)
+    if torch.cuda.is_available():
+        torch.cuda.manual_seed_all(seed)
+
+
+class BaseModel(nn.Module):
+    def __init__(self, num_features: int, num_classes: int):
+        super().__init__()
+        self.network = nn.Sequential(
+            nn.Linear(num_features, 64),
+            nn.ReLU(),
+            nn.Linear(64, 32),
+            nn.ReLU(),
+            nn.Linear(32, num_classes),
+        )
+
+    def forward(self, x):
+        return self.network(x)
+
+    def training_step(self, loader, optimizer, criterion, device):
+        self.train()
+        total_loss = 0.0
+        correct = 0
+        total = 0
+
+        for features, labels in loader:
+            features, labels = features.to(device), labels.to(device)
+
+            optimizer.zero_grad()
+            outputs = self(features)
+            loss = criterion(outputs, labels)
+            loss.backward()
+            optimizer.step()
+
+            total_loss += loss.item() * features.size(0)
+            correct += (outputs.argmax(dim=1) == labels).sum().item()
+            total += features.size(0)
+
+        return total_loss / total, correct / total
+
+    @torch.no_grad()
+    def validation_step(self, loader, criterion, device):
+        self.eval()
+        total_loss = 0.0
+        correct = 0
+        total = 0
+
+        for features, labels in loader:
+            features, labels = features.to(device), labels.to(device)
+            outputs = self(features)
+            loss = criterion(outputs, labels)
+
+            total_loss += loss.item() * features.size(0)
+            correct += (outputs.argmax(dim=1) == labels).sum().item()
+            total += features.size(0)
+
+        return total_loss / total, correct / total
+
+    @torch.no_grad()
+    def test_step(self, loader, criterion, device):
+        self.eval()
+        total_loss = 0.0
+        correct = 0
+        total = 0
+
+        for features, labels in loader:
+            features, labels = features.to(device), labels.to(device)
+            outputs = self(features)
+            loss = criterion(outputs, labels)
+
+            total_loss += loss.item() * features.size(0)
+            correct += (outputs.argmax(dim=1) == labels).sum().item()
+            total += features.size(0)
+
+        return total_loss / total, correct / total
+
+
+def main():
+    set_seed(CFG["seed"])
+
+    device = torch.device("cuda" if torch.cuda.is_available() else "cpu")
+    print(f"[TRAIN] Device: {device}")
+    print(f"[TRAIN] Model: {CFG['model_name']}")
+
+    train_loader, val_loader, test_loader, num_features, num_classes = get_dataloaders(
+        model_name=CFG["model_name"],
+        batch_size=CFG["batch_size"],
+        split_ratios=CFG["split_ratios"],
+        seed=CFG["seed"],
+    )
+
+    model = BaseModel(num_features, num_classes).to(device)
+    criterion = nn.CrossEntropyLoss()
+    optimizer = optim.Adam(model.parameters(), lr=CFG["lr"])
+
+    print(f"[TRAIN] Parameters: {sum(p.numel() for p in model.parameters()):,}")
+
+    ckpt_dir = CFG["checkpoints_dir"][CFG["model_name"]]
+    os.makedirs(ckpt_dir, exist_ok=True)
+    best_ckpt_path = os.path.join(ckpt_dir, "best_model.pt")
+
+    history = {
+        "model_name": CFG["model_name"],
+        "epochs": [],
+        "train_loss": [],
+        "train_acc": [],
+        "val_loss": [],
+        "val_acc": [],
+    }
+
+    best_val_acc = 0.0
+
+    print(f"\n{'Epoch':>6} | {'Train Loss':>10} | {'Train Acc':>9} | {'Val Loss':>10} | {'Val Acc':>9}")
+    print("-" * 60)
+
+    for epoch in range(1, CFG["epochs"] + 1):
+        train_loss, train_acc = model.training_step(train_loader, optimizer, criterion, device)
+        val_loss, val_acc = model.validation_step(val_loader, criterion, device)
+
+        history["epochs"].append(epoch)
+        history["train_loss"].append(round(train_loss, 4))
+        history["train_acc"].append(round(train_acc, 4))
+        history["val_loss"].append(round(val_loss, 4))
+        history["val_acc"].append(round(val_acc, 4))
+
+        marker = ""
+        if val_acc > best_val_acc:
+            best_val_acc = val_acc
+            torch.save(model.state_dict(), best_ckpt_path)
+            marker = " *"
+
+        print(f"{epoch:>6} | {train_loss:>10.4f} | {train_acc:>8.2%} | {val_loss:>10.4f} | {val_acc:>8.2%}{marker}")
+
+    print(f"\nBest validation accuracy: {best_val_acc:.2%}")
+    print(f"Checkpoint saved to: {best_ckpt_path}")
+
+    model.load_state_dict(torch.load(best_ckpt_path, weights_only=True))
+    test_loss, test_acc = model.test_step(test_loader, criterion, device)
+    print(f"\n[TEST] Loss: {test_loss:.4f} | Accuracy: {test_acc:.2%}")
+
+    history["test_loss"] = round(test_loss, 4)
+    history["test_acc"] = round(test_acc, 4)
+
+    logs_dir = CFG["logs_dir"]
+    os.makedirs(logs_dir, exist_ok=True)
+    log_path = os.path.join(logs_dir, f"{CFG['model_name']}_training_log.json")
+
+    with open(log_path, "w") as f:
+        json.dump(history, f, indent=2)
+
+    print(f"[LOG] Training history saved to: {log_path}")
+
+
+if __name__ == "__main__":
+    main()

ui/README.md

@@ -0,0 +1,3 @@
+# ui
+
+Live demo and session view — structure up to the team.