Progress on DataGenerator.

Author: joseph-hellerstein

src/autoencodersb/data_generator.py

@@ -0,0 +1,42 @@
+'''Generates synthetic data for training and testing autoencoders.'''
+
+import autoencodersb.constants as cn
+
+import itertools
+import numpy as np  # type: ignore
+from torch.utils.data import DataLoader
+import pandas as pd    # type: ignore
+from typing import cast, Optional, Tuple
+
+
+class DataGenerator(object):
+
+    def __init__(self, 
+            num_sample: int = 1000,
+            num_independent_feature: int = 2,
+            num_feature: int = 10,
+            num_data_value: int = 10,
+            data_density: float = 1.0,  # Number of values per integer interval
+            noise_std: float = 0.0
+            ):
+        self.num_sample = num_sample
+        self.num_independent_feature = num_independent_feature
+        self.num_feature = num_feature
+        self.num_data_value = num_data_value
+        self.data_density = data_density
+        self.noise_std = noise_std
+
+    def generateFullData(self) -> DataLoader:
+        """Generates the full synthetic dataset."""
+        raise NotImplementedError("This method should be overridden by subclasses.")
+    
+    def generateIndependentFeatures(self) -> np.ndarray:
+        """
+        Generates an array of independent features.
+
+        Returns:
+            np.ndarray (N X I): An array of size independent features.
+                N is self.num_sample 
+                I self.num_independent_feature
+        """
+        raise NotImplementedError("This method should be overridden by subclasses.") 

src/autoencodersb/data_generator_iid.py

@@ -0,0 +1,42 @@
+'''Generates synthetic data where the independent variables are iid.'''
+
+import autoencodersb.constants as cn
+from autoencodersb.data_generator import DataGenerator  # type: ignore
+
+import itertools
+import numpy as np  # type: ignore
+from torch.utils.data import DataLoader
+import pandas as pd    # type: ignore
+from typing import cast, Optional, Tuple
+
+
+class DataGeneratorIID(DataGenerator):
+
+    def __init__(self, 
+            num_sample: int = 1000,
+            num_independent_feature: int = 2,
+            num_feature: int = 10,
+            num_data_value: int = 10,
+            data_density: float = 1.0,  # Number of values per integer interval
+            noise_std: float = 0.0
+            ):
+        super().__init__(
+            num_sample=num_sample,
+            num_independent_feature=num_independent_feature,
+            num_feature=num_feature,
+            num_data_value=num_data_value,
+            data_density=data_density,
+            noise_std=noise_std
+        )
+    
+    def generateIndependentFeatures(self) -> np.ndarray:
+        """
+        Generates an array of independent features.
+
+        Returns:
+            np.ndarray (N X I): An array of size independent features.
+                N is self.num_sample 
+                I self.num_independent_feature
+        """
+        return np.random.randint(1, self.num_data_value + 1, 
+                (self.num_sample, self.num_independent_feature)).astype(np.float32) / self.data_density

src/autoencodersb/data_generator_path.py

@@ -0,0 +1,60 @@
+'''Generates synthetic data where the independent variables are a random walk.'''
+
+import autoencodersb.constants as cn
+from autoencodersb.data_generator import DataGenerator  # type: ignore
+
+import numpy as np  # type: ignore
+import pandas as pd    # type: ignore
+from typing import cast, Optional, Tuple
+
+
+class DataGeneratorPath(DataGenerator):
+
+    def __init__(self, 
+            num_sample: int = 1000,
+            num_independent_feature: int = 2,
+            num_feature: int = 10,
+            num_data_value: int = 10,
+            data_density: float = 1.0,  # Number of values per integer interval
+            noise_std: float = 0.0
+            ):
+        super().__init__(
+            num_sample=num_sample,
+            num_independent_feature=num_independent_feature,
+            num_feature=num_feature,
+            num_data_value=num_data_value,
+            data_density=data_density,
+            noise_std=noise_std
+        )
+
+    def generateIndependentFeature(self) -> np.ndarray:
+        """
+        Generates a random walk
+
+        Returns:
+            np.ndarray (N X I): An array of size independent features.
+                N is self.num_sample
+                I is self.num_independent_feature
+        """
+        path = np.zeros((self.num_sample, self.num_independent_feature), dtype=np.float32)
+        for i in range(1, self.num_sample):
+            path[i] = path[i - 1] + np.random.normal(0, self.noise_std, size=(self.num_independent_feature,))
+        return path
+
+                N is self.num_sample 
+                I self.num_independent_feature
+        """
+        return np.random.randint(1, self.num_data_value + 1, 
+                (self.num_sample, self.num_independent_feature)).astype(np.float32) / self.data_density
+    
+    def generateIndependentFeatures(self) -> np.ndarray:
+        """
+        Generates an array of independent features.
+
+        Returns:
+            np.ndarray (N X I): An array of size independent features.
+                N is self.num_sample 
+                I self.num_independent_feature
+        """
+        return np.random.randint(1, self.num_data_value + 1, 
+                (self.num_sample, self.num_independent_feature)).astype(np.float32) / self.data_density

src/autoencodersb/polynomial.py

@@ -0,0 +1,37 @@
+'''Constructs a polynomial of arrays.'''
+
+from autoencodersb.polynomial_rational import PolynomialRational  # type: ignore
+
+import numpy as np
+from typing import List
+
+"""
+A PolynomialTerm of D variables has a coefficient and specifies exponents for 0 or more
+    independent variables. The term represents the product of the independent variables raised
+    to the power of the exponents times the coefficient.
+A PolynomialDivision is the ratio of two PolynomialTerm.
+An expression is a collection of PolynomialTerm and PolynomialDivision.
+"""
+
+class Polynomial(object):
+
+    def __init__(self, polynomial_rationals: List[PolynomialRational] ):
+        self.num_variable = np.sum([t.num_variable for t in polynomial_rationals])
+        self.num_term = len(polynomial_rationals)
+        self.terms = polynomial_rationals
+
+    def __repr__(self):
+        return " + ".join([str(term) for term in self.terms])
+    
+    def evaluate(self, independent_variable_arr: np.ndarray) -> np.ndarray:
+        """Evaluate the polynomial at the given independent variable values."""
+        return np.sum([term.evaluate(independent_variable_arr) for term in self.terms], axis=0)
+
+    def make_polynomial(self, degree: int) -> np.ndarray:
+        """Construct a polynomial of the given degree."""
+        # Create a grid of independent variable values
+        x = np.linspace(-1, 1, 100)
+        X = np.array(np.meshgrid(*[x]*self.num_variable)).T.reshape(-1, self.num_variable)
+        # Compute the polynomial features
+        poly = np.hstack([X**d for d in range(1, degree + 1)])
+        return poly

src/autoencodersb/polynomial_maker.py

@@ -0,0 +1,32 @@
+'''Represents a product of independent variables raised to an exponent.'''
+
+from autoencodersb.polynomial_term import PolynomialTerm  # type: ignore
+
+import numpy as np
+import pandas as pd  # type: ignore
+from typing import List
+
+
+class PolynomialRational(object):
+
+    def __init__(self, numerator: PolynomialTerm, denominator: PolynomialTerm):
+        self.numerator = numerator
+        self.denominator = denominator
+        exponents = numerator.exponent_arr + denominator.exponent_arr
+        self.num_variable = len([e for e in exponents if e != 0])
+
+    def __repr__(self):
+        return f"({self.numerator}) / ({self.denominator})"
+
+    def evaluate(self, independent_variable_arr: np.ndarray) -> np.ndarray:
+        """Divide this polynomial term by another polynomial term.
+
+        Args:
+            other (PolynomialTerm): The polynomial term to divide by.
+
+        Returns:
+            PolynomialTerm: The resulting polynomial term after division.
+        """
+        numerator_value = self.numerator.evaluate(independent_variable_arr)
+        denominator_value = self.denominator.evaluate(independent_variable_arr)
+        return numerator_value / denominator_value

src/autoencodersb/polynomial_rational.py

@@ -0,0 +1,30 @@
+'''Represents a product of independent variables raised to an exponent.'''
+
+import numpy as np
+import pandas as pd  # type: ignore
+from typing import List
+
+
+class PolynomialTerm(object):
+
+    def __init__(self, coefficient: float, exponents: List[float]):
+        self.coefficient = coefficient
+        self.exponent_arr = np.array(exponents)
+        self.num_variable = len([e for e in exponents if e != 0])
+
+    def __repr__(self):
+        term_strs = [f"X_{n}**{p}" for n, p in enumerate(self.exponent_arr) if p != 0]
+        term_str = f"{self.coefficient} * {' * '.join(term_strs)}"
+        return term_str
+    
+    def evaluate(self, independent_variable_arr: np.ndarray) -> np.ndarray:
+        """Evaluate the polynomial term.
+
+        Args:
+            independent_variable_arr (np.ndarray): N X I array of independent variables
+
+        Returns:
+            np.ndarray: Result of the evaluation
+        """
+        # Evaluate the term by multiplying the coefficient with the independent variables raised to the appropriate powers
+        return self.coefficient * np.prod(independent_variable_arr ** self.exponent_arr, axis=1)

src/autoencodersb/polynomial_term.py

@@ -1,11 +1,11 @@
-from iplane.dataset_csv import DatasetCSV # type: ignore
-import iplane.constants as cn  # type: ignore
+from autoencodersb.dataset_csv import DatasetCSV # type: ignore
+import autoencodersb.constants as cn  # type: ignore
 
 import pandas as pd    # type: ignore
 import unittest
 
 IGNORE_TESTS = False
-IS_PLOT = False
+IS_PLOT = False:while
 NUM_EPOCH = 3
 
 DATASET_CSV_PATH = "tests/test_dataset_csv.csv"