Coverage for mindsdb / integrations / handlers / google_fit_handler / google_fit_tables.py: 0%

1from mindsdb.integrations.libs.api_handler import APITable 

2from mindsdb.integrations.libs.response import HandlerResponse as Response 

3from mindsdb.integrations.utilities.sql_utils import extract_comparison_conditions 

4from mindsdb_sql_parser import ast 

5import datetime 

6import pytz 

7import time 

8from tzlocal import get_localzone 

9 

10 

11class GoogleFitTable(APITable): 

12 

13 def time_parser(self, args) -> int: 

14 """ 

15 Receive raw date string and return the calculated milliseconds based on the time string. 

16 Args: 

17 args: time string in the format of YYYY-MM-DD 

18 Returns: 

19 the input time string in the format of milliseconds 

20 """ 

21 ymd = args.split('-') 

22 epoch0 = datetime.datetime(1970, 1, 1, tzinfo=pytz.utc) 

23 time = pytz.timezone(str(get_localzone())).localize(datetime.datetime(int(ymd[0].rstrip()), int(ymd[1].rstrip()), int(ymd[2].rstrip()))) 

24 return int((time - epoch0).total_seconds() * 1000) 

25 

26 def select(self, query: ast.Select) -> Response: 

27 

28 conditions = extract_comparison_conditions(query.where) 

29 

30 params = {} 

31 # get the local time 

32 now = int(round(time.time() * 1000)) 

33 

34 # hard coded for now as user default query time period 

35 one_month = 2629746000 

36 for op, arg1, arg2 in conditions: 

37 if op == 'or': 

38 raise NotImplementedError('OR is not supported') 

39 if arg1 == 'date': 

40 date = self.time_parser(arg2) 

41 if op == '>': 

42 params['start_time'] = date 

43 params['end_time'] = now 

44 

45 # hard coded as a month 

46 elif op == '<': 

47 params['start_time'] = date - one_month 

48 params['end_time'] = date 

49 else: 

50 raise NotImplementedError 

51 else: 

52 raise NotImplementedError('This query is not supported') 

53 # if time is not provided in the query, the time range is one month ago to now 

54 if not params: 

55 params['start_time'] = now - one_month 

56 params['end_time'] = now 

57 result = self.handler.call_google_fit_api( 

58 method_name='get_steps', 

59 params=params 

60 ) 

61 return result 

62 

63 def get_columns(self): 

64 return [ 

65 'dates', 

66 'steps' 

67 ]