Modify Data and Dataset, add barebone Node, structure config.ini

config.ini

 [GRAPH]
-PACKAGE = decentralizepy.graphs.SmallWorld
-CLASS = decentralizepy
+package = decentralizepy.graphs.SmallWorld
+graph_class = SmallWorld
 
 [DATASET]
-PACKAGE = decentralizepy.datasets.Femnist
-CLASS = Femnist
-MODEL = LogisticRegression
+dataset_package = decentralizepy.datasets.Femnist
+dataset_class = Femnist
+model_class = LogisticRegression
+n_procs = 1.0
+train_dir = 
+test_dir = 
+; python list of fractions below
+sizes = [0.4, 0.2, 0.3, 0.1]
 
-[MODEL_PARAMS]
-USERNAME = xiaoxu
-PASSWORD = xiaoxu
-HOST = 127.0.0.1
-PORT = 5432
-DB = xiaoxu_database
\ No newline at end of file
+[OPTIMIZER_PARAMS]
+optimizer_package = torch.optim
+optimizer_class = SGD
+lr = 0.1
+
+[TRAIN_PARAMS]
+epochs_per_round = 25
+batch_size = 64

main.ipynb

@@ -254,37 +254,110 @@
   },
   {
    "cell_type": "code",
-   "execution_count": 5,
+   "execution_count": 17,
    "metadata": {},
    "outputs": [
     {
      "name": "stdout",
      "output_type": "stream",
      "text": [
+      "Section:  GRAPH\n",
+      "('package', 'decentralizepy.graphs.SmallWorld')\n",
+      "('graph_class', 'SmallWorld')\n",
+      "Section:  DATASET\n",
       "('package', 'decentralizepy.datasets.Femnist')\n",
-      "('class', 'Femnist')\n",
-      "('model', 'LogisticRegression')\n",
-      "('username', 'xiaoxu')\n",
-      "('password', 'xiaoxu')\n",
-      "('host', '127.0.0.1')\n",
-      "('port', '5432')\n",
-      "('db', 'xiaoxu_database')\n"
+      "('dataset_class', 'Femnist')\n",
+      "('model_class', 'LogisticRegression')\n",
+      "('n_procs', 1.0)\n",
+      "('train_dir', '')\n",
+      "('test_dir', '')\n",
+      "('sizes', '[0.4, 0.2, 0.3, 0.1]')\n",
+      "Section:  MODEL_PARAMS\n",
+      "('optimizer_package', 'torch.optim')\n",
+      "('optimizer_class', 'SGD')\n",
+      "('lr', 0.1)\n"
      ]
     }
    ],
    "source": [
-    "import configparser\n",
+    "from localconfig import LocalConfig\n",
     "\n",
     "def read_ini(file_path):\n",
-    "    config = configparser.ConfigParser()\n",
-    "    config.read(file_path)\n",
-    "    for section in config.sections():\n",
-    "        for key in config[section]:\n",
-    "            print((key, config[section][key]))\n",
+    "    config = LocalConfig(file_path)\n",
+    "    for section in config:\n",
+    "        print(\"Section: \", section)\n",
+    "        for key, value in config.items(section):\n",
+    "            print((key, value))\n",
     " \n",
     "read_ini(\"config.ini\")"
    ]
   },
+  {
+   "cell_type": "code",
+   "execution_count": 1,
+   "metadata": {},
+   "outputs": [
+    {
+     "name": "stdout",
+     "output_type": "stream",
+     "text": [
+      "15\n"
+     ]
+    }
+   ],
+   "source": [
+    "def func(a = 1, b = 2, c = 3):\n",
+    "    print(a + b + c)\n",
+    "\n",
+    "l = [3, 5, 7]\n",
+    "\n",
+    "func(*l)"
+   ]
+  },
+  {
+   "cell_type": "code",
+   "execution_count": 3,
+   "metadata": {},
+   "outputs": [
+    {
+     "ename": "TypeError",
+     "evalue": "spawn() got an unexpected keyword argument 'kwargs'",
+     "output_type": "error",
+     "traceback": [
+      "\u001b[0;31m---------------------------------------------------------------------------\u001b[0m",
+      "\u001b[0;31mTypeError\u001b[0m                                 Traceback (most recent call last)",
+      "\u001b[0;32m/tmp/ipykernel_52405/4231740097.py\u001b[0m in \u001b[0;36m<module>\u001b[0;34m\u001b[0m\n\u001b[1;32m      1\u001b[0m \u001b[0;32mfrom\u001b[0m \u001b[0mtorch\u001b[0m \u001b[0;32mimport\u001b[0m \u001b[0mmultiprocessing\u001b[0m \u001b[0;32mas\u001b[0m \u001b[0mmp\u001b[0m\u001b[0;34m\u001b[0m\u001b[0;34m\u001b[0m\u001b[0m\n\u001b[1;32m      2\u001b[0m \u001b[0;34m\u001b[0m\u001b[0m\n\u001b[0;32m----> 3\u001b[0;31m \u001b[0mmp\u001b[0m\u001b[0;34m.\u001b[0m\u001b[0mspawn\u001b[0m\u001b[0;34m(\u001b[0m\u001b[0mfn\u001b[0m \u001b[0;34m=\u001b[0m \u001b[0mfunc\u001b[0m\u001b[0;34m,\u001b[0m \u001b[0mnprocs\u001b[0m \u001b[0;34m=\u001b[0m \u001b[0;36m2\u001b[0m\u001b[0;34m,\u001b[0m \u001b[0margs\u001b[0m \u001b[0;34m=\u001b[0m \u001b[0;34m[\u001b[0m\u001b[0;34m]\u001b[0m\u001b[0;34m,\u001b[0m \u001b[0mkwargs\u001b[0m \u001b[0;34m=\u001b[0m \u001b[0;34m{\u001b[0m\u001b[0;34m'a'\u001b[0m\u001b[0;34m:\u001b[0m \u001b[0;36m4\u001b[0m\u001b[0;34m,\u001b[0m \u001b[0;34m'b'\u001b[0m\u001b[0;34m:\u001b[0m \u001b[0;36m5\u001b[0m\u001b[0;34m,\u001b[0m \u001b[0;34m'c'\u001b[0m\u001b[0;34m:\u001b[0m \u001b[0;36m6\u001b[0m\u001b[0;34m}\u001b[0m\u001b[0;34m)\u001b[0m\u001b[0;34m\u001b[0m\u001b[0;34m\u001b[0m\u001b[0m\n\u001b[0m",
+      "\u001b[0;31mTypeError\u001b[0m: spawn() got an unexpected keyword argument 'kwargs'"
+     ]
+    }
+   ],
+   "source": [
+    "from torch import multiprocessing as mp\n",
+    "\n",
+    "mp.spawn(fn = func, nprocs = 2, args = [], kwargs = {'a': 4, 'b': 5, 'c': 6})"
+   ]
+  },
+  {
+   "cell_type": "code",
+   "execution_count": 19,
+   "metadata": {},
+   "outputs": [
+    {
+     "data": {
+      "text/plain": [
+       "list"
+      ]
+     },
+     "execution_count": 19,
+     "metadata": {},
+     "output_type": "execute_result"
+    }
+   ],
+   "source": [
+    "l = '[0.4, 0.2, 0.3, 0.1]'\n",
+    "type(eval(l))"
+   ]
+  },
   {
    "cell_type": "code",
    "execution_count": null,

setup.cfg

@@ -41,6 +41,7 @@ install_requires =
         jsonlines
         pillow
         smallworld
+        localconfig
 include_package_data = True
 python_requires = >=3.6
 [options.packages.find]

setup.py

-#!$CONDA_PREFIX/bin/env python
+#!$CONDA_PREFIX/python
 from setuptools import setup
 
 # https://packaging.python.org/guides/single-sourcing-package-version/

src/decentralizepy/datasets/Data.py

+class Data:
+    """
+    This class defines the API for Data.
+    """
+
+    def __init__(self, x, y):
+        """
+        Constructor
+        Parameters
+        ----------
+        x : numpy array
+            A numpy array of data samples
+        y : numpy array
+            A numpy array of outputs corresponding to the sample
+        """
+        self.x = x
+        self.y = y
+
+    def __get_item__(self, i):
+        """
+        Function to get the item with index i.
+        Parameters
+        ----------
+        i : int
+            Index
+        Returns
+        -------
+        2-tuple
+            A tuple of the ith data sample and it's corresponding label
+        """
+        return self.x[i], self.y[i]

src/decentralizepy/datasets/Dataset.py

+def __conditional_value__(var, nul, default):
+    if var != nul:
+        return var
+    else:
+        return default
+
+
 class Dataset:
     """
-    This class defines the API for Dataset.
+    This class defines the Dataset API.
+    All datasets must follow this API.
     """
 
-    def __init__(self, x, y):
+    def __init__(self, rank='', n_procs='', train_dir='', test_dir='', sizes=''):
         """
-        Constructor
+        Constructor which reads the data files, instantiates and partitions the dataset
         Parameters
         ----------
-        x : numpy array
-            A numpy array of data samples
-        y : numpy array
-            A numpy array of outputs corresponding to the sample
+        rank : int, optional
+            Rank of the current process (to get the partition). Default value is assigned 0
+        n_procs : int, optional
+            The number of processes among which to divide the data. Default value is assigned 1
+        train_dir : str, optional
+            Path to the training data files. Required to instantiate the training set
+            The training set is partitioned according to n_procs and sizes
+        test_dir : str. optional
+            Path to the testing data files Required to instantiate the testing set
+        sizes : list(int), optional
+            A list of fractions specifying how much data to alot each process. Sum of fractions should be 1.0
+            By default, each process gets an equal amount.
         """
-        self.x = x
-        self.y = y
+        self.rank = __conditional_value__(rank, '', 0)
+        self.n_procs = __conditional_value__(n_procs, '', 1)
+        self.train_dir = __conditional_value__(train_dir, '', None)
+        self.test_dir = __conditional_value__(test_dir, '', None)
+        self.sizes = __conditional_value__(sizes, '', None)
+        if self.sizes:
+            if type(self.sizes) == str:
+                self.sizes = eval(self.sizes)
+
 
-    def __get_item__(self, i):
+        if train_dir:
+            self.__training__ = True
+        else:
+            self.__training__ = False
+
+        if test_dir:
+            self.__testing__ = True
+        else:
+            self.__testing__ = False
+
+    def get_trainset(self):
         """
-        Function to get the item with index i.
-        Parameters
-        ----------
-        i : int
-            Index
+        Function to get the training set
+        Returns
+        -------
+        Dataset
+        Raises
+        ------
+        RuntimeError
+            If the training set was not initialized
+        """
+        raise NotImplementedError
+
+    def get_testset(self):
+        """
+        Function to get the test set
         Returns
         -------
-        2-tuple
-            A tuple of the ith data sample and it's corresponding label
+        Dataset
+        Raises
+        ------
+        RuntimeError
+            If the test set was not initialized
         """
-        return self.x[i], self.y[i]
+        raise NotImplementedError

src/decentralizepy/datasets/Femnist.py

@@ -7,6 +7,7 @@ import torch
 import torch.nn.functional as F
 from torch import nn
 
+from decentralizepy.datasets.Data import Data
 from decentralizepy.datasets.Dataset import Dataset
 from decentralizepy.datasets.Partitioner import DataPartitioner
 
@@ -45,41 +46,41 @@ def __read_dir__(data_dir):
     return clients, num_samples, data
 
 
-class Femnist:
+class Femnist(Dataset):
     """
     Class for the FEMNIST dataset
     """
 
-    def __init__(self, train_dir=None, test_dir=None, rank=0, n_procs=1, sizes=None):
+    def __init__(self, rank='', n_procs='', train_dir='', test_dir='', sizes=''):
         """
         Constructor which reads the data files, instantiates and partitions the dataset
         Parameters
         ----------
+        rank : int, optional
+            Rank of the current process (to get the partition). Default value is assigned 0
+        n_procs : int, optional
+            The number of processes among which to divide the data. Default value is assigned 1
         train_dir : str, optional
             Path to the training data files. Required to instantiate the training set
             The training set is partitioned according to n_procs and sizes
         test_dir : str. optional
             Path to the testing data files Required to instantiate the testing set
-        rank : int, optional
-            Rank of the current process (to get the partition). Default value is 0
-        n_procs : int, optional
-            The number of processes among which to divide the data
         sizes : list(int), optional
-            A list of fractions specifying how much data to alot each process.
+            A list of fractions specifying how much data to alot each process. Sum of fractions should be 1.0
             By default, each process gets an equal amount.
         """
-        if train_dir:
-            self.__training__ = True
+        super().__init__(rank, n_procs, train_dir, test_dir, sizes)
+        if self.__training__:
             clients, num_samples, train_data = __read_dir__(train_dir)
             c_len = len(self.clients)
 
             if sizes == None:  # Equal distribution of data among processes
-                e = c_len // n_procs
+                e = c_len // self.n_procs
                 frac = e / c_len
-                sizes = [frac] * n_procs
-                sizes[-1] += 1.0 - frac * n_procs
+                sizes = [frac] * self.n_procs
+                sizes[-1] += 1.0 - frac * self.n_procs
 
-            my_clients = DataPartitioner(clients, sizes).use(rank)
+            my_clients = DataPartitioner(clients, sizes).use(self.rank)
             my_train_data = []
             self.clients = []
             self.num_samples = []
@@ -96,8 +97,7 @@ class Femnist:
                 my_train_data["y"], dtype=np.dtype("float64")
             ).reshape(-1, 1)
 
-        if test_dir:
-            self.__testing__ = True
+        if self.__testing__:
             _, _, test_data = __read_dir__(test_dir)
             test_data = test_data.values()
             self.test_x = np.array(test_data["x"], dtype=np.dtype("float64")).reshape(
@@ -154,7 +154,7 @@ class Femnist:
             If the training set was not initialized
         """
         if self.__training__:
-            return Dataset(self.train_x, self.train_y)
+            return Data(self.train_x, self.train_y)
         raise RuntimeError("Training set not initialized!")
 
     def get_testset(self):
@@ -169,7 +169,7 @@ class Femnist:
             If the test set was not initialized
         """
         if self.__testing__:
-            return Dataset(self.test_x, self.test_y)
+            return Data(self.test_x, self.test_y)
         raise RuntimeError("Test set not initialized!")
 
 

src/decentralizepy/node/Node.py

+from torch import utils, optim
+import importlib
+
+
+def __remove_keys__(d, keys_to_remove):
+    return {key: d[key] for key in d if key not in keys_to_remove}
+
 class Node:
     """
     This class defines the node (entity that performs learning, sharing and communication).
     """
 
-    def __init__(self, rank, mapping, graph, options):
+    def __init__(self, rank, mapping, graph, config, *args):
         """
         Constructor
         Parameters
         ----------
         rank : int
             Rank of process local to the machine
+        n_procs_local : int
+            Number of processes on current machine
         mapping : decentralizepy.mappings
             The object containing the mapping rank <--> uid
         graph : decentralizepy.graphs
             The object containing the global graph
-        dataset : decentralizepy.datasets class
-            The class whose object will be instantiated to init the dataset
+        config : dict
+            A dictionary of configurations. Must contain the following:
+            [DATASET]
+                dataset_package
+                dataset_class
+                model_class
+            [OPTIMIZER_PARAMS]
+                optimizer_package
+                optimizer_class
+        args : optional
+            Other arguments
         """
         self.rank = rank
         self.graph = graph
         self.mapping = mapping
-        self.options = options
+        
+        dataset_configs = dict(config.items("DATASET"))
+        dataset_module = importlib.import_module(dataset_configs["dataset_package"])
+        dataset_class = getattr(dataset_module, dataset_configs["dataset_class"])
+        
+        dataset_params = __remove_keys__(dataset_configs, ["dataset_package", "dataset_class", "model_class"])
+        self.dataset =  dataset_class(rank, **dataset_params)
+        self.trainset = self.dataset.get_trainset()
+
+        model_class = getattr(dataset_module, dataset_configs["model_class"])
+        self.model = model_class()
+
+        optimizer_configs = dict(config.items("OPTIMIZER_PARAMS"))
+        optimizer_module = importlib.import_module(optimizer_configs["optimizer_package"])
+        optimizer_class = getattr(optimizer_module, optimizer_configs["optimizer_class"])
+        
+        optimizer_params = __remove_keys__(optimizer_configs, ["optimizer_package", "optimizer_class"])
+        self.optimizer = optimizer_class(self.model.parameters(), **optimizer_params)
 
-    def __get_item__(self, i):
+        self.run()
+
+    def train_step(self):
         """
-        Function to get the item with index i.
-        Parameters
-        ----------
-        i : int
-            Index
-        Returns
-        -------
-        2-tuple
-            A tuple of the ith data sample and it's corresponding label
+        The training step
+        """
+        for epoch in self.epochs_per_round: # Instantiate this variable
+            for data, target in self.trainset: 
+                # Perform training step
+                raise NotImplementedError
+
+
+
+
+    def run(self):
+        """
+        The learning loop.
         """
-        return self.x[i], self.y[i]
+        while True:
+            # Receive data
+
+            # Learn
+
+
+            # Send data
+            raise NotImplementedError
+