Initialize FPS toolkit structure

.github/workflows/main.yml

@@ -0,0 +1,40 @@
+# .github/workflows/auto-rebase-and-merge.yml
+name: Auto-rebase-then-merge
+
+on:
+  pull_request_target:
+    types: [opened, synchronize, ready_for_review, reopened]
+
+permissions:
+  contents: write          # needed for pushing & merging
+  pull-requests: write
+
+jobs:
+
+  rebase:
+    if: github.event.pull_request.draft == false
+    runs-on: ubuntu-latest
+    steps:
+      - uses: actions/checkout@v4
+        with:
+          ref: ${{ github.head_ref }}
+          fetch-depth: 0
+      - run: |
+          git config user.name  "merge-bot"
+          git config user.email "bot@example.com"
+          git fetch origin ${{ github.base_ref }}
+          git merge -s ort -X theirs origin/${{ github.base_ref }} \
+            --commit -m "Auto-merge main into PR"
+          git push origin HEAD
+
+  merge:
+    needs: rebase
+    if: ${{ needs.rebase.result == 'success' }}    # CI steps could live here
+    runs-on: ubuntu-latest
+    steps:
+      - name: Squash-merge the PR
+        uses: peter-evans/merge-pull-request@v3
+        with:
+          token: ${{ secrets.GITHUB_TOKEN }}
+          merge_method: squash      # or merge / rebase
+          commit_title: "♻️ Auto-merge PR #${{ github.event.pull_request.number }}"

README.md

@@ -1,2 +1,40 @@
-# fps-simulation-validation
-A minimal‑yet‑rigorous code scaffold for reproducing, stress‑testing, and visualising the Fractale Pulsante Spiralée (FPS) model described in the paper and roadmap. Designed for fast iteration, statistical reproducibility, and transparent peer review.
+# FPS Simulation & Validation Toolkit (v0.1)
+
+A minimal-yet-rigorous code scaffold for reproducing, stress-testing, and visualising the Fractale Pulsante Spiralée (FPS) model. Designed for fast iteration, statistical reproducibility, and transparent peer review.
+
+## Quick start
+
+1. **Install dependencies**
+
+   ```bash
+   pip install -e .
+   ```
+
+2. **Run a baseline simulation**
+
+   ```bash
+   python -m fps.simulate --config config/default.yaml
+   ```
+
+   Results are written to `data/logs/`. Adjust parameters in
+   `config/default.yaml` to experiment with different settings.
+
+3. **Launch the web server** (requires `uvicorn`)
+
+   ```bash
+   uvicorn fps.server:app --reload
+   ```
+
+   Trigger a run via
+
+   ```bash
+   curl -X POST http://127.0.0.1:8000/run
+   ```
+
+   View the coherence plot at
+
+   ```bash
+   http://127.0.0.1:8000/plot/<run_name>
+   ```
+
+   Replace `<run_name>` with the `run_name` specified in the config.

config/default.yaml

@@ -0,0 +1,21 @@
+run_name: "baseline_N5"
+seed: 42
+T: 20.0
+Δt: 0.01
+n_strata: 5
+strata_defaults:
+  A0: 1.0
+  f0: 1.0
+  φ0: 0.0
+  γ0: 1.0
+  α: 0.1
+  β: 0.05
+  λ: 0.01
+feedback:
+  G: "tanh"
+noise:
+  type: "uniform"
+  scale: 0.1
+logging:
+  every_step: 10
+  format: "csv"

notebooks/exploratory.ipynb

@@ -0,0 +1 @@
+{}

pyproject.toml

@@ -0,0 +1,22 @@
+[build-system]
+requires = ["setuptools>=42"]
+build-backend = "setuptools.build_meta"
+
+[project]
+name = "fps-toolkit"
+version = "0.1.0"
+requires-python = ">=3.10"
+dependencies = [
+    "numpy",
+    "scipy",
+    "pydantic",
+    "matplotlib",
+    "fastapi",
+    "uvicorn",
+]
+
+[project.optional-dependencies]
+dev = ["pytest"]
+
+[project.scripts]
+fps-simulate = "fps.simulate:main"

src/fps/__init__.py

@@ -0,0 +1,7 @@
+"""FPS Simulation Toolkit package."""
+
+from .simulate import FPSSimulation
+from .parameters import RunConfig
+from .server import app as server_app
+
+__all__ = ["FPSSimulation", "RunConfig", "server_app"]

src/fps/metrics.py

@@ -0,0 +1,13 @@
+import numpy as np
+from .strata import Stratum
+
+
+def coherence(strata: list[Stratum]) -> float:
+    phases = np.array([s.φ for s in strata])
+    mean_phase = np.angle(np.mean(np.exp(1j * phases)))
+    return float(np.mean(np.cos(phases - mean_phase)))
+
+
+def effort(strata: list[Stratum]) -> float:
+    # simple placeholder effort as sum of amplitudes
+    return float(sum(abs(s.A) for s in strata))

src/fps/parameters.py

@@ -0,0 +1,45 @@
+from __future__ import annotations
+from pydantic import BaseModel, Field, validator
+from pathlib import Path
+from typing import Literal
+
+class FeedbackConfig(BaseModel):
+    G: Literal["tanh", "damped_sine", "sinc", "custom"] = "tanh"
+
+class NoiseConfig(BaseModel):
+    type: Literal["uniform", "gaussian", "real_signal"] = "uniform"
+    scale: float = Field(0.0, ge=0)
+
+class StrataDefaults(BaseModel):
+    A0: float = 1.0
+    f0: float = 1.0
+    φ0: float = 0.0
+    γ0: float = 1.0
+    α: float = 0.1
+    β: float = 0.05
+    λ: float = 0.01
+
+class RunConfig(BaseModel):
+    run_name: str = "baseline"
+    seed: int | None = None
+    T: float = Field(..., gt=0)
+    Δt: float = Field(..., gt=0)
+    n_strata: int = Field(..., ge=1)
+    strata_defaults: StrataDefaults = StrataDefaults()
+    feedback: FeedbackConfig = FeedbackConfig()
+    noise: NoiseConfig = NoiseConfig()
+    logging_every_step: int = 10
+    log_format: Literal["csv", "hdf5"] = "csv"
+
+    @validator("logging_every_step")
+    def nonzero(cls, v):
+        if v < 1:
+            raise ValueError("logging_every_step must be ≥1")
+        return v
+
+    def write_seed(self, path: Path = Path("data/seeds.txt")) -> None:
+        if self.seed is None:
+            return
+        path.parent.mkdir(parents=True, exist_ok=True)
+        with path.open("a") as f:
+            f.write(f"{self.run_name},{self.seed}\n")

src/fps/server.py

@@ -0,0 +1,58 @@
+from fastapi import FastAPI, Response
+from pathlib import Path
+from io import BytesIO
+import csv
+import yaml
+import matplotlib.pyplot as plt
+
+from .simulate import FPSSimulation
+from .parameters import RunConfig
+
+app = FastAPI()
+
+
+def _load_cfg(cfg_path: str) -> RunConfig:
+    with open(cfg_path, "r") as f:
+        cfg_dict = yaml.safe_load(f)
+    logging_cfg = cfg_dict.get("logging", {})
+    return RunConfig(
+        run_name=cfg_dict.get("run_name", "run"),
+        seed=cfg_dict.get("seed"),
+        T=cfg_dict.get("T", 1.0),
+        Δt=cfg_dict.get("Δt", 0.01),
+        n_strata=cfg_dict.get("n_strata", 1),
+        strata_defaults=cfg_dict.get("strata_defaults", {}),
+        feedback=cfg_dict.get("feedback", {}),
+        noise=cfg_dict.get("noise", {}),
+        logging_every_step=logging_cfg.get("every_step", 10),
+        log_format=logging_cfg.get("format", "csv"),
+    )
+
+
+@app.post("/run")
+def run_simulation(cfg_path: str = "config/default.yaml"):
+    cfg = _load_cfg(cfg_path)
+    sim = FPSSimulation(cfg)
+    sim.run()
+    return {"status": "ok", "run_name": cfg.run_name}
+
+
+@app.get("/plot/{run_name}")
+def plot(run_name: str):
+    log_path = Path("data/logs") / f"{run_name}.csv"
+    if not log_path.exists():
+        return {"error": "log not found"}
+    t, C = [], []
+    with log_path.open() as f:
+        r = csv.DictReader(f)
+        for row in r:
+            t.append(float(row["t"]))
+            C.append(float(row["C"]))
+    fig, ax = plt.subplots()
+    ax.plot(t, C)
+    ax.set_xlabel("t")
+    ax.set_ylabel("coherence")
+    buf = BytesIO()
+    fig.savefig(buf, format="png")
+    plt.close(fig)
+    return Response(buf.getvalue(), media_type="image/png")

src/fps/simulate.py

@@ -0,0 +1,97 @@
+import numpy as np
+import time
+import csv
+from pathlib import Path
+from .parameters import RunConfig
+from .strata import Stratum
+from .spiral import G_factory
+from .metrics import coherence, effort
+
+
+class FPSSimulation:
+    def __init__(self, cfg: RunConfig):
+        self.cfg = cfg
+        self.G = G_factory(cfg.feedback.G)
+        rng = np.random.default_rng(cfg.seed)
+        self.noise = rng
+        self.strata = [
+            Stratum(
+                cfg.strata_defaults.A0,
+                cfg.strata_defaults.f0,
+                cfg.strata_defaults.φ0,
+                cfg.strata_defaults.γ0,
+                cfg.strata_defaults.α,
+                cfg.strata_defaults.β,
+                cfg.strata_defaults.λ,
+            )
+            for _ in range(cfg.n_strata)
+        ]
+        self.steps = int(cfg.T / cfg.Δt)
+        self._prepare_logging()
+        cfg.write_seed()
+
+    def _prepare_logging(self) -> None:
+        self.rows: list[tuple[float, float, float, float]] = []
+        self.t0 = time.perf_counter()
+
+    def run(self) -> None:
+        for k in range(self.steps):
+            t = k * self.cfg.Δt
+            for s in self.strata:
+                I = self.noise.uniform(-1, 1) * self.cfg.noise.scale
+                feedback = self.G(I)
+                s.update(I, feedback, self.cfg.Δt)
+            if k % self.cfg.logging_every_step == 0:
+                self._log_step(t)
+        self._dump()
+
+    def _log_step(self, t: float) -> None:
+        C = coherence(self.strata)
+        E = effort(self.strata)
+        cpu = (time.perf_counter() - self.t0) / (len(self.rows) + 1)
+        self.rows.append((t, C, E, cpu))
+
+    def _dump(self) -> None:
+        out = Path("data/logs")
+        out.mkdir(parents=True, exist_ok=True)
+        fname = out / f"{self.cfg.run_name}.{self.cfg.log_format}"
+        if self.cfg.log_format == "csv":
+            with fname.open("w", newline="") as f:
+                w = csv.writer(f)
+                w.writerow(["t", "C", "effort", "cpu_step"])
+                w.writerows(self.rows)
+        else:
+            import h5py
+
+            with h5py.File(fname, "w") as h5:
+                h5.create_dataset("dataset", data=np.array(self.rows))
+
+
+def main(argv=None) -> None:
+    import argparse, yaml
+
+    parser = argparse.ArgumentParser(description="Run FPS simulation")
+    parser.add_argument("--config", default="config/default.yaml")
+    args = parser.parse_args(argv)
+
+    with open(args.config, "r") as f:
+        cfg_dict = yaml.safe_load(f)
+    logging_cfg = cfg_dict.get("logging", {})
+    cfg = RunConfig(
+        run_name=cfg_dict.get("run_name", "run"),
+        seed=cfg_dict.get("seed"),
+        T=cfg_dict.get("T", 1.0),
+        Δt=cfg_dict.get("Δt", 0.01),
+        n_strata=cfg_dict.get("n_strata", 1),
+        strata_defaults=cfg_dict.get("strata_defaults", {}),
+        feedback=cfg_dict.get("feedback", {}),
+        noise=cfg_dict.get("noise", {}),
+        logging_every_step=logging_cfg.get("every_step", 10),
+        log_format=logging_cfg.get("format", "csv"),
+    )
+    sim = FPSSimulation(cfg)
+    sim.run()
+
+
+if __name__ == "__main__":
+    main()

src/fps/spiral.py

@@ -0,0 +1,12 @@
+import numpy as np
+from typing import Callable
+
+
+def G_factory(kind: str) -> Callable[[float], float]:
+    if kind == "tanh":
+        return lambda x: np.tanh(x)
+    if kind == "damped_sine":
+        return lambda x: np.exp(-np.abs(x)) * np.sin(x)
+    if kind == "sinc":
+        return lambda x: np.sinc(x / np.pi)
+    raise ValueError(f"Unknown G kind: {kind}")

src/fps/strata.py

@@ -0,0 +1,19 @@
+import numpy as np
+from dataclasses import dataclass
+
+@dataclass
+class Stratum:
+    A: float
+    f: float
+    φ: float
+    γ: float
+    α: float
+    β: float
+    λ: float
+
+    def update(self, I: float, feedback: float, Δt: float) -> None:
+        """Update state according to Eq. (1) discretisation."""
+        I_filt = 1 / (1 + np.exp(-I))
+        self.A += self.α * I_filt - self.β * feedback
+        self.f += self.λ * feedback
+        self.φ = (self.φ + 2 * np.pi * self.f * Δt) % (2 * np.pi)