Repository: RADAR-base/radarpipeline (language: python)

Issue:
refactor feature and feature group abstract classes
Added `preprocess` function in the abstract class. The idea is that first the dataframe will be fetched and preprocessed in the `preprocess` method and the output from that will be passed on to the `compute` method. 


Discussion / hints:

Fix patch:
diff --git a/.flake8 b/.flake8
new file mode 100644
index 0000000..5b8ab6b
--- /dev/null
+++ b/.flake8
@@ -0,0 +1,5 @@
+[flake8]
+    ignore = E203, E266, E501, W503, F403, F401
+    max-line-length = 88
+    max-complexity = 18
+    select = B,C,E,F,W,T4,B9
diff --git a/.gitattributes b/.gitattributes
new file mode 100644
index 0000000..6313b56
--- /dev/null
+++ b/.gitattributes
@@ -0,0 +1,1 @@
+* text=auto eol=lf
diff --git a/.github/CODE_OF_CONDUCT.md b/.github/CODE_OF_CONDUCT.md
new file mode 100644
index 0000000..be2a88a
--- /dev/null
+++ b/.github/CODE_OF_CONDUCT.md
@@ -0,0 +1,128 @@
+# Contributor Covenant Code of Conduct
+
+## Our Pledge
+
+We as members, contributors, and leaders pledge to make participation in our
+community a harassment-free experience for everyone, regardless of age, body
+size, visible or invisible disability, ethnicity, sex characteristics, gender
+identity and expression, level of experience, education, socio-economic status,
+nationality, personal appearance, race, religion, or sexual identity
+and orientation.
+
+We pledge to act and interact in ways that contribute to an open, welcoming,
+diverse, inclusive, and healthy community.
+
+## Our Standards
+
+Examples of behavior that contributes to a positive environment for our
+community include:
+
+* Demonstrating empathy and kindness toward other people
+* Being respectful of differing opinions, viewpoints, and experiences
+* Giving and gracefully accepting constructive feedback
+* Accepting responsibility and apologizing to those affected by our mistakes,
+  and learning from the experience
+* Focusing on what is best not just for us as individuals, but for the
+  overall community
+
+Examples of unacceptable behavior include:
+
+* The use of sexualized language or imagery, and sexual attention or
+  advances of any kind
+* Trolling, insulting or derogatory comments, and personal or political attacks
+* Public or private harassment
+* Publishing others' private information, such as a physical or email
+  address, without their explicit permission
+* Other conduct which could reasonably be considered inappropriate in a
+  professional setting
+
+## Enforcement Responsibilities
+
+Community leaders are responsible for clarifying and enforcing our standards of
+acceptable behavior and will take appropriate and fair corrective action in
+response to any behavior that they deem inappropriate, threatening, offensive,
+or harmful.
+
+Community leaders have the right and responsibility to remove, edit, or reject
+comments, commits, code, wiki edits, issues, and other contributions that are
+not aligned to this Code of Conduct, and will communicate reasons for moderation
+decisions when appropriate.
+
+## Scope
+
+This Code of Conduct applies within all community spaces, and also applies when
+an individual is officially representing the community in public spaces.
+Examples of representing our community include using an official e-mail address,
+posting via an official social media account, or acting as an appointed
+representative at an online or offline event.
+
+## Enforcement
+
+Instances of abusive, harassing, or otherwise unacceptable behavior may be
+reported to the community leaders responsible for enforcement at
+radar-base@kcl.ac.uk.
+All complaints will be reviewed and investigated promptly and fairly.
+
+All community leaders are obligated to respect the privacy and security of the
+reporter of any incident.
+
+## Enforcement Guidelines
+
+Community leaders will follow these Community Impact Guidelines in determining
+the consequences for any action they deem in violation of this Code of Conduct:
+
+### 1. Correction
+
+**Community Impact**: Use of inappropriate language or other behavior deemed
+unprofessional or unwelcome in the community.
+
+**Consequence**: A private, written warning from community leaders, providing
+clarity around the nature of the violation and an explanation of why the
+behavior was inappropriate. A public apology may be requested.
+
+### 2. Warning
+
+**Community Impact**: A violation through a single incident or series
+of actions.
+
+**Consequence**: A warning with consequences for continued behavior. No
+interaction with the people involved, including unsolicited interaction with
+those enforcing the Code of Conduct, for a specified period of time. This
+includes avoiding interactions in community spaces as well as external channels
+like social media. Violating these terms may lead to a temporary or
+permanent ban.
+
+### 3. Temporary Ban
+
+**Community Impact**: A serious violation of community standards, including
+sustained inappropriate behavior.
+
+**Consequence**: A temporary ban from any sort of interaction or public
+communication with the community for a specified period of time. No public or
+private interaction with the people involved, including unsolicited interaction
+with those enforcing the Code of Conduct, is allowed during this period.
+Violating these terms may lead to a permanent ban.
+
+### 4. Permanent Ban
+
+**Community Impact**: Demonstrating a pattern of violation of community
+standards, including sustained inappropriate behavior,  harassment of an
+individual, or aggression toward or disparagement of classes of individuals.
+
+**Consequence**: A permanent ban from any sort of public interaction within
+the community.
+
+## Attribution
+
+This Code of Conduct is adapted from the [Contributor Covenant][homepage],
+version 2.0, available at
+https://www.contributor-covenant.org/version/2/0/code_of_conduct.html.
+
+Community Impact Guidelines were inspired by [Mozilla's code of conduct
+enforcement ladder](https://github.com/mozilla/diversity).
+
+[homepage]: https://www.contributor-covenant.org
+
+For answers to common questions about this code of conduct, see the FAQ at
+https://www.contributor-covenant.org/faq. Translations are available at
+https://www.contributor-covenant.org/translations.
diff --git a/.github/ISSUE_TEMPLATE/bug_report.md b/.github/ISSUE_TEMPLATE/bug_report.md
new file mode 100644
index 0000000..df8f902
--- /dev/null
+++ b/.github/ISSUE_TEMPLATE/bug_report.md
@@ -0,0 +1,40 @@
+---
+name: Bug report
+about: Create a report to help us improve
+title: ''
+labels: 'bug'
+assignees: ''
+---
+
+**Describe the bug**
+A clear and concise description of what the bug is.
+
+**To Reproduce**
+Steps to reproduce the behavior:
+
+1. Go to '...'
+2. Click on '....'
+3. Scroll down to '....'
+4. See error
+
+**Expected behaviour**
+A clear and concise description of what you expected to happen.
+
+**Screenshots**
+If applicable, add screenshots to help explain your problem.
+
+**Desktop (please complete the following information):**
+
+-   OS: [e.g. iOS]
+-   Browser [e.g. chrome, safari]
+-   Version [e.g. 22]
+
+**Smartphone (please complete the following information):**
+
+-   Device: [e.g. iPhone6]
+-   OS: [e.g. iOS8.1]
+-   Browser [e.g. stock browser, safari]
+-   Version [e.g. 22]
+
+**Additional context**
+Add any other context about the problem here.
diff --git a/.github/ISSUE_TEMPLATE/feature_request.md b/.github/ISSUE_TEMPLATE/feature_request.md
new file mode 100644
index 0000000..2f28cea
--- /dev/null
+++ b/.github/ISSUE_TEMPLATE/feature_request.md
@@ -0,0 +1,19 @@
+---
+name: Feature request
+about: Suggest an idea for this project
+title: ''
+labels: ''
+assignees: ''
+---
+
+**Is your feature request related to a problem? Please describe.**
+A clear and concise description of what the problem is. Ex. I'm always frustrated when [...]
+
+**Describe the solution you'd like**
+A clear and concise description of what you want to happen.
+
+**Describe alternatives you've considered**
+A clear and concise description of any alternative solutions or features you've considered.
+
+**Additional context**
+Add any other context or screenshots about the feature request here.
diff --git a/.gitignore b/.gitignore
index b6e4761..591d2ef 100644
--- a/.gitignore
+++ b/.gitignore
@@ -127,3 +127,15 @@ dmypy.json
 
 # Pyre type checker
 .pyre/
+
+# VS Code files
+.vscode/
+
+# Logs
+log/
+
+# RADAR Directories
+data/
+features/
+temp/
+output/
diff --git a/.gitmodules b/.gitmodules
new file mode 100644
index 0000000..cc8e051
--- /dev/null
+++ b/.gitmodules
@@ -0,0 +1,4 @@
+[submodule "mockdata"]
+	path = mockdata
+	url = https://github.com/RADAR-base-Analytics/mockdata
+	branch = main
diff --git a/.isort.cfg b/.isort.cfg
new file mode 100644
index 0000000..2d18e92
--- /dev/null
+++ b/.isort.cfg
@@ -0,0 +1,3 @@
+[settings]
+known_third_party = git,pandas,pyspark,requests,setuptools,yaml
+profile = black
diff --git a/.pre-commit-config.yaml b/.pre-commit-config.yaml
new file mode 100644
index 0000000..690304e
--- /dev/null
+++ b/.pre-commit-config.yaml
@@ -0,0 +1,30 @@
+repos:
+    - repo: https://github.com/pre-commit/pre-commit-hooks
+      rev: v4.3.0
+      hooks:
+          - id: check-yaml
+          - id: end-of-file-fixer
+          - id: trailing-whitespace
+            args: [--markdown-linebreak-ext=md]
+
+    - repo: https://github.com/asottile/seed-isort-config
+      rev: v2.2.0
+      hooks:
+          - id: seed-isort-config
+
+    - repo: https://github.com/pycqa/isort
+      rev: 5.10.1
+      hooks:
+          - id: isort
+            name: isort (python)
+
+    - repo: https://github.com/psf/black
+      rev: 22.6.0
+      hooks:
+          - id: black
+            language_version: python3.9
+
+    - repo: https://gitlab.com/pycqa/flake8
+      rev: 3.9.2
+      hooks:
+          - id: flake8
diff --git a/README.md b/README.md
index 79649c3..6f2e875 100644
--- a/README.md
+++ b/README.md
@@ -1,2 +1,102 @@
-# radar-pipeline
-A python feature generation and visualzation package use with RADAR project data.
+<h1 align="center">RADAR Pipeline</h1>
+
+<p align="center">
+<a href="https://github.com/RADAR-base/radarpipeline/issues"><img alt="GitHub issues" src="https://img.shields.io/github/issues/RADAR-base/radarpipeline"></a>
+<a href="https://github.com/thepushkarp/radarpipeline/pulls"><img alt="GitHub pull requests" src="https://img.shields.io/github/issues-pr/radar-base/radarpipeline"></a>
+<a href="https://github.com/RADAR-base/radarpipeline/network"><img alt="GitHub forks" src="https://img.shields.io/github/forks/RADAR-base/radarpipeline"></a>
+<a href="https://github.com/RADAR-base/radarpipeline/stargazers"><img alt="GitHub stars" src="https://img.shields.io/github/stars/RADAR-base/radarpipeline"></a>
+<a href="https://github.com/RADAR-base/radarpipeline/blob/main/LICENSE"><img alt="GitHub license" src="https://img.shields.io/github/license/RADAR-base/radarpipeline"></a>
+<a href="https://github.com/psf/black"><img alt="Code style: black" src="https://img.shields.io/badge/code%20style-black-000000.svg"></a>
+</p>
+
+<p align="center">An open-source python feature generation and visualization package use with RADAR project data.</p>
+
+---
+
+## Wiki
+
+Please visit the [RADAR Pipeline Wiki](https://github.com/RADAR-base/radarpipeline/wiki) to learn more about RADAR Pipeline.
+
+Wiki resources:
+
+-   [Home](https://github.com/RADAR-base/radarpipeline/wiki)
+-   [Installation](https://github.com/RADAR-base/radarpipeline/wiki/Installation#installation-instructions)
+-   [Contributor Guide](https://github.com/RADAR-base/radarpipeline/wiki/Contributor-Guide)
+-   [Mock Pipeline](https://github.com/RADAR-base/radarpipeline/wiki/Mock-Pipeline)
+-   [Configuration](https://github.com/RADAR-base/radarpipeline/wiki/Configuration)
+
+## How to run
+
+> **Note**
+>
+> If you are using Windows, please install Spark and set environment variables as mentioned [here](https://sparkbyexamples.com/spark/apache-spark-installation-on-windows/) before going through the installation below. You'll need to set the environment variables given [here](https://spark.apache.org/docs/1.6.0/configuration.html#environment-variables).
+
+-   Clone the repository (with all the submodules):
+
+    ```bash
+    $ git clone --recurse-submodules https://github.com/RADAR-base/radarpipeline.git
+    ```
+
+-   Change the directory to `radarpipeline`:
+
+    ```bash
+    $ cd radarpipeline
+    ```
+
+-   Checkout the development branch:
+
+    ```bash
+    $ git checkout dev
+    ```
+
+-   Create a [virtual environment](https://virtualenv.pypa.io/en/latest/installation.html) and activate it. The instructions here use `virtualenv` but feel free to use any python virtual environment manager of your choice.
+
+    -   Install the virtualenv package:
+
+        ```bash
+        $ python -m pip install --user virtualenv
+        ```
+
+    -   Create a python virtual environment:
+
+        ```bash
+        $ python -m virtualenv env
+        ```
+
+    -   Activate the virtual environment:
+
+        On Windows, run:
+
+        ```bash
+        $ .\env\Scripts\activate
+        ```
+
+        On Linux or MacOS, run:
+
+        ```bash
+        $ source ./env/bin/activate
+        ```
+
+-   Install the dependencies:
+
+    ```bash
+    $ python -m pip install -r requirements.txt
+    ```
+
+-   Install the module as a python package by running the command
+
+    ```bash
+    $ python -m pip install -e .
+    ```
+
+-   To verify the installations, run the following command in the project root directory to run the pipeline:
+
+    ```bash
+    $ python .
+    ```
+
+-   The pipeline would do a mock run and ingest the data in the `mock-data` directory. You can see some outputs in the CLI and if the project is installed correctly, the mock pipeline would run without errors and save the data to the `output` directory.
+
+## License
+
+This project is licensed under the [ Apache License, Version 2.0](https://github.com/RADAR-base/radarpipeline/blob/main/LICENSE).
diff --git a/__init__.py b/__init__.py
new file mode 100644
index 0000000..e69de29
diff --git a/__main__.py b/__main__.py
new file mode 100644
index 0000000..e2d9553
--- /dev/null
+++ b/__main__.py
@@ -0,0 +1,4 @@
+from radarpipeline import radarpipeline
+
+if __name__ == "__main__":
+    radarpipeline.run()
diff --git a/config.yaml b/config.yaml
new file mode 100644
index 0000000..d1b7717
--- /dev/null
+++ b/config.yaml
@@ -0,0 +1,23 @@
+project:
+    project_name: mock_project
+    description: mock_description
+    version: mock_version
+
+input_data:
+    data_location: mock
+    local_directory: mockdata/mockdata
+    data_format: csv
+
+configurations:
+    df_type: 'pandas'
+
+features:
+    - location: 'https://github.com/RADAR-base-Analytics/mockfeatures'
+      feature_groups:
+          - MockFeatureGroup
+
+output_data:
+    output_location: local
+    local_directory: output/mockdata
+    data_format: csv
+    compress: false
diff --git a/mockdata b/mockdata
new file mode 160000
index 0000000..2792051
--- /dev/null
+++ b/mockdata
@@ -0,0 +1,1 @@
+Subproject commit 2792051f2e3673f7f7811c8d63d035b4923ede79
diff --git a/pyproject.toml b/pyproject.toml
new file mode 100644
index 0000000..eaf7f9e
--- /dev/null
+++ b/pyproject.toml
@@ -0,0 +1,16 @@
+[tool.black]
+line-length = 88
+include = '.pyi?$'
+exclude = '''
+/(
+    .git
+    | .hg
+    | .mypy_cache
+    | .tox
+    | .venv
+    | _build
+    | buck-out
+    | build
+    | dist
+)/
+'''
diff --git a/radarpipeline/__init__.py b/radarpipeline/__init__.py
index e69de29..4eea2a6 100644
--- a/radarpipeline/__init__.py
+++ b/radarpipeline/__init__.py
@@ -0,0 +1,1 @@
+from radarpipeline.project import Project
diff --git a/radarpipeline/common/__init__.py b/radarpipeline/common/__init__.py
new file mode 100644
index 0000000..8c03e04
--- /dev/null
+++ b/radarpipeline/common/__init__.py
@@ -0,0 +1,1 @@
+from radarpipeline.common import constants, logger, utils
diff --git a/radarpipeline/common/constants.py b/radarpipeline/common/constants.py
new file mode 100644
index 0000000..3424510
--- /dev/null
+++ b/radarpipeline/common/constants.py
@@ -0,0 +1,3 @@
+ENCODING = "utf-8"
+LINESEP = "\n"
+CSV_DELIMITER = ","
diff --git a/radarpipeline/common/logger.py b/radarpipeline/common/logger.py
new file mode 100644
index 0000000..4872937
--- /dev/null
+++ b/radarpipeline/common/logger.py
@@ -0,0 +1,47 @@
+import logging
+import logging.handlers
+import os
+import pathlib
+from datetime import datetime
+
+
+def logger_init(level: int = logging.INFO) -> None:
+    """
+    Initialize the logger
+
+    Parameters
+    ----------
+    level : int, optional
+        The level of the logger.
+        The default is logging.INFO.
+    """
+
+    # Get logger
+    logger = logging.getLogger()  # root logger
+    logger.setLevel(level)
+
+    # File handler
+    path = "log"
+    year = str(datetime.now().year)
+    month = str(datetime.now().month)
+    filename = datetime.now().strftime("%Y%m%d") + "_logfile.log"
+    log_directory = os.path.join(path, year, month)
+    log_filename = os.path.join(log_directory, filename)
+
+    pathlib.Path(log_directory).mkdir(parents=True, exist_ok=True)
+    file = logging.handlers.TimedRotatingFileHandler(
+        log_filename, when="midnight", interval=1
+    )
+    file_format = logging.Formatter("%(asctime)s [%(levelname)s]: %(message)s")
+    file.setLevel(logging.INFO)
+    file.setFormatter(file_format)
+
+    # Stream handler
+    stream = logging.StreamHandler()
+    stream_format = logging.Formatter("%(asctime)s [%(levelname)s]: %(message)s")
+    stream.setLevel(logging.INFO)
+    stream.setFormatter(stream_format)
+
+    # Adding all handlers to the logs
+    logger.addHandler(file)
+    logger.addHandler(stream)
diff --git a/radarpipeline/common/utils.py b/radarpipeline/common/utils.py
new file mode 100644
index 0000000..c202dbc
--- /dev/null
+++ b/radarpipeline/common/utils.py
@@ -0,0 +1,130 @@
+import os
+from functools import reduce
+from typing import Any, Dict, List
+from urllib.parse import urlparse
+
+import pyspark.sql as ps
+import requests
+import yaml
+
+from radarpipeline.common import constants
+
+
+def read_yaml(yaml_file_path: str) -> Dict[str, Any]:
+    """
+    Read yaml file and return a dict of the content
+
+    Parameters
+    ----------
+    yaml_file_path : str
+        Path to yaml file
+
+    Returns
+    -------
+    Dict[str, Any]
+        Dict of the content of the yaml file
+
+    Raises
+    ------
+    ValueError
+        If yaml file is not found or is not a yaml file, or is empty
+    """
+
+    if not os.path.exists(yaml_file_path):
+        raise ValueError("Input file does not exist ")
+    if not yaml_file_path.lower().endswith(".yaml"):
+        raise ValueError("Input file is not a yaml file")
+    if os.stat(yaml_file_path).st_size == 0:
+        raise ValueError("Input file is empty")
+
+    with open(yaml_file_path, "r", encoding=constants.ENCODING) as file:
+        config = yaml.load(file, Loader=yaml.FullLoader)
+
+    return config
+
+
+def combine_pyspark_dfs(dfs: List[ps.DataFrame]) -> ps.DataFrame:
+    """
+    Combine multiple pyspark dataframes of same schema into one
+
+    Parameters
+    ----------
+    dfs : List[ps.DataFrame]
+        List of pyspark dataframes
+
+    Returns
+    -------
+    ps.DataFrame
+        Combined pyspark dataframe
+    """
+    df_comb = reduce(
+        lambda df1, df2: df1.unionByName(df2, allowMissingColumns=True), dfs
+    )
+    return df_comb
+
+
+def is_valid_github_path(path: str) -> bool:
+    """
+    Check if the current path is a valid github path
+
+    Parameters
+    ----------
+    path : str
+        Path to check
+    Returns
+    -------
+    bool
+        True if the path is valid, False otherwise
+    """
+
+    parsed = urlparse(path)
+    if parsed.scheme not in ["https", "http"] or parsed.netloc != "github.com":
+        return False
+
+    if requests.head(path, allow_redirects=True).status_code != 200:
+        return False
+
+    return True
+
+
+def get_repo_name_from_url(url: str) -> str:
+    """
+    Return the name of the repository from the url
+    Reference: https://stackoverflow.com/a/55137835/10307491
+
+    Parameters
+    ----------
+    url : str
+        Url of the repository
+
+    Returns
+    -------
+    str
+        Name of the repository
+    """
+    last_slash_index = url.rfind("/")
+    last_suffix_index = url.rfind(".git")
+    if last_suffix_index < 0:
+        last_suffix_index = len(url)
+
+    if last_slash_index < 0 or last_suffix_index <= last_slash_index:
+        raise Exception(f"Badly formatted url {url}")
+
+    return url[last_slash_index + 1 : last_suffix_index]
+
+
+def pascal_to_snake_case(s: str) -> str:
+    """
+    Convert a string from PascalCase to snake_case
+
+    Parameters
+    ----------
+    s : str
+        String to convert
+
+    Returns
+    -------
+    str
+        Converted string
+    """
+    return "".join(["_" + i.lower() if i.isupper() else i for i in s]).lstrip("_")
diff --git a/radarpipeline/datalib/__init__.py b/radarpipeline/datalib/__init__.py
new file mode 100644
index 0000000..54e761a
--- /dev/null
+++ b/radarpipeline/datalib/__init__.py
@@ -0,0 +1,4 @@
+from radarpipeline.datalib.abc import Data
+from radarpipeline.datalib.radar_data import RadarData
+from radarpipeline.datalib.radar_user_data import RadarUserData
+from radarpipeline.datalib.radar_variable_data import RadarVariableData
diff --git a/radarpipeline/datalib/abc.py b/radarpipeline/datalib/abc.py
new file mode 100644
index 0000000..fae5066
--- /dev/null
+++ b/radarpipeline/datalib/abc.py
@@ -0,0 +1,37 @@
+from abc import ABC, abstractmethod
+from typing import Any, List
+
+
+class Data(ABC):
+    """
+    Abstract class for data.
+    """
+
+    _data: Any
+    df_type: str
+
+    def __init__(self) -> None:
+        pass
+
+    @abstractmethod
+    def get_data(self) -> Any:
+        pass
+
+    @abstractmethod
+    def set_data(self, data: Any) -> None:
+        pass
+
+    @abstractmethod
+    def get_data_keys(self) -> List[str]:
+        pass
+
+    @abstractmethod
+    def get_data_size(self) -> int:
+        pass
+
+    def _preprocess_data(self) -> None:
+        """
+        Preprocess the data
+        """
+        for key in self.get_data_keys():
+            self._data[key]._preprocess_data()
diff --git a/radarpipeline/datalib/radar_data.py b/radarpipeline/datalib/radar_data.py
new file mode 100644
index 0000000..92d967b
--- /dev/null
+++ b/radarpipeline/datalib/radar_data.py
@@ -0,0 +1,142 @@
+from typing import Dict, List, Optional, Union
+
+import pandas as pd
+
+from radarpipeline.common import utils
+from radarpipeline.datalib.abc import Data
+from radarpipeline.datalib.radar_user_data import RadarUserData
+from radarpipeline.datatypes import DataType
+
+
+class RadarData(Data):
+    """
+    Class for reading the RADAR data
+    """
+
+    _data: Dict[str, RadarUserData]
+
+    def __init__(self, data: Dict[str, RadarUserData], df_type: str = "pandas") -> None:
+        self._data = data
+        self.df_type = df_type
+
+    def get_data(self) -> Dict[str, RadarUserData]:
+        return self._data
+
+    def set_data(self, data: Dict[str, RadarUserData]) -> None:
+        self._data = data
+
+    def get_data_keys(self) -> List[str]:
+        return list(self._data.keys())
+
+    def get_data_size(self) -> int:
+        return len(self._data.items())
+
+    def _get_data_by_key(self, key: str) -> Optional[RadarUserData]:
+        return self._data.get(key, None)
+
+    def _get_all_user_ids(self) -> List[str]:
+        """
+        Get all user ids in the data of the RADAR data
+
+        Returns
+        -------
+        List[str]
+            The list of all user ids in the data of the RADAR data
+        """
+
+        return self.get_data_keys()
+
+    def get_combined_data_by_variable(
+        self, variables: Union[str, List[str]]
+    ) -> Union[DataType, List[DataType]]:
+        """
+        Returns the combined data of the RADAR data for the given variables
+
+        Parameters
+        ----------
+        variables : Union[str, List[str]]
+            The variable(s) to get the data for
+
+        Returns
+        -------
+        Union[
+            DataType,
+            List[DataType]
+        ]
+            The combined data of the RADAR data for the given variables
+        """
+
+        is_only_one_var = False
+        if isinstance(variables, str):
+            is_only_one_var = True
+            variables = [variables]
+
+        all_user_ids = self._get_all_user_ids()
+        variable_data_list = []
+        variable_dict = {}
+
+        # Store the data of the given variables of each user in a dictionary
+        for user_id in all_user_ids:
+            user_data = self._get_data_by_key(user_id)
+            if user_data is not None:
+                user_variables = user_data._get_all_variables()
+                for var in variables:
+                    if var in user_variables:
+                        var_data = user_data.get_data_by_variable(var)
+                        if var_data is not None:
+                            if var not in variable_dict:
+                                variable_dict[var] = []
+                            variable_dict[var].append(var_data.get_data())
+
+        # Combine the all data for each variable
+        for var in variable_dict:
+            if len(variable_dict[var]) > 0:
+                if self.df_type == "spark":
+                    combined_df = utils.combine_pyspark_dfs(variable_dict[var])
+                else:
+                    combined_df = pd.concat(variable_dict[var], ignore_index=True)
+                variable_data_list.append(combined_df)
+
+        if is_only_one_var:
+            return variable_data_list[0]
+        else:
+            return variable_data_list
+
+    def get_data_by_user_id(
+        self, user_ids: Union[str, List[str]]
+    ) -> Union[RadarUserData, List[RadarUserData]]:
+        """
+        Returns the data of the RADAR data for the given user ids
+
+        Parameters
+        ----------
+        user_ids : Union[str, List[str]]
+            The user id(s) to get the data for
+
+        Returns
+        -------
+        Union[
+            RadarUserData,
+            List[RadarUserData]
+        ]
+            The data of the RADAR data for the given user ids
+        """
+
+        is_only_one_var = False
+        if isinstance(user_ids, str):
+            is_only_one_var = True
+            user_ids = [user_ids]
+
+        all_user_ids = self._get_all_user_ids()
+        user_data_list = []
+
+        for user_id in user_ids:
+            if user_id in all_user_ids:
+                user_data = self._get_data_by_key(user_id)
+                if user_data is not None:
+                    user_data_list.append(user_data)
+
+        if is_only_one_var:
+            return user_data_list[0]
+        else:
+            return user_data_list
diff --git a/radarpipeline/datalib/radar_user_data.py b/radarpipeline/datalib/radar_user_data.py
new file mode 100644
index 0000000..3a0e984
--- /dev/null
+++ b/radarpipeline/datalib/radar_user_data.py
@@ -0,0 +1,84 @@
+from typing import Dict, List, Optional, Union
+
+from radarpipeline.datalib.abc import Data
+from radarpipeline.datalib.radar_variable_data import RadarVariableData
+
+
+class RadarUserData(Data):
+    """
+    Class for reading data of a single user
+    """
+
+    _data: Dict[str, RadarVariableData]
+
+    def __init__(
+        self, data: Dict[str, RadarVariableData], df_type: str = "pandas"
+    ) -> None:
+        self._data = data
+        self.df_type = df_type
+
+    def get_data(self) -> Dict[str, RadarVariableData]:
+        return self._data
+
+    def set_data(self, data: Dict[str, RadarVariableData]) -> None:
+        self._data = data
+
+    def get_data_keys(self) -> List[str]:
+        return list(self._data.keys())
+
+    def get_data_size(self) -> int:
+        return len(self._data.items())
+
+    def _get_data_by_key(self, key: str) -> Optional[RadarVariableData]:
+        return self._data.get(key, None)
+
+    def _get_all_variables(self) -> List[str]:
+        """
+        Get all variables in the data of the user
+
+        Returns
+        -------
+        List[str]
+            The list of all variables in the data of the user
+        """
+
+        return self.get_data_keys()
+
+    def get_data_by_variable(
+        self, variables: Union[str, List[str]]
+    ) -> Union[RadarVariableData, List[RadarVariableData]]:
+        """
+        Returns the data of the user for the given variables
+
+        Parameters
+        ----------
+        variables : Union[str, List[str]]
+            The variable(s) to get the data for
+
+        Returns
+        -------
+        Union[
+            RadarVariableData,
+            List[RadarVariableData],
+        ]
+            The data of the user for the given variables
+        """
+
+        is_only_one_var = False
+        if isinstance(variables, str):
+            is_only_one_var = True
+            variables = [variables]
+
+        all_variables = self._get_all_variables()
+        variable_data_list = []
+
+        for var in variables:
+            if var in all_variables:
+                var_data = self._get_data_by_key(var)
+                if var_data is not None:
+                    variable_data_list.append(var_data)
+
+        if is_only_one_var:
+            return variable_data_list[0]
+        else:
+            return variable_data_list
diff --git a/radarpipeline/datalib/radar_variable_data.py b/radarpipeline/datalib/radar_variable_data.py
new file mode 100644
index 0000000..08908db
--- /dev/null
+++ b/radarpipeline/datalib/radar_variable_data.py
@@ -0,0 +1,67 @@
+from typing import List
+
+import pandas as pd
+import pyspark.sql.functions as f
+
+from radarpipeline.datalib.abc import Data
+from radarpipeline.datatypes import DataType
+
+
+class RadarVariableData(Data):
+    """
+    Class for reading data for a single variable of a single user
+    """
+
+    _data: DataType
+
+    def __init__(self, data: DataType, df_type: str = "pandas") -> None:
+        self._data = data
+        self.df_type = df_type
+        self._preprocess_data()
+
+    def get_data(self) -> DataType:
+        return self._data
+
+    def set_data(self, data: DataType) -> None:
+        self._data = data
+
+    def get_data_keys(self) -> List[str]:
+        return list(self._data.columns)
+
+    def get_data_size(self) -> int:
+        if self.df_type == "pandas":
+            return len(self._data.index)
+        else:
+            return int(self._data.count())
+
+    def _preprocess_data(self) -> None:
+        """
+        Converts all time value columns to datetime format
+        """
+
+        if self.df_type == "spark":
+            if "value.time" in self.get_data_keys():
+                self._data = self._data.withColumn(
+                    "value.time", f.to_date(self._data["`value.time`"])
+                )
+            if "value.timeReceived" in self.get_data_keys():
+                self._data = self._data.withColumn(
+                    "value.timeReceived", f.to_date(self._data["`value.timeReceived`"])
+                )
+            if "value.dateTime" in self.get_data_keys():
+                self._data = self._data.withColumn(
+                    "value.dateTime", f.to_date(self._data["`value.dateTime`"])
+                )
+        else:
+            if "value.time" in self.get_data_keys():
+                self._data["value.time"] = pd.to_datetime(
+                    self._data["value.time"], unit="s"
+                )
+            if "value.timeReceived" in self.get_data_keys():
+                self._data["value.timeReceived"] = pd.to_datetime(
+                    self._data["value.timeReceived"], unit="s"
+                )
+            if "value.dateTime" in self.get_data_keys():
+                self._data["value.dateTime"] = pd.to_datetime(
+                    self._data["value.dateTime"], unit="s"
+                )
diff --git a/radarpipeline/datatypes/__init__.py b/radarpipeline/datatypes/__init__.py
new file mode 100644
index 0000000..cedef3f
--- /dev/null
+++ b/radarpipeline/datatypes/__init__.py
@@ -0,0 +1,1 @@
+from radarpipeline.datatypes.data_types import DataType
diff --git a/radarpipeline/datatypes/data_types.py b/radarpipeline/datatypes/data_types.py
new file mode 100644
index 0000000..069dad0
--- /dev/null
+++ b/radarpipeline/datatypes/data_types.py
@@ -0,0 +1,6 @@
+from typing import Union
+
+import pandas as pd
+import pyspark.sql as ps
+
+DataType = Union[pd.DataFrame, ps.DataFrame]
diff --git a/radarpipeline/features/__init__.py b/radarpipeline/features/__init__.py
new file mode 100644
index 0000000..2e27363
--- /dev/null
+++ b/radarpipeline/features/__init__.py
@@ -0,0 +1,2 @@
+from radarpipeline.features.feature import Feature
+from radarpipeline.features.feature_group import FeatureGroup
diff --git a/radarpipeline/features/feature.py b/radarpipeline/features/feature.py
new file mode 100644
index 0000000..49af01f
--- /dev/null
+++ b/radarpipeline/features/feature.py
@@ -0,0 +1,39 @@
+from abc import ABC, abstractmethod
+from typing import List
+
+from radarpipeline.datalib import RadarData
+from radarpipeline.datatypes.data_types import DataType
+
+
+class Feature(ABC):
+    name: str
+    description: str
+    required_input_data: List[str]
+
+    def __init__(
+        self, name: str, description: str, required_input_data: List[str]
+    ) -> None:
+        self.name = name
+        self.description = description
+        self.required_input_data = required_input_data
+
+    def __str__(self) -> str:
+        return self.name
+
+    def get_required_data(self) -> List[str]:
+        return self.required_input_data
+
+    @abstractmethod
+    def preprocess(self, data: DataType) -> DataType:
+        """
+        Preprocess the data for tje feature.
+        If there's nothing to preprocess, please return the input
+        """
+        pass
+
+    @abstractmethod
+    def calculate(self, data: DataType) -> DataType:
+        """
+        Calculates the feature.
+        """
+        pass
diff --git a/radarpipeline/features/feature_group.py b/radarpipeline/features/feature_group.py
new file mode 100644
index 0000000..f82857d
--- /dev/null
+++ b/radarpipeline/features/feature_group.py
@@ -0,0 +1,61 @@
+from abc import ABC, abstractmethod
+from typing import List, Tuple
+
+from radarpipeline.datalib import RadarData
+from radarpipeline.datatypes import DataType
+from radarpipeline.features.feature import Feature
+
+
+class FeatureGroup(ABC):
+    """
+    A class to hold a group of features.
+    """
+
+    name: str
+    description: str
+    features: List["Feature"]
+    required_input_data: List[str]
+
+    def __init__(self, name: str, description: str, features: List["Feature"]):
+        self.name = name
+        self.description = description
+        self.features = [f() for f in features]
+        self.required_input_data = self._compute_required_data()
+
+    def __str__(self):
+        return self.name
+
+    def get_required_data(self) -> List[str]:
+        return self.required_input_data
+
+    def _compute_required_data(self) -> List[str]:
+        """
+        Compute the required data for each feature in the group.
+        """
+
+        required_input_data = set()
+        for feature in self.features:
+            required_input_data.update(feature.get_required_data())
+        return list(required_input_data)
+
+    @abstractmethod
+    def preprocess(self, data: RadarData) -> DataType:
+        """
+        Preprocess the data for each feature in the group.
+        If there's nothing to process, please return the input
+        """
+        pass
+
+    def get_all_features(self, data: RadarData) -> Tuple[List[str], List[DataType]]:
+        """
+        Compute the features for each feature in the group.
+        """
+
+        feature_names = []
+        feature_values = []
+        preprocessed_data = self.preprocess(data)
+        for feature in self.features:
+            feature_names.append(feature.name)
+            preprocessed_feature = feature.preprocess(preprocessed_data)
+            feature_values.append(feature.calculate(preprocessed_feature))
+        return feature_names, feature_values
diff --git a/radarpipeline/io/__init__.py b/radarpipeline/io/__init__.py
new file mode 100644
index 0000000..028152f
--- /dev/null
+++ b/radarpipeline/io/__init__.py
@@ -0,0 +1,3 @@
+from radarpipeline.io.abc import DataReader, SchemaReader
+from radarpipeline.io.reader import AvroSchemaReader, SparkCSVDataReader
+from radarpipeline.io.writer import *
diff --git a/radarpipeline/io/abc.py b/radarpipeline/io/abc.py
new file mode 100644
index 0000000..0f8f561
--- /dev/null
+++ b/radarpipeline/io/abc.py
@@ -0,0 +1,57 @@
+from abc import ABC, abstractmethod
+from typing import Dict
+
+from radarpipeline.datalib import Data
+from radarpipeline.datatypes.data_types import DataType
+
+
+class DataReader(ABC):
+    """
+    Abstract class for reading the RADAR data
+    """
+
+    config: Dict
+
+    def __init__(self, config: Dict) -> None:
+        self.config = config
+
+    @abstractmethod
+    def read_data(self) -> Data:
+        pass
+
+
+class SchemaReader(ABC):
+    """
+    Abstract class for reading the RADAR data schema
+    """
+
+    schema_dir: str
+    schema_file: str
+
+    def __init__(self, schema_dir: str) -> None:
+        self.schema_dir = schema_dir
+
+    @abstractmethod
+    def is_schema_present() -> bool:
+        pass
+
+    @abstractmethod
+    def get_schema(self):
+        pass
+
+
+class DataWriter(ABC):
+    """
+    Abstract class for writing the RADAR data
+    """
+
+    features: Dict[str, DataType]
+    output_dir: str
+
+    def __init__(self, features: Dict[str, DataType], output_dir: str) -> None:
+        self.features = features
+        self.output_dir = output_dir
+
+    @abstractmethod
+    def write_data(self) -> None:
+        pass
diff --git a/radarpipeline/io/reader.py b/radarpipeline/io/reader.py
new file mode 100644
index 0000000..4531c26
--- /dev/null
+++ b/radarpipeline/io/reader.py
@@ -0,0 +1,264 @@
+import json
+import logging
+import os
+from typing import Any, Dict, List, Optional, Union
+
+import pyspark.sql as ps
+from pyspark.sql import SparkSession
+from pyspark.sql.types import (
+    BooleanType,
+    DoubleType,
+    FloatType,
+    IntegerType,
+    LongType,
+    StringType,
+    StructField,
+    StructType,
+)
+
+from radarpipeline.common import constants
+from radarpipeline.datalib import RadarData, RadarUserData, RadarVariableData
+from radarpipeline.io.abc import DataReader, SchemaReader
+
+logger = logging.getLogger(__name__)
+
+
+class SparkCSVDataReader(DataReader):
+    """
+    Read CSV data from local directory using pySpark
+    """
+
+    def __init__(self, config: Dict, required_data: List[str], df_type: str = "pandas"):
+        super().__init__(config)
+        self.required_data = required_data
+        self.df_type = df_type
+        self.source_path = self.config.get("local_directory", "")
+        self.spark = self._initialize_spark_session()
+
+    def _initialize_spark_session(self) -> ps.SparkSession:
+        """
+        Initializes and returns a SparkSession
+
+        Returns
+        -------
+        SparkSession
+            A SparkSession object
+        """
+        spark = (
+            SparkSession.builder.master("local").appName("radarpipeline").getOrCreate()
+        )
+
+        # Enable Apache Arrow for optimizations in Spark to Pandas conversion
+        spark.conf.set("spark.sql.execution.arrow.pyspark.enabled", "true")
+        # Fallback to use non-Arrow conversion in case of errors
+        spark.conf.set("spark.sql.execution.arrow.pyspark.fallback.enabled", "true")
+        # For further reading: https://spark.apache.org/docs/3.0.1/sql-pyspark-pandas-with-arrow.html
+
+        spark.sparkContext.setLogLevel("ERROR")
+        logger.info("Spark Session created")
+
+        return spark
+
+    def read_data(self) -> RadarData:
+        """
+        Reads RADAR data from local CSV files
+
+        Returns
+        -------
+        RadarData
+            A RadarData object containing all the read data
+        """
+
+        user_data_dict = {}
+
+        for uid in os.listdir(self.source_path):
+            # Skip hidden files
+            if uid[0] == ".":
+                continue
+
+            logger.info(f"Reading data for user: {uid}")
+
+            variable_data_dict = {}
+
+            for dirname in self.required_data:
+                if dirname not in os.listdir(os.path.join(self.source_path, uid)):
+                    continue
+
+                logger.info(f"Reading data for variable: {dirname}")
+
+                absolute_dirname = os.path.abspath(
+                    os.path.join(self.source_path, uid, dirname)
+                )
+
+                data_files = [
+                    os.path.join(absolute_dirname, f)
+                    for f in os.listdir(absolute_dirname)
+                    if f.endswith(".csv.gz")
+                ]
+
+                schema = None
+                schema_reader = AvroSchemaReader(absolute_dirname)
+
+                if schema_reader.is_schema_present():
+                    logger.info("Schema found")
+                    schema = schema_reader.get_schema()
+                else:
+                    logger.info("Schema not found, inferring from data file")
+
+                variable_data = self._read_variable_data_files(data_files, schema)
+
+                if variable_data.get_data_size() > 0:
+                    variable_data_dict[dirname] = variable_data
+
+            user_data_dict[uid] = RadarUserData(variable_data_dict, self.df_type)
+
+        radar_data = RadarData(user_data_dict, self.df_type)
+        return radar_data
+
+    def _read_variable_data_files(
+        self,
+        data_files: List[str],
+        schema: Optional[StructType] = None,
+    ) -> RadarVariableData:
+        """
+        Reads data from a list of data files and returns a RadarVariableData object
+
+        Parameters
+        ----------
+        data_files : List[str]
+            List of data files to read
+        schema : Optional[StructType]
+            Schema to use for optimized reading
+
+        Returns
+        -------
+        RadarVariableData
+            A RadarVariableData object containing all the read data
+        """
+
+        if schema:
+            df = self.spark.read.load(
+                data_files,
+                format="csv",
+                header=True,
+                schema=schema,
+                inferSchema="false",
+                encoding=constants.ENCODING,
+            )
+        else:
+            df = self.spark.read.load(
+                data_files,
+                format="csv",
+                header=True,
+                inferSchema="true",
+                encoding=constants.ENCODING,
+            )
+
+        if self.df_type == "pandas":
+            df = df.toPandas()
+
+        # To print the dataframe stats for cross-checking
+        # print(df.show(5))
+
+        variable_data = RadarVariableData(df, self.df_type)
+
+        return variable_data
+
+
+class AvroSchemaReader(SchemaReader):
+    """
+    Reads schema from local directory
+    """
+
+    data_type_mapping = {
+        "boolean": BooleanType(),
+        "int": IntegerType(),
+        "long": LongType(),
+        "float": FloatType(),
+        "double": DoubleType(),
+        "string": StringType(),
+        "enum": StringType(),
+    }
+
+    def __init__(self, schema_dir: str) -> None:
+        super().__init__(schema_dir)
+
+    def is_schema_present(self) -> bool:
+        """
+        Checks if schema is present in local directory
+
+        Returns
+        -------
+        bool
+            True if schema is present, False otherwise
+        """
+
+        schema_dir_base = os.path.basename(self.schema_dir)
+        self.schema_file = os.path.join(
+            self.schema_dir, f"schema-{schema_dir_base}.json"
+        )
+
+        if os.path.exists(self.schema_file):
+            self.schema_file = self.schema_file
+            return True
+
+        return False
+
+    def get_schema(self) -> StructType:
+        """
+        Reads schema from local directory
+
+        Returns
+        -------
+        StructType
+            A StructType object defining the schema for pySpark
+        """
+
+        schema_fields = []
+        schema_dict = json.load(
+            open(
+                os.path.join(self.schema_dir, self.schema_file),
+                "r",
+                encoding=constants.ENCODING,
+            )
+        )
+
+        key_dict = schema_dict["fields"][0]
+        value_dict = schema_dict["fields"][1]
+
+        for key in key_dict["type"]["fields"]:
+            name = key["name"]
+            schema_fields.append(StructField(f"key.{name}", StringType()))
+
+        for value in value_dict["type"]["fields"]:
+            name = value["name"]
+            typ = value["type"]
+            if type(typ) is dict and "type" in typ:
+                typ = typ["type"]
+            field_type = self._get_field(typ)
+            schema_fields.append(StructField(f"value.{name}", field_type, True))
+
+        schema = StructType(schema_fields)
+        return schema
+
+    def _get_field(self, data_type: Union[str, Dict]) -> Any:
+        """
+        Returns a Spark data type for a given data type
+
+        Parameters
+        ----------
+        data_type : Union[str, Dict]
+            Data type to convert to a Spark data type
+
+        Returns
+        -------
+        Any
+            A Spark data type
+        """
+
+        if type(data_type) is str and data_type in self.data_type_mapping:
+            return self.data_type_mapping[data_type]
+        elif type(data_type) is dict:
+            return StringType()
+        else:
+            raise ValueError(f"Unknown data type: {data_type}")
diff --git a/radarpipeline/io/writer.py b/radarpipeline/io/writer.py
new file mode 100644
index 0000000..3897db0
--- /dev/null
+++ b/radarpipeline/io/writer.py
@@ -0,0 +1,90 @@
+import logging
+import os
+from typing import Dict, Optional
+
+import pandas as pd
+import pyspark.sql as ps
+
+from radarpipeline.common import constants, utils
+from radarpipeline.datatypes.data_types import DataType
+from radarpipeline.io.abc import DataWriter
+
+logger = logging.getLogger(__name__)
+
+
+class SparkDataWriter(DataWriter):
+    """
+    Writes the data to a local directory using pySpark
+    """
+
+    features: Dict[str, ps.DataFrame]
+    num_files: Optional[int] = None
+    compression: str
+
+    def __init__(
+        self,
+        features: Dict[str, DataType],
+        output_dir: str,
+        num_files: Optional[int] = 1,
+        compress: bool = False,
+    ) -> None:
+        super().__init__(features, output_dir)
+        self.num_files = num_files
+        self.compression = "gzip" if compress is True else "none"
+
+    def write_data(self) -> None:
+        for feature_name, feature_df in self.features.items():
+            folder_name = utils.pascal_to_snake_case(feature_name)
+            folder_path = os.path.join(self.output_dir, folder_name)
+            try:
+                feature_df.write.csv(
+                    path=folder_path,
+                    header=True,
+                    sep=constants.CSV_DELIMITER,
+                    encoding=constants.ENCODING,
+                    compression=self.compression,
+                    lineSep=constants.LINESEP,
+                )
+            except Exception as e:
+                logger.error(
+                    f"Error writing data to file {folder_path}: {e}", exc_info=True
+                )
+            logger.info(f"Feature {feature_name} data exported to {folder_path}")
+
+
+class PandasDataWriter(DataWriter):
+    """
+    Writes the data to a local directory using pandas
+    """
+
+    compression: str
+    features: Dict[str, pd.DataFrame]
+
+    def __init__(
+        self,
+        features: Dict[str, DataType],
+        output_dir: str,
+        compress: bool = False,
+    ) -> None:
+        super().__init__(features, output_dir)
+        self.compression = "gzip" if compress is True else "infer"
+
+    def write_data(self) -> None:
+        for feature_name, feature_df in self.features.items():
+            file_name = utils.pascal_to_snake_case(feature_name) + ".csv"
+            file_name += ".gz" if self.compression == "gzip" else ""
+            file_path = os.path.join(self.output_dir, file_name)
+            try:
+                feature_df.to_csv(
+                    file_path,
+                    index=False,
+                    sep=constants.CSV_DELIMITER,
+                    encoding=constants.ENCODING,
+                    line_terminator=constants.LINESEP,
+                    compression=self.compression,
+                )
+            except Exception as e:
+                logger.error(
+                    f"Error writing data to file {file_path}: {e}", exc_info=True
+                )
+            logger.info(f"Feature {feature_name} data exported to {file_path}")
diff --git a/radarpipeline/project/__init__.py b/radarpipeline/project/__init__.py
new file mode 100644
index 0000000..e6bb37c
--- /dev/null
+++ b/radarpipeline/project/__init__.py
@@ -0,0 +1,1 @@
+from radarpipeline.project.project import Project
diff --git a/radarpipeline/project/project.py b/radarpipeline/project/project.py
new file mode 100644
index 0000000..8fc7f53
--- /dev/null
+++ b/radarpipeline/project/project.py
@@ -0,0 +1,437 @@
+import importlib
+import inspect
+import logging
+import os
+import pathlib
+import sys
+from typing import Any, Dict, List, Union
+
+from git.repo import Repo
+
+from radarpipeline.common import utils
+from radarpipeline.features import Feature, FeatureGroup
+from radarpipeline.io import PandasDataWriter, SparkCSVDataReader, SparkDataWriter
+
+logger = logging.getLogger(__name__)
+
+
+class Project:
+    def __init__(self, input_data: Union[str, dict]) -> None:
+        """
+        Initialize the project with the data from the config file, or a config dict
+
+        Parameters
+        ---------
+        input_data: Union[str, dict]
+            Path to the config file or a dict containing the config
+        """
+
+        self.valid_input_formats = ["csv", "csv.gz"]
+        self.valid_output_formats = ["csv"]
+        self.input_data = input_data
+        self.feature_path = os.path.abspath(
+            os.path.join("radarpipeline", "features", "features")
+        )
+        self.features = {}
+        self.config = self._get_config()
+        self._validate_config()
+        self.feature_groups = self._get_feature_groups()
+        self.total_required_data = self._get_total_required_data()
+
+    def _get_config(self) -> Dict[str, Any]:
+        """
+        Reads the config and returns it as a dictionary
+
+        Returns
+        -------
+        Dict[str, Any]
+            Config dictionary
+        """
+
+        if isinstance(self.input_data, str):
+            config = utils.read_yaml(self.input_data)
+        elif isinstance(self.input_data, dict):
+            config = self.input_data
+        else:
+            raise ValueError("Wrong input data type. Should be yaml file path or dict.")
+
+        logger.info("Config file read successfully")
+        return config
+
+    def _validate_config(self) -> None:
+        """
+        Validates the config to check if all the keys are present
+        """
+
+        required_keys = ["project", "input_data", "features", "output_data"]
+        for key in required_keys:
+            if key not in self.config:
+                raise ValueError(f"Key not present in the config: {key}")
+
+        self._validate_input()
+        self._validate_configurations()
+        self._validate_features()
+        self._validate_output()
+
+        logger.info("Config file validated successfully")
+
+    def _validate_input(self) -> None:
+        """
+        Validates the input data config
+        """
+
+        if self.config["input_data"]["data_location"] == "sftp":
+            sftp_config_keys = [
+                "sftp_host",
+                "sftp_username",
+                "sftp_directory",
+                "sftp_private_key",
+            ]
+            for key in sftp_config_keys:
+                if key not in self.config["input_data"]:
+                    raise ValueError(f"Key not present in the config: {key}")
+
+        elif self.config["input_data"]["data_location"] == "local":
+            if "local_directory" not in self.config["input_data"]:
+                raise ValueError("Key not present in the config: local_directory")
+            else:
+                # Check if local_directory is absolute path. If not, then set it.
+                local_directory = self.config["input_data"]["local_directory"]
+                local_directory = self._get_absolute_path(local_directory)
+                if not os.path.exists(local_directory):
+                    raise ValueError(f"Path does not exist: {local_directory}")
+                self.config["input_data"]["local_directory"] = local_directory
+
+            # Raise error if data_format it not valid input formats
+            if self.config["input_data"]["data_format"] not in self.valid_input_formats:
+                raise ValueError("Invalid value for key in input_data: data_format")
+
+        elif self.config["input_data"]["data_location"] == "mock":
+            self._update_mock_data()
+
+        else:
+            raise ValueError("Invalid value for the key: data_location")
+
+    def _validate_configurations(self) -> None:
+        """
+        Validates the configurations config
+        """
+
+        if "configurations" not in self.config:
+            self.config["configurations"] = {}
+
+        valid_df_types = ["pandas", "spark"]
+
+        if "df_type" not in self.config["configurations"]:
+            self.config["configurations"]["df_type"] = "pandas"
+        elif self.config["configurations"]["df_type"] not in valid_df_types:
+            raise ValueError("Invalid value for the key: df_type")
+
+    def _validate_features(self) -> None:
+        """
+        Validates the features config
+        """
+
+        if len(self.config["features"]) == 0:
+            raise ValueError("features array cannot be empty")
+
+        for index, feature in enumerate(self.config["features"]):
+            if "location" not in feature:
+                raise ValueError(
+                    f"Key not present in the config: location at index {index}"
+                )
+
+            if "feature_groups" not in feature:
+                raise ValueError(
+                    f"Key not present in the config: feature_groups at index {index}"
+                )
+
+            if len(feature["feature_groups"]) == 0:
+                raise ValueError(
+                    f"feature_groups array cannot be empty at index {index}"
+                )
+
+            feature_location = feature["location"]
+
+            if utils.is_valid_github_path(feature_location):
+                repo_name = utils.get_repo_name_from_url(feature_location)
+                cache_dir = os.path.join(
+                    os.path.expanduser("~"), ".cache", "radarpipeline", repo_name
+                )
+
+                if os.path.exists(cache_dir):
+                    repo = Repo(cache_dir)
+                    repo.git.reset("--hard")
+                    repo.git.clean("-xdf")
+                    repo.git.checkout(repo.active_branch)
+                    repo.remotes.origin.pull(repo.active_branch.name)
+                else:
+                    Repo.clone_from(feature_location, cache_dir)
+
+                feature_location = cache_dir
+                logger.info(f"Using feature from cache location: {feature_location}")
+            else:
+                feature_location = os.path.expanduser(feature_location)
+                if not os.path.isdir(feature_location):
+                    raise ValueError(f"Invalid feature location: {feature_location}")
+                logger.info(f"Using feature from local path: {feature_location}")
+
+            self.config["features"][index]["location"] = feature_location
+
+    def _validate_output(self) -> None:
+        """
+        Validates the output data config
+        """
+
+        if self.config["output_data"]["output_location"] == "postgres":
+            postgres_config_keys = [
+                "postgres_host",
+                "postgres_username",
+                "postgres_database",
+                "postgres_password",
+                "postgres_table",
+            ]
+            for key in postgres_config_keys:
+                if key not in self.config["output_data"]:
+                    raise ValueError(f"Key not present in the config: {key}")
+
+        elif self.config["output_data"]["output_location"] == "local":
+            if "local_directory" not in self.config["output_data"]:
+                raise ValueError("Key not present in the config: local_directory")
+            else:
+                # Check if local_directory is absolute path. If not, then set it.
+                local_directory = self.config["output_data"]["local_directory"]
+                local_directory = self._get_absolute_path(local_directory)
+                if not os.path.exists(local_directory):
+                    os.makedirs(local_directory, exist_ok=True)
+                self.config["output_data"]["output_directory"] = local_directory
+
+            # Raise error if data_format it not valid output formats
+            if (
+                self.config["output_data"]["data_format"]
+                not in self.valid_output_formats
+            ):
+                raise ValueError("Invalid value for key in output_data: data_format")
+
+            if "compress" not in self.config["output_data"]:
+                self.config["output_data"]["compress"] = False
+            if self.config["output_data"]["compress"] == "true":
+                self.config["output_data"]["compress"] = True
+
+        elif self.config["output_data"]["output_location"] == "mock":
+            pass
+
+        else:
+            raise ValueError("Key not present in the config: output_location")
+
+    def _get_absolute_path(self, path: str) -> str:
+        """
+        Returns the absolute path of the path
+
+        Parameters
+        ----------
+        path: str
+            Path to be converted to absolute path
+
+        Returns
+        -------
+        str
+            Absolute path of the path
+        """
+
+        if not os.path.isabs(path):
+            pipeline_dir = pathlib.Path(__file__).parent.parent.parent.resolve()
+            path = os.path.join(pipeline_dir, path)
+        return path
+
+    def _update_mock_data(self) -> None:
+        """
+        Updates the mock data submodule of the project
+        """
+
+        repo = Repo(
+            os.path.dirname(os.path.abspath(__file__)),
+            search_parent_directories=True,
+        )
+        sms = repo.submodules
+        try:
+            mock_data_submodule = sms["mockdata"]
+        except IndexError:
+            raise ValueError("mockdata submodule not found in the repository")
+        if not (mock_data_submodule.exists() and mock_data_submodule.module_exists()):
+            logger.info("Mock data submodule not found. Cloning it...")
+            repo.git.submodule("update", "--init", "--recursive")
+            logger.info("Mock data input cloned")
+
+    def _get_feature_groups(self) -> List[FeatureGroup]:
+        """
+        Get all the feature groups from the config
+
+        Returns
+        -------
+        List[FeatureGroup]
+            List of all the feature groups
+        """
+
+        features = self.config.get("features", [])
+        feature_groups = set()
+
+        if "mock" in features:
+            logger.info("Using mock features")
+        else:
+            for feature in features:
+                feature_groups.update(self._get_feature_group(feature))
+            feature_group_names = [
+                feature_group.name for feature_group in feature_groups
+            ]
+            logger.info(f"Number of feature groups found: {len(list(feature_groups))}")
+            logger.info(f"List of Feature groups: {feature_group_names}")
+
+        return list(feature_groups)
+
+    def _get_feature_group(self, feature: dict) -> List[FeatureGroup]:
+        """
+        Get feature group from filepath or name
+
+        Parameters
+        ----------
+        feature: dict
+            Feature to get from the specified location
+
+        Returns
+        -------
+        List[FeatureGroup]
+            List of feature group(s)
+        """
+
+        feature_group_classes = set()
+
+        feature_location = feature["location"]
+        req_feature_groups = feature["feature_groups"]
+
+        # Get feature class from __init__.py file in feature_location
+        all_feature_group_classes = self._get_feature_groups_from_filepath(
+            feature_location
+        )
+
+        # Search feature_name in all_feature_classes
+        for feature_group_class in all_feature_group_classes:
+            if feature_group_class.name in req_feature_groups:
+                feature_group_classes.add(feature_group_class)
+
+        if len(feature_group_classes) == 0:
+            raise ValueError(f"Feature not found: {feature}")
+
+        return list(feature_group_classes)
+
+    def _get_feature_groups_from_filepath(self, feature_location: str) -> List[Any]:
+        """
+        Gets the feature group classes from the filepath
+
+        Parameters
+        ----------
+        feature_location: str
+            Filepath of the feature group classes
+
+        Returns
+        -------
+        List[Any]
+            List of feature group classes
+        """
+
+        parent_dir = os.path.dirname(feature_location)
+        base_name = os.path.basename(feature_location)
+
+        sys.path.insert(1, parent_dir)
+
+        # Import the feature module
+        feature_module = importlib.import_module(base_name)
+
+        # Get the feature group class
+        feature_groups = set()
+        for name, obj in inspect.getmembers(feature_module, inspect.isclass):
+            if obj != Feature and obj != FeatureGroup:
+                instance = obj()
+                if isinstance(instance, FeatureGroup):
+                    feature_groups.add(instance)
+
+        return list(feature_groups)
+
+    def _get_total_required_data(self) -> List[str]:
+        """
+        Get the total required data fromm the feature groups
+
+        Returns
+        -------
+        List[str]
+            List of all the required data
+        """
+
+        total_required_data = set()
+        for feature_group in self.feature_groups:
+            total_required_data.update(feature_group.get_required_data())
+
+        logger.info(f"Total required data: {total_required_data}")
+
+        return list(total_required_data)
+
+    def fetch_data(self) -> None:
+        """
+        Fetches the data from the data source
+        """
+
+        if self.config["input_data"]["data_location"] == "local":
+            if self.config["input_data"]["data_format"] in self.valid_input_formats:
+                self.data = SparkCSVDataReader(
+                    self.config["input_data"],
+                    self.total_required_data,
+                    self.config["configurations"]["df_type"],
+                ).read_data()
+            else:
+                raise ValueError("Wrong data format")
+
+        elif self.config["input_data"]["data_location"] == "mock":
+            mock_config_input = {
+                "local_directory": os.path.join("mockdata", "mockdata")
+            }
+            mock_required_data = [
+                "android_phone_battery_level",
+                "android_phone_step_count",
+            ]
+            self.data = SparkCSVDataReader(
+                mock_config_input, mock_required_data
+            ).read_data()
+
+        else:
+            raise ValueError("Wrong data location")
+
+    def compute_features(self) -> None:
+        """
+        Computes the features from the ingested data
+        """
+
+        for feature_group in self.feature_groups:
+            feature_names, feature_values = feature_group.get_all_features(self.data)
+            for feature_name, feature_value in zip(feature_names, feature_values):
+                self.features[feature_name] = feature_value
+
+    def export_data(self) -> None:
+        """
+        Exports the computed features to the specified location
+        """
+
+        if self.config["configurations"]["df_type"] == "pandas":
+            writer = PandasDataWriter(
+                self.features,
+                self.config["output_data"]["local_directory"],
+                self.config["output_data"]["compress"],
+            )
+        elif self.config["configurations"]["df_type"] == "spark":
+            writer = SparkDataWriter(
+                self.features,
+                self.config["output_data"]["local_directory"],
+                self.config["output_data"]["compress"],
+            )
+        else:
+            raise ValueError("Wrong df_type")
+        writer.write_data()
diff --git a/radarpipeline/radarpipeline.py b/radarpipeline/radarpipeline.py
new file mode 100644
index 0000000..c4b4073
--- /dev/null
+++ b/radarpipeline/radarpipeline.py
@@ -0,0 +1,37 @@
+import logging
+import sys
+import traceback
+
+from radarpipeline import Project
+from radarpipeline.common.logger import logger_init
+
+logger_init()
+
+logger = logging.getLogger(__name__)
+
+
+def run():
+    """
+    Pipeline entry point.
+    """
+    try:
+        logger.info("Starting the pipeline run...")
+        logger.info("Reading and Validating the configuration file...")
+        project = Project(input_data="config.yaml")
+        logger.info("Fetching the data...")
+        project.fetch_data()
+        logger.info("Computing the features...")
+        project.compute_features()
+        logger.info("Exporting the features data...")
+        project.export_data()
+        logger.info("Pipeline run completed successfully")
+    except KeyboardInterrupt:
+        logger.info("Pipeline run interrupted by user")
+        sys.exit(0)
+    except Exception:
+        logger.info(traceback.format_exc())
+        sys.exit(1)
+
+
+if __name__ == "__main__":
+    run()
diff --git a/radarpipeline/visualization/__init__.py b/radarpipeline/visualization/__init__.py
new file mode 100644
index 0000000..e69de29
diff --git a/requirements.txt b/requirements.txt
index c9133d6..c02ac0a 100644
--- a/requirements.txt
+++ b/requirements.txt
@@ -1,1 +1,12 @@
-twine
\ No newline at end of file
+twine==3.8.0
+pyYaml==6.0
+numpy==1.22.3
+pandas==1.4.1
+matplotlib==3.5.1
+seaborn==0.11.2
+scipy==1.8.0
+pysftp==0.2.9
+dask[complete]==2022.5.1
+pyspark[sql]==3.3.0
+GitPython==3.1.27
+pre-commit
diff --git a/setup.py b/setup.py
index 8e6978c..7b8af11 100644
--- a/setup.py
+++ b/setup.py
@@ -1,19 +1,20 @@
 #!/usr/bin/env python3
-from setuptools import setup, find_packages
+from setuptools import find_packages, setup
 
-setup(name='radarpipeline',
-      version='0.0.1',
-      description='A python feature generation and visualzation package use with RADAR project data.',
-      url='https://github.com/RADAR-base/radar-pipeline',
-      author='Heet Sankesara',
-      author_email='heet.sankesara@kcl.ac.uk',
-      packages=find_packages(),
-      install_requires=[],
-      test_suite='tests',
-      include_package_data=True,
-      classifiers=[
-          "Programming Language :: Python :: 3",
-          "License :: OSI Approved :: Apache Software License",
-          "Intended Audience :: Science/Research"
-          ]
-)
\ No newline at end of file
+setup(
+    name="radarpipeline",
+    version="0.0.1",
+    description="A python feature generation and visualization package use with RADAR project data.",
+    url="https://github.com/RADAR-base/radarpipeline",
+    author="Heet Sankesara",
+    author_email="heet.sankesara@kcl.ac.uk",
+    packages=find_packages(),
+    install_requires=[],
+    test_suite="tests",
+    include_package_data=True,
+    classifiers=[
+        "Programming Language :: Python :: 3",
+        "License :: OSI Approved :: Apache Software License",
+        "Intended Audience :: Science/Research",
+    ],
+)


Test patch:
diff --git a/tests/tests_datalib/test_radar_file_data.py b/tests/tests_datalib/test_radar_file_data.py
new file mode 100644
index 0000000..e69de29
diff --git a/tests/tests_datalib/test_radar_user_data.py b/tests/tests_datalib/test_radar_user_data.py
new file mode 100644
index 0000000..e69de29
diff --git a/tests/tests_datalib/test_radar_variable_data.py b/tests/tests_datalib/test_radar_variable_data.py
new file mode 100644
index 0000000..e69de29
diff --git a/tests/tests_datalib/test_radardata.py b/tests/tests_datalib/test_radardata.py
new file mode 100644
index 0000000..e69de29
diff --git a/tests/tests_features/test_feature.py b/tests/tests_features/test_feature.py
new file mode 100644
index 0000000..e69de29
diff --git a/tests/tests_features/test_feature_group.py b/tests/tests_features/test_feature_group.py
new file mode 100644
index 0000000..e69de29
diff --git a/tests/tests_io/test_reader.py b/tests/tests_io/test_reader.py
new file mode 100644
index 0000000..e69de29
diff --git a/tests/tests_io/test_writer.py b/tests/tests_io/test_writer.py
new file mode 100644
index 0000000..e69de29
diff --git a/tests/tests_project/test_project.py b/tests/tests_project/test_project.py
new file mode 100644
index 0000000..e69de29