data ingestion added

config/config.yaml

@@ -0,0 +1,7 @@
+artifacts_root: artifacts
+
+data_ingestion:
+  root_dir: artifacts/data_ingestion
+  source_dir: C:\\mlops project\\archive.zip
+  local_data_file : artifacts/data_ingestion/archive.zip
+  unzip_dir : artifacts/data_ingestion

main.py

@@ -0,0 +1,13 @@
+from src.imagecolorization.pipeline.stage01_data_ingestion import DataIngestionPipeline
+from src.imagecolorization.logging import logger
+
+STAGE_NAME = 'Data Ingestion Config'
+
+try:
+   logger.info(f">>>>>> stage {STAGE_NAME} started <<<<<<") 
+   data_ingestion = DataIngestionPipeline()
+   data_ingestion.main()
+   logger.info(f">>>>>> stage {STAGE_NAME} completed <<<<<<\n\nx==========x")
+except Exception as e:
+        logger.exception(e)
+        raise e

params.yaml

@@ -0,0 +1 @@
+key: val

research/data_ingestion.ipynb

@@ -0,0 +1,208 @@
+{
+ "cells": [
+  {
+   "cell_type": "code",
+   "execution_count": 1,
+   "metadata": {},
+   "outputs": [],
+   "source": [
+    "import os\n",
+    "os.chdir('../')"
+   ]
+  },
+  {
+   "cell_type": "code",
+   "execution_count": 2,
+   "metadata": {},
+   "outputs": [
+    {
+     "data": {
+      "text/plain": [
+       "'c:\\\\mlops project\\\\image-colorization-mlops'"
+      ]
+     },
+     "execution_count": 2,
+     "metadata": {},
+     "output_type": "execute_result"
+    }
+   ],
+   "source": [
+    "%pwd"
+   ]
+  },
+  {
+   "cell_type": "code",
+   "execution_count": 3,
+   "metadata": {},
+   "outputs": [],
+   "source": [
+    "from dataclasses import dataclass\n",
+    "from pathlib import Path\n",
+    "\n",
+    "@dataclass(frozen=True)\n",
+    "class DataIngestionConfig:\n",
+    "    root_dir : Path\n",
+    "    source_dir : Path\n",
+    "    local_data_file: Path\n",
+    "    unzip_dir : Path"
+   ]
+  },
+  {
+   "cell_type": "code",
+   "execution_count": 4,
+   "metadata": {},
+   "outputs": [],
+   "source": [
+    "from src.imagecolorization.constants import *\n",
+    "from src.imagecolorization.utils.common import read_yaml, create_directories\n",
+    "\n",
+    "class ConfigurationManager:\n",
+    "    def __init__(\n",
+    "        self,\n",
+    "        config_filepath = CONFIG_FILE_PATH,\n",
+    "        params_filepath = PARAMS_FILE_PATH):\n",
+    "\n",
+    "        self.config = read_yaml(config_filepath)\n",
+    "        self.params = read_yaml(params_filepath)\n",
+    "\n",
+    "        create_directories([self.config.artifacts_root])\n",
+    "\n",
+    "    \n",
+    "\n",
+    "    def get_data_ingestion_config(self) -> DataIngestionConfig:\n",
+    "        config = self.config.data_ingestion\n",
+    "\n",
+    "        create_directories([config.root_dir])\n",
+    "\n",
+    "        data_ingestion_config = DataIngestionConfig(\n",
+    "            root_dir=config.root_dir,\n",
+    "            source_dir=config.source_dir,\n",
+    "            local_data_file=config.local_data_file,\n",
+    "            unzip_dir=config.unzip_dir \n",
+    "        )\n",
+    "\n",
+    "        return data_ingestion_config\n",
+    "        \n",
+    "    \n",
+    "    "
+   ]
+  },
+  {
+   "cell_type": "code",
+   "execution_count": 5,
+   "metadata": {},
+   "outputs": [],
+   "source": [
+    "import os\n",
+    "import zipfile\n",
+    "from src.imagecolorization.logging import logger\n",
+    "from tqdm.notebook import tqdm\n",
+    "from dataclasses import replace\n",
+    "\n",
+    "class DataIngestion:\n",
+    "    def __init__(self, config: DataIngestionConfig):\n",
+    "        self.config = config\n",
+    "    \n",
+    "    def load_file(self):\n",
+    "        if os.path.exists(self.config.source_dir):\n",
+    "            self.config = replace(self.config, local_data_file=self.config.source_dir)\n",
+    "            logger.info(f'File Found at: {self.config.local_data_file}')\n",
+    "        else:\n",
+    "            logger.info(f\"File not found at {self.config.source_dir}\")\n",
+    "            raise FileNotFoundError(f'No file found at {self.config.source_dir}')\n",
+    "        \n",
+    "        \n",
+    "        \n",
+    "    def extract_zip_file(self):\n",
+    "        unzip_path = self.config.unzip_dir\n",
+    "        os.makedirs(unzip_path, exist_ok=True)\n",
+    "        \n",
+    "        # open the zip file\n",
+    "        with zipfile.ZipFile(self.config.local_data_file, 'r') as zip_ref:\n",
+    "            total_files = len(zip_ref.infolist())\n",
+    "            \n",
+    "            for file in tqdm(iterable=zip_ref.infolist(), total=total_files, desc='Extracting files'):\n",
+    "                zip_ref.extract(member=file, path=unzip_path)\n",
+    "                \n",
+    "            logger.info(f'Extacted {self.config.local_data_file} to {unzip_path}')"
+   ]
+  },
+  {
+   "cell_type": "code",
+   "execution_count": 6,
+   "metadata": {},
+   "outputs": [
+    {
+     "name": "stdout",
+     "output_type": "stream",
+     "text": [
+      "[2024-08-18 02:08:07,443: INFO: common: yaml file: config\\config.yaml loaded successfully]\n",
+      "[2024-08-18 02:08:07,444: INFO: common: yaml file: params.yaml loaded successfully]\n",
+      "[2024-08-18 02:08:07,445: INFO: common: created directory at: artifacts]\n",
+      "[2024-08-18 02:08:07,446: INFO: common: created directory at: artifacts/data_ingestion]\n",
+      "[2024-08-18 02:08:07,446: INFO: 2749353352: File Found at: C:\\\\mlops project\\\\archive.zip]\n"
+     ]
+    },
+    {
+     "data": {
+      "application/vnd.jupyter.widget-view+json": {
+       "model_id": "b91326c533ac4f588a5224910549cd65",
+       "version_major": 2,
+       "version_minor": 0
+      },
+      "text/plain": [
+       "Extracting files:   0%|          | 0/5 [00:00<?, ?it/s]"
+      ]
+     },
+     "metadata": {},
+     "output_type": "display_data"
+    },
+    {
+     "name": "stdout",
+     "output_type": "stream",
+     "text": [
+      "[2024-08-18 02:08:34,311: INFO: 2749353352: Extacted C:\\\\mlops project\\\\archive.zip to artifacts/data_ingestion]\n"
+     ]
+    }
+   ],
+   "source": [
+    "try:\n",
+    "    config = ConfigurationManager()\n",
+    "    data_ingestion_config = config.get_data_ingestion_config()\n",
+    "    data_ingestion = DataIngestion(config=data_ingestion_config)\n",
+    "    data_ingestion.load_file()\n",
+    "    data_ingestion.extract_zip_file()\n",
+    "except Exception as e:\n",
+    "    raise e"
+   ]
+  },
+  {
+   "cell_type": "code",
+   "execution_count": null,
+   "metadata": {},
+   "outputs": [],
+   "source": []
+  }
+ ],
+ "metadata": {
+  "kernelspec": {
+   "display_name": "Python 3",
+   "language": "python",
+   "name": "python3"
+  },
+  "language_info": {
+   "codemirror_mode": {
+    "name": "ipython",
+    "version": 3
+   },
+   "file_extension": ".py",
+   "mimetype": "text/x-python",
+   "name": "python",
+   "nbconvert_exporter": "python",
+   "pygments_lexer": "ipython3",
+   "version": "3.11.0"
+  }
+ },
+ "nbformat": 4,
+ "nbformat_minor": 2
+}

src/imagecolorization/config/configuration.py

@@ -0,0 +1,32 @@
+from src.imagecolorization.constants import *
+from src.imagecolorization.utils.common import read_yaml, create_directories
+from src.imagecolorization.entity.config_entity import DataIngestionConfig
+class ConfigurationManager:
+    def __init__(
+        self,
+        config_filepath = CONFIG_FILE_PATH,
+        params_filepath = PARAMS_FILE_PATH):
+
+        self.config = read_yaml(config_filepath)
+        self.params = read_yaml(params_filepath)
+
+        create_directories([self.config.artifacts_root])
+
+    
+
+    def get_data_ingestion_config(self) -> DataIngestionConfig:
+        config = self.config.data_ingestion
+
+        create_directories([config.root_dir])
+
+        data_ingestion_config = DataIngestionConfig(
+            root_dir=config.root_dir,
+            source_dir=config.source_dir,
+            local_data_file=config.local_data_file,
+            unzip_dir=config.unzip_dir 
+        )
+
+        return data_ingestion_config
+        
+    
+    

src/imagecolorization/conponents/data_ingestion.py

@@ -0,0 +1,33 @@
+import os
+import zipfile
+from src.imagecolorization.logging import logger
+from tqdm.notebook import tqdm
+from dataclasses import replace
+from src.imagecolorization.entity.config_entity import DataIngestionConfig
+
+class DataIngestion:
+    def __init__(self, config: DataIngestionConfig):
+        self.config = config
+    
+    def load_file(self):
+        if os.path.exists(self.config.source_dir):
+            self.config = replace(self.config, local_data_file=self.config.source_dir)
+            logger.info(f'File Found at: {self.config.local_data_file}')
+        else:
+            logger.info(f"File not found at {self.config.source_dir}")
+            raise FileNotFoundError(f'No file found at {self.config.source_dir}')
+        
+        
+        
+    def extract_zip_file(self):
+        unzip_path = self.config.unzip_dir
+        os.makedirs(unzip_path, exist_ok=True)
+        
+        # open the zip file
+        with zipfile.ZipFile(self.config.local_data_file, 'r') as zip_ref:
+            total_files = len(zip_ref.infolist())
+            
+            for file in tqdm(iterable=zip_ref.infolist(), total=total_files, desc='Extracting files'):
+                zip_ref.extract(member=file, path=unzip_path)
+                
+            logger.info(f'Extacted {self.config.local_data_file} to {unzip_path}')

src/imagecolorization/constants/__init__.py

@@ -0,0 +1,4 @@
+from pathlib import  Path
+
+CONFIG_FILE_PATH = Path('config/config.yaml')
+PARAMS_FILE_PATH = Path('params.yaml')

src/imagecolorization/entity/config_entity.py

@@ -0,0 +1,9 @@
+from dataclasses import dataclass
+from pathlib import Path
+
+@dataclass(frozen=True)
+class DataIngestionConfig:
+    root_dir : Path
+    source_dir : Path
+    local_data_file: Path
+    unzip_dir : Path

src/imagecolorization/logging/__init__.py

@@ -0,0 +1,22 @@
+import os
+import sys
+import logging
+
+logging_str = "[%(asctime)s: %(levelname)s: %(module)s: %(message)s]"
+log_dir = "logs"
+log_filepath = os.path.join(log_dir,"running_logs.log")
+os.makedirs(log_dir, exist_ok=True)
+
+
+
+logging.basicConfig(
+    level= logging.INFO,
+    format= logging_str,
+
+    handlers=[
+        logging.FileHandler(log_filepath),
+        logging.StreamHandler(sys.stdout)
+    ]
+)
+
+logger = logging.getLogger("imagecolorizationLogger")

src/imagecolorization/pipeline/stage01_data_ingestion.py

@@ -0,0 +1,13 @@
+from src.imagecolorization.conponents.data_ingestion import DataIngestion
+from src.imagecolorization.config.configuration import ConfigurationManager
+
+class DataIngestionPipeline:
+    def __init__(self):
+        pass
+    
+    def main(self):
+        config = ConfigurationManager()
+        data_ingestion_config = config.get_data_ingestion_config()
+        data_ingesion = DataIngestion(config=data_ingestion_config)
+        data_ingesion.load_file()
+        data_ingesion.extract_zip_file()

src/imagecolorization/utils/common.py

@@ -0,0 +1,83 @@
+import os
+from box.exceptions import BoxValueError
+import yaml
+from src.imagecolorization.logging import logger
+from ensure import ensure_annotations
+from box import ConfigBox
+from pathlib import Path
+from typing import Any
+import json
+
+
+@ensure_annotations
+def read_yaml(path_to_yaml: Path) -> ConfigBox:
+    """reads yaml file and returns
+
+    Args:
+        path_to_yaml (str): path like input
+
+    Raises:
+        ValueError: if yaml file is empty
+        e: empty file
+
+    Returns:
+        ConfigBox: ConfigBox type
+    """
+    try:
+        with open(path_to_yaml) as yaml_file:
+            content = yaml.safe_load(yaml_file)
+            logger.info(f"yaml file: {path_to_yaml} loaded successfully")
+            return ConfigBox(content)
+    except BoxValueError:
+        raise ValueError("yaml file is empty")
+    except Exception as e:
+        raise e
+    
+    
+@ensure_annotations
+def create_directories(path_to_directories: list, verbose=True):
+    """create list of directories
+
+    Args:
+        path_to_directories (list): list of path of directories
+        ignore_log (bool, optional): ignore if multiple dirs is to be created. Defaults to False.
+    """
+    for path in path_to_directories:
+        os.makedirs(path, exist_ok=True)
+        if verbose:
+            logger.info(f"created directory at: {path}")
+    
+    
+@ensure_annotations
+def save_json(path: Path, data: dict):
+    """save json data
+
+    Args:
+        path (Path): path to json file
+        data (dict): data to be saved in json file
+    """
+    
+    
+    with open(path, 'w') as f:
+        json.dump(data, f, indent=4)
+        
+    logger.info(f'Json file saved at: {path}')
+    
+    
+@ensure_annotations
+def load_json(path: Path) -> ConfigBox:
+    """load json files data
+
+    Args:
+        path (Path): path to json file
+
+    Returns:
+        ConfigBox: data as class attributes instead of dict
+    """
+    
+    with open(path, 'r') as f:
+        content = json.load(f)
+
+    logger.info(f"Json file loaded successfully from: {path}")
+    return ConfigBox
+