w1 project

w1/data_processor.py

@@ -79,5 +79,15 @@ def aggregate(self, column_name: str) -> float:
         aggregate should be 105.58
         """
         ######################################## YOUR CODE HERE ##################################################
+         # get generator from data_reader
+        data_reader_gen = (row for row in self.data_reader)
+
+        # skip first row as it is the column name
+        _ = next(data_reader_gen)
+
+        agg = 0
 
+        for row in data_reader_gen:
+            agg = agg + self.to_float(row[column_name])
+        return agg
         ######################################## YOUR CODE HERE ##################################################

w1/main.py

@@ -8,6 +8,7 @@
 from global_utils import get_file_name, make_dir, plot_sales_data
 from datetime import datetime
 import json
+from collections import defaultdict
 
 
 CURRENT_FOLDER_NAME = os.path.dirname(os.path.abspath(__file__))
@@ -44,7 +45,18 @@ def revenue_per_region(dp: DataProcessor) -> Dict:
     }
     """
     ######################################## YOUR CODE HERE ##################################################
+    data_reader_gen = (row for row in dp.data_reader)
 
+    _ = next(data_reader_gen)
+
+    rev_region = defaultdict(lambda: 0.0)
+
+    for row in data_reader_gen:
+        country = row['Country']
+        val = dp.to_float(row['TotalPrice'])
+        rev_region[country] += val
+
+    return rev_region
     ######################################## YOUR CODE HERE ##################################################
 
 

w1/utils.py

@@ -2,6 +2,7 @@
 import numpy as np
 from typing import Generator, List
 import os
+import csv
 
 CURRENT_FOLDER = os.path.dirname(os.path.abspath(__file__))
 
@@ -112,7 +113,13 @@ def __iter__(self) -> Generator:
         }
         """
     ######################################## YOUR CODE HERE ##################################################
-
+        with open(self.get_file_path(), 'r') as f:
+            rows = csv.reader(f)
+            for row in rows:
+                single_row = {}
+                for idx, col in enumerate(self.get_column_names()):
+                    single_row[col] = row[idx]
+                yield single_row
     ######################################## YOUR CODE HERE ##################################################
 
     def get_file_path(self):

w2/index.html

@@ -41,7 +41,7 @@
             var url = new URL(urlString);
 
             // Make a websocket connection
-            var ws = new WebSocket(`ws://localhost:8000/ws`);
+            var ws = new WebSocket(`wss://8000-saustar-coursepython4pr-6zrtc9iyya8.ws-us98.gitpod.io/ws`);
 
             // On receiving a message, append it to the unordered list `messages`
             function processMessage(event) {

w2/server.py

@@ -42,7 +42,7 @@ async def get() -> Dict:
     """
 
     ######################################## YOUR CODE HERE ##################################################
-
+    return {"status": "ok"}
     ######################################## YOUR CODE HERE ##################################################
 
 
@@ -53,7 +53,9 @@ async def get() -> HTMLResponse:
     should render the HTML file - index.html when a user goes to http://127.0.0.1:8000/
     """
     ######################################## YOUR CODE HERE ##################################################
-
+    with open('index.html') as f:
+        contentis = f.read()
+        return HTMLResponse(content=contentis)
     ######################################## YOUR CODE HERE ##################################################
 
 
@@ -64,5 +66,7 @@ async def get() -> List[ProcessStatus]:
     Get all the records from the process table and return it using the pydantic model ProcessStatus
     """
     ######################################## YOUR CODE HERE ##################################################
-
+    db = DB()
+    processes = db.read_all()
+    return [ProcessStatus(**proc) for proc in processes]
     ######################################## YOUR CODE HERE ##################################################

w2/utils/database.py

@@ -45,6 +45,22 @@ def create_table(self) -> None:
         Read more about datatypes in Sqlite here -> https://www.sqlite.org/datatype3.html
         """
     ######################################## YOUR CODE HERE ##################################################
+        query = f''' 
+            CREATE TABLE IF NOT EXISTS {self._table_name}
+                 (
+                     process_id TEXT NOT NULL,
+                     file_name TEXT DEFAULT NULL,
+                     file_path TEXT NOT NULL,
+                     description TEXT DEFAULT NULL,
+                     start_time TEXT NOT NULL,
+                     end_time TEXT DEFAULT NULL,
+                     percentage REAL DEFAULT NULL
+                 )    
+        '''
+        self._connection.execute(query)
+        self._connection.commit()
+
+
 
     ######################################## YOUR CODE HERE ##################################################
 
@@ -63,7 +79,12 @@ def insert(self, process_id, start_time, file_name=None, file_path=None,
         :return: None
         """
     ######################################## YOUR CODE HERE ##################################################
-
+        query = f'''
+            INSERT INTO {self._table_name} (process_id, start_time, file_name, file_path, description, end_time, percentage)
+            VALUES ('{process_id}','{start_time}', '{file_name}', '{file_path}', '{description}', '{end_time}', '{percentage}')
+        ''' 
+        self._connection.execute(query)
+        self._connection.commit()
     ######################################## YOUR CODE HERE ##################################################
 
     def read_all(self) -> List[Dict]:
@@ -95,7 +116,10 @@ def update_percentage(self, process_id, percentage):
         :return: None
         """
     ######################################## YOUR CODE HERE ##################################################
+        self._connection.execute(f'''UPDATE {self._table_name} SET percentage='{percentage}'
+                                     WHERE process_id='{process_id}';''')
 
+        self._connection.commit()
     ######################################## YOUR CODE HERE ##################################################
 
 

w3/main.py

@@ -144,7 +144,7 @@ def main() -> List[Dict]:
     """
 
     st = time.time()
-    n_processes = 3 # you may modify this number - check out multiprocessing.cpu_count() as well
+    n_processes = 6 # you may modify this number - check out multiprocessing.cpu_count() as well
 
     parser = argparse.ArgumentParser(description="Choose from one of these : [tst|sml|bg]")
     parser.add_argument('--type',
@@ -164,14 +164,18 @@ def main() -> List[Dict]:
     batches = batch_files(file_paths=file_paths, n_processes=n_processes)
 
     ######################################## YOUR CODE HERE ##################################################
-
+    with multiprocessing.Pool(processes=n_processes) as pool:
+        revenue_data = pool.starmap(run, [(batch,n_process) for n_process,batch in enumerate(batches)])
+        revenue_data = flatten(lst=revenue_data)
+        pool.close()
+        pool.join()
     ######################################## YOUR CODE HERE ##################################################
 
     en = time.time()
     print("Overall time taken : {}".format(en-st))
 
     # should return revenue data
-    return [{}]
+    return revenue_data
 
 
 if __name__ == '__main__':