-
Notifications
You must be signed in to change notification settings - Fork 1
/
tools.py
181 lines (140 loc) · 5.64 KB
/
tools.py
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
import re
from typing import List
import json
import datetime
import random
from openai import OpenAI
import streamlit as st
from collections import defaultdict
from langchain.agents import tool
from prompts import (
RECOMMEND_DISCIPLINE
)
CLIENT = OpenAI()
@tool
def determine_fitness_discipline(
user_workouts: str,
preferences: str
) -> str:
"""Determine the recommended fitness disciplines for the next workout.
Utilizes the user's recent workouts and preferences to determine the
fitness disciplines that should be the focus of the next workout. The
returned disciplines will be used to find candidate classes for the workout.
"""
prompt = RECOMMEND_DISCIPLINE.format(
USER_WORKOUTS=user_workouts,
PREFERENCES=preferences
)
chat_completion = CLIENT.chat.completions.create(
model="gpt-4o",
messages=[{"role": "user", "content": prompt}]
)
return chat_completion.choices[0].message.content
@tool
def get_peloton_classes(fitness_discipline: str) -> str:
"""Get recent Peloton classes available on the platform.
fitness_discipline representes the type of class. must be one of strength,
cycling, stretching, yoga, or cardio.
"""
response = st.session_state["pelo_interface"].get_recent_classes(fitness_discipline)
# response = json.load(open("peloton_classes.json", "r"))
# Get available classes from the past N days.
today = datetime.datetime.today().date()
recent_classes = []
# Create the instructors lookup.
instructors = {i['id']: i['name'] for i in response['instructors']}
for w in response['data']:
workout_date = datetime.datetime.fromtimestamp(w['original_air_time']).date()
if (today - workout_date).days > 14:
break
recent_classes.append(
{
'id': w["id"],
'description': w['description'],
'difficulty': w['difficulty_estimate'],
'duration': w['duration'],
'instructor': instructors[w['instructor_id']],
'title': w['title'],
'disciplie': w['fitness_discipline_display_name']
}
)
random.shuffle(recent_classes)
return json.dumps(recent_classes)
@tool
def get_recent_user_workouts() -> str:
"""Call the Peloton API to get the recent workouts.
Use this tool to get a summary of the user's recent Peloton history.
"""
# Check if the user has already loaded user workouts in this session.
# Since the workouts shouldn't change much load them from the sesion
# state if they are available.
if "user_workouts" in st.session_state:
return json.dumps(st.session_state["user_workouts"])
# Otherwise use the PelotonAPI to get the recent workouts for the
# user. Cache the results and return the list of workouts.
try:
response = st.session_state["pelo_interface"].get_user_workouts(
user_id=st.session_state["pelo_user_id"]
)
except Exception:
return "There was a problem getting workouts from Peloton. Check \
the logs for more information."
st.session_state["user_workouts"] = response
# Provide better formatting tags to the output.
user_classes = []
for k, classes in response.items():
workouts = []
for cl in classes:
class_str = f"<class>{cl}</class"
workouts.append(class_str)
workout_str = "\n".join(workouts)
day_classes = f"<day>Date: {k}\nClasses: {workout_str}</day>"
user_classes.append(day_classes)
user_class_str = "\n".join(user_classes)
return f"These are the recent user workouts for the past week: {user_class_str}"
# return f"These are the recent user workouts for the past week: {json.dumps(response)}"
@tool
def add_class_to_stack(recommended_classes: List[str]) -> str:
"""Allows a user to add selected workout to the Peloton stack if the user explicitly asks to.
recommended_classes: The list of recommended class IDs for the user. Could be one or more classes. ID should align with the classes recommended to the user.
"""
# Iterate through the classes and add each to my stack.
for class_id in recommended_classes:
join_token = st.session_state["pelo_interface"].convert_ride_to_class_id(class_id)
response = st.session_state["pelo_interface"].stack_class(join_token)
if response == False:
return "Sorry something went wrong."
return "Classes added to your stack."
@tool
def get_classes_in_stack() -> str:
"""Retrieves the classes in the user's Peloton stack."""
response = st.session_state["pelo_interface"].get_stack()
if not response:
return "No classes in your stack."
return response
@tool
def clear_classes_in_stack() -> str:
"""Clears the classes in the user's Peloton stack."""
response = st.session_state["pelo_interface"].clear_stack()
if not response:
return "No classes in your stack."
return "Stack cleared."
@tool
def get_user_workout_preferences() -> str:
"""Retrieves the user workout preferences from the file system.
Used to allow the user to review the preferences they have set.
"""
try:
with open('goals.txt', 'r') as f:
return f.read()
except FileNotFoundError:
return "No goal found."
@tool
def set_user_workout_preferences(preferences: str) -> str:
"""
Writes the workout preferences from the user to the file system.
preferences: The user's workout preferences.
"""
with open('goals.txt', 'w') as f:
f.write(preferences)
return "Preferences saved."