Data operations removed from main and added to data_access

src/app/main.py

@@ -3,59 +3,29 @@ from pathlib import Path
 main_path: Path = Path(__file__).resolve().parent.parent
 
 from sys import path
-path.append(main_path)
+if str(main_path) not in path:
+    path.append(str(main_path))
 
-import pandas as pd
 import plotly.express as px
 import streamlit as st
-from datetime import (datetime,
+from data_access import LocalDataAccess
-                    timedelta)
-from data_access import LocalTestData
 
 
 st.set_page_config(page_title = 'Health Data Visualization',
                     page_icon = ':bar_chart:',
                     layout = 'wide')
 
-local_data = LocalTestData('data')
+local_data = LocalDataAccess('data')
-data = local_data.get_data('huawei_health_data.json')
+_ = local_data.get_data('huawei_health_data.json')
-
+heart_rate = local_data.get_heart_rate()
-heart_rate = []
-for d in data:
-    if d['type'] == 7:
-        heart_rate.append(
-            {
-                'rate': float(d['samplePoints'][0]['value']),
-                'time': (datetime.fromtimestamp(
-                            int(str(d['samplePoints'][0]['endTime'])[:10])
-                    ) + timedelta(
-                            hours = int(data[0]['timeZone'][1:].replace('0', ''))
-                        )
-                )
-            }
-        )
 
 if st.sidebar.checkbox(f'All Data ({len(heart_rate)})', False):
     if st.sidebar.checkbox('Average of Days', False):
-        heart_rate = pd.DataFrame(
+        x, y = local_data.get_average_heart_rate_for_days_as_axis()
-            list(
-                map(
-                    lambda t: {'rate': t['rate'], 'date': t['time'].strftime("%d-%m-%Y")}, heart_rate
-                )
-            )
-        )
-        heart_rate_grouped = heart_rate.groupby('date', sort=False).mean()['rate']
-        x = heart_rate_grouped.keys()
-        y = heart_rate_grouped.values
         labels = {'x': 'Date of The Day', 'y': 'Average Heart Rate'}
 
     else:
-        heart_rate = pd.DataFrame(heart_rate)
+        x, y = local_data.get_heart_rate_for_all_days_as_axis()
-        x = list(map(
-            lambda t: t.strftime("%d-%m-%Y %X"),
-            heart_rate['time']
-        ))
-        y = heart_rate['rate']
         labels = {'x': 'Date and Time', 'y': 'Heart Rate'}
 
 else:
@@ -66,19 +36,7 @@ else:
         max_value = heart_rate[-1]['time']
     ).strftime("%d-%m-%Y")
 
-    heart_rate = pd.DataFrame(
+    x, y = local_data.get_heart_rate_for_one_day(day)
-        list(
-            filter(
-                lambda t: t['time'].strftime("%d-%m-%Y") == day, heart_rate
-            )
-        )
-    )
-
-    x = list(map(
-        lambda t: t.strftime("%d-%m-%Y %X"),
-        heart_rate['time']
-    ))
-    y = heart_rate['rate']
     labels = {'x': 'Date and Time', 'y': 'Heart Rate'}
 
 st.sidebar.header('Split Data:')
@@ -90,14 +48,7 @@ average_number = st.sidebar.number_input(
 )
 
 if average_number > 1:
-    heart_rate2 = []
+    x, y = local_data.get_averages_of_heart_rates(y, average_number)
-    rates = list(y)
-
-    for i in range(0, len(rates), average_number):
-        t = rates[i:i + average_number]
-        heart_rate2.append(sum(t) / len(t))
-    x = range(len(heart_rate2))
-    y = heart_rate2
     labels = {'x': 'Number of Heart Rate', 'y': 'Heart Rate'}
 
 chart_type = st.sidebar.selectbox(

src/data_access/__init__.py

@@ -1 +1 @@
-from .local_data_access import LocalTestData
+from .local_data_access import LocalDataAccess

src/data_access/data_opeartions.py

@@ -0,0 +1,97 @@
+import pandas as pd
+from datetime import (datetime,
+                    timedelta)
+from typing import (List,
+                    Tuple)
+
+
+class DataOperations:
+    heart_rate = None
+
+    def is_data_none_wrapper(func):
+        def __is_data_none(self, *args, **kwargs) -> None | NameError:
+            if self.data is None:
+                raise NameError(f'data not found. You must call the get_data function before calling the {func.__name__} function')
+            return func(self, *args, **kwargs)
+        return __is_data_none
+
+    def is_heart_rate_none_wrapper(func):
+        def __is_heart_rate_none(self, *args, **kwargs) -> None | NameError:
+            if self.heart_rate is None:
+                raise NameError(f'heart_rate data not found. You must call the get_heart_rate function before calling the {func.__name__} function')
+            return func(self, *args, **kwargs)
+        return __is_heart_rate_none
+
+    @is_data_none_wrapper
+    def get_heart_rate(self) -> List:
+        # sourcery skip: inline-immediately-returned-variable, list-comprehension
+        self.heart_rate = []
+        for d in self.data:
+            if d['type'] == 7:
+                self.heart_rate.append(
+                    {
+                        'rate': float(d['samplePoints'][0]['value']),
+                        'time': (datetime.fromtimestamp(
+                                    int(str(d['samplePoints'][0]['endTime'])[:10])
+                            ) + timedelta(
+                                    hours = int(self.data[0]['timeZone'][1:].replace('0', ''))
+                                )
+                        )
+                    }
+                )
+        return self.heart_rate
+
+    @is_heart_rate_none_wrapper
+    def get_heart_rate_for_all_days_as_axis(self) -> Tuple:
+        heart_rate = pd.DataFrame(self.heart_rate)
+        return (
+            list (
+                map(
+                    lambda t: t.strftime("%d-%m-%Y %X"),
+                    heart_rate['time']
+                )
+            ),
+            heart_rate['rate']
+        )
+
+    @is_heart_rate_none_wrapper
+    def get_average_heart_rate_for_days_as_axis(self) -> Tuple:
+        heart_rate_grouped = pd.DataFrame(
+            list(
+                map(
+                    lambda t: {'rate': t['rate'], 'date': t['time'].strftime("%d-%m-%Y")}, self.heart_rate
+                )
+            )
+        ).groupby('date', sort=False).mean()['rate']
+
+        return (heart_rate_grouped.keys(), heart_rate_grouped.values)
+
+    @is_heart_rate_none_wrapper
+    def get_heart_rate_for_one_day(self, day) -> Tuple:
+        heart_rate = pd.DataFrame(
+            list(
+                filter(
+                    lambda t: t['time'].strftime("%d-%m-%Y") == day, self.heart_rate
+                )
+            )
+        )
+
+        return (
+            list (
+                map(
+                    lambda t: t.strftime("%d-%m-%Y %X"),
+                    heart_rate['time']
+                )
+            ),
+            heart_rate['rate']
+        )
+
+    def get_averages_of_heart_rates(self, y: pd.Series, average_number: int) -> Tuple:
+        heart_rate2 = []
+        rates = list(y)
+
+        for i in range(0, len(rates), average_number):
+            t = rates[i:i + average_number]
+            heart_rate2.append(sum(t) / len(t))
+
+        return (range(len(heart_rate2)), heart_rate2)

src/data_access/local_data_access.py

@@ -1,13 +1,16 @@
 from pathlib import Path
 import json
 from typing import List
+from .data_opeartions import DataOperations
 
 
-class LocalTestData:
+class LocalDataAccess(DataOperations):
     def __init__(self, data_folder_name: str) -> None:
-        self.data_dir: Path = Path(__file__).resolve().parent.parent.parent / data_folder_name
+        self.data = None
+        self.data_dir: Path =\
+            Path(__file__).resolve().parent.parent.parent / data_folder_name
 
     def get_data(self, file_name: str) -> List:
         with open(self.data_dir / file_name) as f:
-            data: List = json.load(f)
+            self.data: List = json.load(f)
-        return data
+        return self.data