{"id":833,"date":"2026-05-02T07:09:48","date_gmt":"2026-05-01T23:09:48","guid":{"rendered":"https:\/\/connectword.dpdns.org\/?p=833"},"modified":"2026-05-02T07:09:48","modified_gmt":"2026-05-01T23:09:48","slug":"a-coding-implementation-of-end-to-end-brain-decoding-from-meg-signals-using-neuralset-and-deep-learning-for-predicting-linguistic-features","status":"publish","type":"post","link":"https:\/\/connectword.dpdns.org\/?p=833","title":{"rendered":"A Coding Implementation of End-to-End Brain Decoding from MEG Signals Using NeuralSet and Deep Learning for Predicting Linguistic Features"},"content":{"rendered":"

In this tutorial, we explore how we can decode linguistic features directly from brain signals using a modern neuroAI<\/strong><\/a> pipeline. We work with MEG data and build an end-to-end system that transforms raw neural activity into meaningful predictions, in this case, estimating word length from brain responses. We set up the environment, load and process neural events, design a custom feature extractor, and construct a structured data pipeline using NeuralSet. From there, we train a convolutional neural network to learn patterns in the temporal and spatial structure of MEG signals. Throughout the process, we focus on building a clean, modular workflow that mirrors real-world neuroAI research practices.<\/p>\n

\n
\n
\n
\n
<\/div>\n
<\/div>\n
<\/div>\n<\/div>\n
Copy Code<\/span>Copied<\/span>Use a different Browser<\/span><\/a><\/div>\n<\/div>\n
import subprocess, sys, importlib, pkgutil\ndef pip_install(*pkgs):\n   print(f\"pip install {' '.join(pkgs)} ...\")\n   r = subprocess.run([sys.executable, \"-m\", \"pip\", \"install\", \"-q\", *pkgs],\n                      capture_output=True, text=True)\n   if r.returncode != 0:\n       print(\"pip STDOUT:\", r.stdout[-2000:])\n       print(\"pip STDERR:\", r.stderr[-2000:])\n       raise RuntimeError(\"pip install failed; see output above.\")\n   print(\"  ok\")\n\n\npip_install(\"numpy>=2.0,<2.3\")\npip_install(\"neuralset\")\npip_install(\"neuralfetch\")\n\n\nimport numpy as np\nfrom numpy._core.umath import _center\nprint(f\"numpy {np.__version__} OK\")\n\n\nimport warnings, typing as tp\nwarnings.filterwarnings(\"ignore\")\nimport pandas as pd\nimport torch\nimport torch.nn as nn\nimport torch.nn.functional as F\nfrom torch.utils.data import DataLoader\nimport matplotlib.pyplot as plt\n\n\nimport neuralset as ns\nfrom neuralset import extractors as ext_mod<\/code><\/pre>\n<\/div>\n<\/div>\n
We install and validate all required dependencies, ensuring critical packages such as NumPy and NeuralSet are properly configured. We perform a quick NumPy check to avoid runtime issues later in the pipeline. We then import all core libraries needed for data processing, modeling, and visualization.<\/p>\n
\n
\n
\n
\n
<\/div>\n
<\/div>\n
<\/div>\n<\/div>\n
Copy Code<\/span>Copied<\/span>Use a different Browser<\/span><\/a><\/div>\n<\/div>\n
def deep_import(pkg_name: str):\n   try:\n       pkg = importlib.import_module(pkg_name)\n   except Exception as e:\n       print(f\"\"\u26a0\"  could not import {pkg_name}: {e}\")\n       return\n   if not hasattr(pkg, \"__path__\"):\n       return\n   for m in pkgutil.walk_packages(pkg.__path__, prefix=pkg_name + \".\"):\n       try:\n           importlib.import_module(m.name)\n       except Exception:\n           pass\n\n\ndeep_import(\"neuralfetch\")\ndeep_import(\"neuralset\")\n\n\ntorch.manual_seed(0); np.random.seed(0)\n\n\ncatalog = ns.Study.catalog()\nprint(f\"n{len(catalog)} studies registered.\")\npreferred = [\"Fake2025Meg\", \"Test2025Meg\", \"Test2023Meg\"]\nstudy_name = next((n for n in preferred if n in catalog), None)\n\n\nif study_name is None:\n   meg_studies = [n for n, c in catalog.items() if \"Meg\" in c.neuro_types()]\n   study_name = meg_studies[0] if meg_studies else None\n\n\nif study_name is None:\n   raise RuntimeError(\n       \"No MEG study available. Catalog: \"\n       f\"{sorted(catalog.keys())[:20]}\u2026  \"\n       \"Install neuralfetch correctly (pip install neuralfetch) and re-run.\"\n   )\nprint(f\"\u2192 Using study: {study_name}\")<\/code><\/pre>\n<\/div>\n<\/div>\n
We dynamically import all submodules from NeuralFetch and NeuralSet to ensure that all available studies are properly registered. We seed the random number generator for reproducibility and inspect the study catalog to identify available MEG datasets. We then select an appropriate study to use as the foundation for our pipeline.<\/p>\n
\n
\n
\n
\n
<\/div>\n
<\/div>\n
<\/div>\n<\/div>\n
Copy Code<\/span>Copied<\/span>Use a different Browser<\/span><\/a><\/div>\n<\/div>\n
class CharCount(ext_mod.BaseStatic):\n   event_types: tp.Literal[\"Word\"] = \"Word\"\n   def get_static(self, event) -> torch.Tensor:\n       return torch.tensor([float(len(event.text))], dtype=torch.float32)\n\n\nprint(\"nBuilding chain...\")\nchain = ns.Chain(steps=[\n   {\"name\": study_name, \"path\": str(ns.CACHE_FOLDER)},\n   {\"name\": \"QueryEvents\", \"query\": \"type in ['Word', 'Meg']\"},\n])\nevents = chain.run()\nprint(f\"  \u2192 {len(events)} events; types={sorted(events.type.unique().tolist())}\")\nprint(f\"  \u2192 Words: {(events.type=='Word').sum()} | \"\n     f\"timelines: {events.timeline.nunique()}\")\nprint(\"nSample words:\")\nprint(events[events.type=='Word'][[\"start\",\"duration\",\"text\",\"timeline\"]]\n     .head(5).to_string(index=False))\n\n\nprint(\"nBuilding segmenter...\")\nsegmenter = ns.dataloader.Segmenter(\n   extractors={\n       \"meg\":        {\"name\": \"MegExtractor\", \"frequency\": 100.0},\n       \"char_count\": CharCount(aggregation=\"trigger\"),\n   },\n   trigger_query=\"type == 'Word'\",\n   start=-0.2, duration=0.8,\n   drop_incomplete=True,\n)\ndataset = segmenter.apply(events)\nprint(f\"  \u2192 SegmentDataset: {len(dataset)} segments\")\n\n\ns0 = dataset[0]\nprint(f\"nSingle item:n  meg        : {tuple(s0.data['meg'].shape)}\")\nprint(f\"  char_count : {s0.data['char_count'].item()}  \"\n     f\"(word: {s0.segments[0].trigger.text!r})\")<\/code><\/pre>\n<\/div>\n<\/div>\n
We define a custom extractor that computes the character count of each word event, enabling us to create a supervised learning target. We build a processing chain to load and filter relevant events from the selected study. We then segment the MEG signals around word events and construct a dataset ready for modeling.<\/p>\n
\n
\n
\n
\n
<\/div>\n
<\/div>\n
<\/div>\n<\/div>\n
Copy Code<\/span>Copied<\/span>Use a different Browser<\/span><\/a><\/div>\n<\/div>\n
rng  = np.random.RandomState(42)\nperm = rng.permutation(len(dataset))\nn_tr, n_va = int(0.70*len(dataset)), int(0.15*len(dataset))\ntrain_ds = dataset.select(perm[:n_tr])\nval_ds   = dataset.select(perm[n_tr:n_tr+n_va])\ntest_ds  = dataset.select(perm[n_tr+n_va:])\nprint(f\"nSplit | train={len(train_ds)}  val={len(val_ds)}  test={len(test_ds)}\")\n\n\nmk = lambda d, sh: DataLoader(d, batch_size=32, shuffle=sh,\n                             collate_fn=d.collate_fn, drop_last=False)\ntrain_loader, val_loader, test_loader = mk(train_ds, True), mk(val_ds, False), mk(test_ds, False)\n\n\nprobe = next(iter(train_loader))\nn_ch, n_t = probe.data[\"meg\"].shape[-2:]\nprint(f\"  \u2192 batch[meg]  shape: {tuple(probe.data['meg'].shape)}\")\nprint(f\"  \u2192 batch[char] shape: {tuple(probe.data['char_count'].shape)}\")\n\n\nclass MEGDecoder(nn.Module):\n   def __init__(self, n_channels: int, mid: int = 64):\n       super().__init__()\n       self.spatial   = nn.Conv1d(n_channels, mid, 1)\n       self.bn0       = nn.BatchNorm1d(mid)\n       self.temporal1 = nn.Conv1d(mid, mid, 7, padding=3)\n       self.bn1       = nn.BatchNorm1d(mid)\n       self.temporal2 = nn.Conv1d(mid, mid\/\/2, 7, padding=3)\n       self.bn2       = nn.BatchNorm1d(mid\/\/2)\n       self.pool      = nn.AdaptiveAvgPool1d(1)\n       self.head      = nn.Linear(mid\/\/2, 1)\n       self.drop      = nn.Dropout(0.3)\n   def forward(self, x):\n       x = F.gelu(self.bn0(self.spatial(x)))\n       x = F.gelu(self.bn1(self.temporal1(x)))\n       x = self.drop(x)\n       x = F.gelu(self.bn2(self.temporal2(x)))\n       return self.head(self.pool(x).squeeze(-1)).squeeze(-1)\n\n\ndevice = torch.device(\"cuda\" if torch.cuda.is_available() else \"cpu\")\nmodel  = MEGDecoder(n_channels=n_ch).to(device)\nprint(f\"nDevice: {device} | params: {sum(p.numel() for p in model.parameters()):,}\")\n\n\ntrain_targets = torch.cat([b.data[\"char_count\"].squeeze(-1) for b in train_loader])\ny_mean, y_std = train_targets.mean().item(), train_targets.std().item() + 1e-6\nprint(f\"Target  \u03bc={y_mean:.2f}  \u03c3={y_std:.2f}\")\n\n\ndef prep(batch):\n   x = batch.data[\"meg\"].to(device).float()\n   y = batch.data[\"char_count\"].squeeze(-1).to(device).float()\n   x = (x - x.mean(-1, keepdim=True)) \/ (x.std(-1, keepdim=True) + 1e-6)\n   y = (y - y_mean) \/ y_std\n   return x, y<\/code><\/pre>\n<\/div>\n<\/div>\n
We split the dataset into training, validation, and test sets to ensure proper model evaluation. We create data loaders and inspect batch shapes to confirm correct data formatting. We then define a convolutional neural network and prepare normalized inputs and targets for stable training.<\/p>\n
\n
\n
\n
\n
<\/div>\n
<\/div>\n
<\/div>\n<\/div>\n
Copy Code<\/span>Copied<\/span>Use a different Browser<\/span><\/a><\/div>\n<\/div>\n
EPOCHS  = 15\nopt     = torch.optim.AdamW(model.parameters(), lr=1e-3, weight_decay=1e-4)\nsched   = torch.optim.lr_scheduler.CosineAnnealingLR(opt, T_max=EPOCHS)\nloss_fn = nn.MSELoss()\nhist    = {\"tr\": [], \"va\": [], \"r\": []}\n\n\ndef pearson(a, b):\n   a, b = a - a.mean(), b - b.mean()\n   return (a*b).sum() \/ (a.norm()*b.norm() + 1e-8)\n\n\nprint(\"n\" + \"=\"*64)\nprint(f\"{'Epoch':>5} | {'train':>9} | {'val':>9} | {'val_r':>7}\")\nprint(\"=\"*64)\nfor ep in range(EPOCHS):\n   model.train(); tr = []\n   for batch in train_loader:\n       x, y = prep(batch)\n       loss = loss_fn(model(x), y)\n       opt.zero_grad(); loss.backward()\n       torch.nn.utils.clip_grad_norm_(model.parameters(), 1.0)\n       opt.step(); tr.append(loss.item())\n   sched.step()\n\n\n   model.eval(); va, P, T = [], [], []\n   with torch.no_grad():\n       for batch in val_loader:\n           x, y = prep(batch); p = model(x)\n           va.append(loss_fn(p, y).item()); P.append(p.cpu()); T.append(y.cpu())\n   P, T = torch.cat(P), torch.cat(T)\n   r = pearson(P, T).item()\n   hist[\"tr\"].append(np.mean(tr)); hist[\"va\"].append(np.mean(va)); hist[\"r\"].append(r)\n   print(f\"{ep+1:>5d} | {np.mean(tr):>9.4f} | {np.mean(va):>9.4f} | {r:>+7.3f}\")\n\n\nmodel.eval(); P, T = [], []\nwith torch.no_grad():\n   for batch in test_loader:\n       x, y = prep(batch)\n       P.append(model(x).cpu()); T.append(y.cpu())\nP, T = torch.cat(P), torch.cat(T)\ntest_r   = pearson(P, T).item()\ntest_mse = ((P - T) ** 2).mean().item()\nprint(f\"nTEST  |  Pearson r = {test_r:+.3f}   MSE = {test_mse:.3f}\")\nprint(f\"(Synthetic-MEG signals are random by design \u2014 small\/zero r is expected.)\")\n\n\nfig, ax = plt.subplots(1, 3, figsize=(15, 4))\nax[0].plot(hist[\"tr\"], label=\"train\"); ax[0].plot(hist[\"va\"], label=\"val\")\nax[0].set(xlabel=\"Epoch\", ylabel=\"MSE\", title=\"Loss curves\"); ax[0].legend(); ax[0].grid(alpha=.3)\nax[1].plot(hist[\"r\"], color=\"C2\"); ax[1].axhline(0, color=\"k\", ls=\"--\", alpha=.4)\nax[1].set(xlabel=\"Epoch\", ylabel=\"Pearson r\", title=\"Validation correlation\"); ax[1].grid(alpha=.3)\nm = float(max(T.abs().max(), P.abs().max()))\nax[2].scatter(T.numpy(), P.numpy(), s=10, alpha=.35)\nax[2].plot([-m, m], [-m, m], \"k--\", alpha=.4)\nax[2].set(xlabel=\"True (z-scored char count)\", ylabel=\"Predicted\",\n         title=f\"Test predictions (r = {test_r:+.3f})\"); ax[2].grid(alpha=.3)\nplt.tight_layout(); plt.show()\n\n\nprint(\"n\"\u2705\" Tutorial complete!\")\nprint(f\"  \u2022 Study used        : {study_name}\")\nprint(f\"  \u2022 Pipeline          : Chain \u2192 Segmenter \u2192 SegmentDataset \u2192 DataLoader\")\nprint(f\"  \u2022 Custom extractor  : CharCount (subclass of BaseStatic)\")\nprint(f\"  \u2022 Built-in extractor: MegExtractor @ 100 Hz\")\nprint(f\"  \u2022 Model             : 1\u00d71 spatial conv + 2 temporal convs + linear head\")<\/code><\/pre>\n<\/div>\n<\/div>\n
We train the neural network using a structured training loop with loss tracking and learning rate scheduling. We evaluate the model on the validation and test sets using metrics such as MSE and Pearson\u2019s correlation. Also, we visualize training performance and predictions to understand how well the model learns from the data.<\/p>\n
In conclusion, we demonstrated how we can bridge neural data and language understanding using deep learning. We implemented a full pipeline, from raw event extraction to model training and evaluation, while maintaining flexibility through reusable components like chains, segmenters, and extractors. Although we worked with synthetic MEG signals, the framework we built is directly applicable to real-world datasets and more complex decoding tasks. This exercise highlights how we can combine neuroscience, machine learning, and structured pipelines to advance interpretable brain decoding systems, laying a strong foundation for more advanced neuroAI applications.<\/p>\n
\n
Check out\u00a0the\u00a0Full Codes and Notebook here<\/a><\/strong>.\u00a0<\/strong>Also,\u00a0feel free to follow us on\u00a0Twitter<\/mark><\/a><\/strong>\u00a0and don\u2019t forget to join our\u00a0130k+ ML SubReddit<\/a><\/strong>\u00a0and Subscribe to\u00a0our Newsletter<\/a><\/strong>. Wait! are you on telegram?\u00a0now you can join us on telegram as well.<\/a><\/strong><\/p>\n
Need to partner with us for promoting your GitHub Repo OR Hugging Face Page OR Product Release OR Webinar etc.?\u00a0Connect with us<\/mark><\/a><\/strong><\/p>\n
The post A Coding Implementation of End-to-End Brain Decoding from MEG Signals Using NeuralSet and Deep Learning for Predicting Linguistic Features<\/a> appeared first on MarkTechPost<\/a>.<\/p>","protected":false},"excerpt":{"rendered":"
In this tutorial, we explore h…<\/p>\n","protected":false},"author":1,"featured_media":29,"comment_status":"open","ping_status":"open","sticky":false,"template":"","format":"standard","meta":{"footnotes":""},"categories":[1],"tags":[],"class_list":["post-833","post","type-post","status-publish","format-standard","has-post-thumbnail","hentry","category-uncategorized"],"_links":{"self":[{"href":"https:\/\/connectword.dpdns.org\/index.php?rest_route=\/wp\/v2\/posts\/833","targetHints":{"allow":["GET"]}}],"collection":[{"href":"https:\/\/connectword.dpdns.org\/index.php?rest_route=\/wp\/v2\/posts"}],"about":[{"href":"https:\/\/connectword.dpdns.org\/index.php?rest_route=\/wp\/v2\/types\/post"}],"author":[{"embeddable":true,"href":"https:\/\/connectword.dpdns.org\/index.php?rest_route=\/wp\/v2\/users\/1"}],"replies":[{"embeddable":true,"href":"https:\/\/connectword.dpdns.org\/index.php?rest_route=%2Fwp%2Fv2%2Fcomments&post=833"}],"version-history":[{"count":0,"href":"https:\/\/connectword.dpdns.org\/index.php?rest_route=\/wp\/v2\/posts\/833\/revisions"}],"wp:featuredmedia":[{"embeddable":true,"href":"https:\/\/connectword.dpdns.org\/index.php?rest_route=\/wp\/v2\/media\/29"}],"wp:attachment":[{"href":"https:\/\/connectword.dpdns.org\/index.php?rest_route=%2Fwp%2Fv2%2Fmedia&parent=833"}],"wp:term":[{"taxonomy":"category","embeddable":true,"href":"https:\/\/connectword.dpdns.org\/index.php?rest_route=%2Fwp%2Fv2%2Fcategories&post=833"},{"taxonomy":"post_tag","embeddable":true,"href":"https:\/\/connectword.dpdns.org\/index.php?rest_route=%2Fwp%2Fv2%2Ftags&post=833"}],"curies":[{"name":"wp","href":"https:\/\/api.w.org\/{rel}","templated":true}]}}