feat: initial CORNET framework implementation

Author: srikrishna3118

.github/workflows/test.yml

@@ -0,0 +1,24 @@
+name: CI
+
+on:
+  push:
+    branches: [main]
+  pull_request:
+    branches: [main]
+
+jobs:
+  test:
+    runs-on: ubuntu-latest
+    steps:
+      - uses: actions/checkout@v4
+
+      - name: Set up Python
+        uses: actions/setup-python@v5
+        with:
+          python-version: "3.10"
+
+      - name: Install package
+        run: pip install -e .[dev]
+
+      - name: Run tests
+        run: pytest tests/ -v --tb=short

.gitignore

@@ -0,0 +1,14 @@
+__pycache__/
+*.py[cod]
+*.egg-info/
+dist/
+build/
+.venv/
+venv/
+*.egg
+.pytest_cache/
+.coverage
+htmlcov/
+tasks/*/generated_launch_*.py
+tasks/*/leaderboard.json.bak.*
+tasks/*/results/

LICENSE

@@ -0,0 +1,21 @@
+MIT License
+
+Copyright (c) 2025 RBCCPS, Indian Institute of Science
+
+Permission is hereby granted, free of charge, to any person obtaining a copy
+of this software and associated documentation files (the "Software"), to deal
+in the Software without restriction, including without limitation the rights
+to use, copy, modify, merge, publish, distribute, sublicense, and/or sell
+copies of the Software, and to permit persons to whom the Software is
+furnished to do so, subject to the following conditions:
+
+The above copyright notice and this permission notice shall be included in all
+copies or substantial portions of the Software.
+
+THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
+IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
+FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE
+AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
+LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,
+OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE
+SOFTWARE.

README.md

@@ -0,0 +1,44 @@
+# CORNET — Generic Co-Simulation Framework for Networked Robots
+
+[![CI](https://github.com/rbccps-iisc/CORNET/actions/workflows/test.yml/badge.svg)](https://github.com/rbccps-iisc/CORNET/actions)
+
+**CORNET** is a task-driven, plugin-based co-simulation framework that lets you run network-robotics experiments with a single config file and one command:
+
+```bash
+python -m cornet tasks/pendulum_nr_control
+```
+
+## Features
+
+- **Unified config schema** — one YAML covers network (NS-3 5G NR or Mininet-WiFi) + robot (Gazebo + ROS 2)
+- **Plugin architecture** — swap network or robot backends without touching orchestration code
+- **Task-folder convention** — each experiment is fully self-contained under `tasks/<name>/`
+- **Parameter sweep** — declare axes in config; framework runs all combinations automatically
+- **EvalTool interface** — standardized metric extraction per task
+- **Experiment leaderboard** — append-only JSON + `rich` terminal viewer across runs
+
+## Quickstart
+
+```bash
+pip install cornet-framework
+# System prerequisites: NS-3 (with 5G NR), Mininet-WiFi, Gazebo Classic 11, ROS 2 Humble
+python -m cornet tasks/pendulum_nr_control
+python -m cornet view tasks/pendulum_nr_control
+```
+
+See [docs/INSTALL.md](docs/INSTALL.md) for system prerequisites and [docs/GETTING_STARTED.md](docs/GETTING_STARTED.md) for a walkthrough.
+
+## CORNET Family
+
+| Version | Repo | Publication |
+|---------|------|-------------|
+| 1.0 | [srikrishna3118/CORNET](https://github.com/srikrishna3118/CORNET) | COMSNETS 2020 |
+| 2.0 | [rbccps-iisc/CORNET2.0](https://github.com/rbccps-iisc/CORNET2.0) | arXiv:2109.06979, COMSNETS 2022 |
+| 3.0 | [rbccps-iisc/CORNET3.0](https://github.com/rbccps-iisc/CORNET3.0) | PhD thesis (IISc, 2025) |
+| Flagship | **this repo** | — |
+
+See [docs/LINEAGE.md](docs/LINEAGE.md) for the full family history and BibTeX citations.
+
+## License
+
+MIT — see [LICENSE](LICENSE).

cornet/__init__.py

@@ -0,0 +1,3 @@
+"""CORNET: Generic Co-Simulation Framework for Networked Robots."""
+
+__version__ = "0.1.0"

cornet/__main__.py

@@ -0,0 +1,54 @@
+"""Entry point: python -m cornet tasks/<name>  or  python -m cornet view tasks/<name>"""
+
+import argparse
+import sys
+
+
+def main() -> None:
+    parser = argparse.ArgumentParser(
+        prog="cornet",
+        description="CORNET co-simulation framework",
+    )
+    subparsers = parser.add_subparsers(dest="command")
+
+    # run subcommand (default when a path is given directly)
+    run_parser = subparsers.add_parser("run", help="Run an experiment task")
+    run_parser.add_argument("task", help="Path to task directory or config.yaml")
+
+    # view subcommand
+    view_parser = subparsers.add_parser("view", help="View experiment leaderboard")
+    view_parser.add_argument("task", help="Path to task directory")
+    view_parser.add_argument(
+        "--higher-is-better",
+        action="store_true",
+        default=False,
+        help="Sort leaderboard descending (higher metric = better)",
+    )
+
+    # Allow bare positional: `python -m cornet tasks/foo` treated as run
+    parser.add_argument("_task_positional", nargs="?", help=argparse.SUPPRESS)
+
+    args = parser.parse_args()
+
+    if args.command == "view":
+        from cornet.leaderboard.viewer import show
+        show(args.task, higher_is_better=args.higher_is_better)
+
+    elif args.command == "run":
+        from cornet.orchestrator import Orchestrator
+        orch = Orchestrator()
+        orch.run(task_dir=args.task)
+
+    elif args._task_positional:
+        # bare positional: treat as run
+        from cornet.orchestrator import Orchestrator
+        orch = Orchestrator()
+        orch.run(task_dir=args._task_positional)
+
+    else:
+        parser.print_help()
+        sys.exit(1)
+
+
+if __name__ == "__main__":
+    main()

cornet/config/__init__.py

@@ -0,0 +1 @@
+# Config sub-package

cornet/config/loader.py

@@ -0,0 +1,57 @@
+"""Config loader for CORNET unified schema.
+
+Usage:
+    from cornet.config.loader import load_unified
+    cfg = load_unified("tasks/my_task/config.yaml")
+"""
+
+from __future__ import annotations
+
+from pathlib import Path
+
+import yaml
+
+from cornet.config.schema import ConfigValidationError, UnifiedConfig
+
+
+def load_unified(path: str | Path) -> UnifiedConfig:
+    """Load and validate a unified-v1 YAML config file.
+
+    Raises:
+        ConfigValidationError: if the file is not unified-v1 or fails validation.
+        FileNotFoundError: if *path* does not exist.
+    """
+    p = Path(path)
+    if not p.exists():
+        raise FileNotFoundError(f"Config not found: {p}")
+
+    with p.open() as fh:
+        raw = yaml.safe_load(fh)
+
+    if not isinstance(raw, dict):
+        raise ConfigValidationError(f"Config must be a YAML mapping, got {type(raw).__name__}")
+
+    schema_tag = raw.get("_schema", "")
+    if schema_tag != "unified-v1":
+        raise ConfigValidationError(
+            f"Expected '_schema: unified-v1', got '{schema_tag}'. "
+            "This loader only handles unified configs."
+        )
+
+    try:
+        return UnifiedConfig.model_validate(raw)
+    except Exception as exc:
+        raise ConfigValidationError(str(exc)) from exc
+
+
+def is_unified(path: str | Path) -> bool:
+    """Return True if the YAML file at *path* declares ``_schema: unified-v1``."""
+    p = Path(path)
+    if not p.exists():
+        return False
+    try:
+        with p.open() as fh:
+            raw = yaml.safe_load(fh)
+        return isinstance(raw, dict) and raw.get("_schema") == "unified-v1"
+    except Exception:
+        return False

cornet/config/schema.py

@@ -0,0 +1,116 @@
+"""Pydantic v2 unified config schema for CORNET experiments."""
+
+from __future__ import annotations
+
+from pathlib import Path
+from typing import Any, Literal
+
+from pydantic import BaseModel, ConfigDict, field_validator, model_validator
+
+
+class ConfigValidationError(ValueError):
+    """Raised when a unified config fails schema validation."""
+
+
+# ── Node models ──────────────────────────────────────────────────────────────
+
+class ContainerConfig(BaseModel):
+    image: str
+    environment: dict[str, str] = {}
+    volumes: list[str] = []
+
+
+class NodeConfig(BaseModel):
+    name: str
+    type: Literal["MOBILE", "STATIC", "UE", "GNB", "EPC"] = "MOBILE"
+    container: ContainerConfig | None = None
+    # Arbitrary extra keys (e.g. position, speed) are allowed
+    model_config = ConfigDict(extra="allow")
+
+
+# ── Network section ───────────────────────────────────────────────────────────
+
+class MininetConfig(BaseModel):
+    wmediumd: bool = False
+    ssid: str = "cornet-net"
+    mode: str = "g"
+    channel: int = 1
+
+
+class NetworkConfig(BaseModel):
+    plugin: str
+    type: Literal["ns3", "mininet", "ns3+mininet"]
+    nodes: list[NodeConfig] = []
+    mininet: MininetConfig | None = None
+    # NS-3 specific keys forwarded as-is
+    model_config = ConfigDict(extra="allow")
+
+
+# ── Robot section ─────────────────────────────────────────────────────────────
+
+class ModelConfig(BaseModel):
+    type: Literal["urdf", "sdf"]
+    path: str  # relative to task dir or absolute
+
+
+class PoseConfig(BaseModel):
+    x: float = 0.0
+    y: float = 0.0
+    z: float = 0.0
+    roll: float = 0.0
+    pitch: float = 0.0
+    yaw: float = 0.0
+
+
+class RobotEntry(BaseModel):
+    name: str
+    model: ModelConfig
+    pose: PoseConfig = PoseConfig()
+    ros_namespace: str | None = None
+
+
+class RobotConfig(BaseModel):
+    plugin: str
+    launch_file: str | None = None
+    world: str | None = None
+    robots: list[RobotEntry] = []
+
+
+# ── Experiment section ────────────────────────────────────────────────────────
+
+class SweepConfig(BaseModel):
+    axes: dict[str, list[Any]]       # e.g. {"network.numerology": [2,4], "network.bandwidth": [20,40]}
+    parallel: bool = False
+    repeats: int = 1
+
+    @field_validator("repeats")
+    @classmethod
+    def _positive_repeats(cls, v: int) -> int:
+        if v < 1:
+            raise ConfigValidationError("sweep.repeats must be >= 1")
+        return v
+
+
+class ExperimentConfig(BaseModel):
+    name: str
+    duration: float                  # seconds
+    output_dir: str = "results"
+    primary_metric: str | None = None
+    higher_is_better: bool = False
+    sweep: SweepConfig | None = None
+
+
+# ── Top-level ─────────────────────────────────────────────────────────────────
+
+class UnifiedConfig(BaseModel):
+    """Root config model for CORNET unified schema (``_schema: unified-v1``)."""
+
+    _schema: str = "unified-v1"      # literal marker; not validated by pydantic
+    network: NetworkConfig
+    robot: RobotConfig
+    experiment: ExperimentConfig
+
+    @model_validator(mode="after")
+    def _cross_validate(self) -> "UnifiedConfig":
+        # Mininet wmediumd requires Docker support — warn but don't error
+        return self

cornet/context.py

@@ -0,0 +1,32 @@
+"""ExperimentContext — shared state passed between plugins during a run."""
+
+from __future__ import annotations
+
+from dataclasses import dataclass, field
+
+
+@dataclass
+class NetworkContext:
+    """State populated by the network plugin after start()."""
+
+    # Maps node/UE name → assigned IP address (set by network plugin after start)
+    node_ips: dict[str, str] = field(default_factory=dict)
+
+
+@dataclass
+class RobotContext:
+    """State populated by the robot plugin after start()."""
+
+    # Maps robot name → spawned namespace (set by robot plugin after start)
+    robot_namespaces: dict[str, str] = field(default_factory=dict)
+
+
+@dataclass
+class ExperimentContext:
+    """Top-level context object shared across all plugins in a single run."""
+
+    network: NetworkContext = field(default_factory=NetworkContext)
+    robot: RobotContext = field(default_factory=RobotContext)
+
+    # Variant ID for sweep runs (e.g. "numerology=2_bandwidth=40"); "default" otherwise
+    variant_id: str = "default"