diff --git a/src/app/main.py b/src/app/main.py index 07ac575..9228e64 100644 --- a/src/app/main.py +++ b/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, - timedelta) -from data_access import LocalTestData +from data_access import LocalDataAccess st.set_page_config(page_title = 'Health Data Visualization', page_icon = ':bar_chart:', layout = 'wide') -local_data = LocalTestData('data') -data = local_data.get_data('huawei_health_data.json') - -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', '')) - ) - ) - } - ) +local_data = LocalDataAccess('data') +_ = local_data.get_data('huawei_health_data.json') +heart_rate = local_data.get_heart_rate() if st.sidebar.checkbox(f'All Data ({len(heart_rate)})', False): if st.sidebar.checkbox('Average of Days', False): - heart_rate = pd.DataFrame( - 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 + x, y = local_data.get_average_heart_rate_for_days_as_axis() labels = {'x': 'Date of The Day', 'y': 'Average Heart Rate'} else: - heart_rate = pd.DataFrame(heart_rate) - x = list(map( - lambda t: t.strftime("%d-%m-%Y %X"), - heart_rate['time'] - )) - y = heart_rate['rate'] + x, y = local_data.get_heart_rate_for_all_days_as_axis() 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( - 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'] + x, y = local_data.get_heart_rate_for_one_day(day) 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 = [] - 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 + x, y = local_data.get_averages_of_heart_rates(y, average_number) labels = {'x': 'Number of Heart Rate', 'y': 'Heart Rate'} chart_type = st.sidebar.selectbox( diff --git a/src/data_access/__init__.py b/src/data_access/__init__.py index 49c100a..4105972 100644 --- a/src/data_access/__init__.py +++ b/src/data_access/__init__.py @@ -1 +1 @@ -from .local_data_access import LocalTestData \ No newline at end of file +from .local_data_access import LocalDataAccess \ No newline at end of file diff --git a/src/data_access/data_opeartions.py b/src/data_access/data_opeartions.py new file mode 100644 index 0000000..e908508 --- /dev/null +++ b/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) \ No newline at end of file diff --git a/src/data_access/local_data_access.py b/src/data_access/local_data_access.py index 4a66066..0e99b9b 100644 --- a/src/data_access/local_data_access.py +++ b/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) - return data \ No newline at end of file + self.data: List = json.load(f) + return self.data \ No newline at end of file