cli to view workouts
This commit is contained in:
@@ -2,3 +2,7 @@
|
||||
backend_base_url = "http://localhost:8080"
|
||||
backend_user = "test@example.com"
|
||||
backend_pw = "test"
|
||||
|
||||
log_file = "local/logs/taiga_cli.log"
|
||||
log_stream = false
|
||||
|
||||
|
||||
@@ -21,8 +21,8 @@ else
|
||||
fi
|
||||
|
||||
banner "Uploading test files"
|
||||
uv run taiga activities upload local/test-data/test.fit
|
||||
uv run taiga activities upload local/test-data/test2.fit
|
||||
uv run taiga activities upload local/test-data/test3.fit
|
||||
uv run taiga workouts upload local/test-data/test.fit
|
||||
uv run taiga workouts upload local/test-data/test2.fit
|
||||
uv run taiga workouts upload local/test-data/test3.fit
|
||||
|
||||
banner "Setup complete!"
|
||||
|
||||
@@ -7,6 +7,7 @@ import taiga_pycli.cli.register
|
||||
import taiga_pycli.cli.login
|
||||
import taiga_pycli.cli.hats
|
||||
import taiga_pycli.cli.activities
|
||||
import taiga_pycli.cli.workouts
|
||||
|
||||
_logger = logging.getLogger(__name__)
|
||||
|
||||
@@ -25,6 +26,7 @@ def parse_args():
|
||||
taiga_pycli.cli.login.setup_parser(subparsers)
|
||||
taiga_pycli.cli.hats.setup_parser(subparsers)
|
||||
taiga_pycli.cli.activities.setup_parser(subparsers)
|
||||
taiga_pycli.cli.workouts.setup_parser(subparsers)
|
||||
|
||||
args = parser.parse_args()
|
||||
return args
|
||||
|
||||
@@ -5,6 +5,7 @@ import pathlib
|
||||
|
||||
import taiga_pycli.config
|
||||
import taiga_pycli.service
|
||||
import taiga_pycli.workout_utils
|
||||
from taiga_pycli.exceptions import (
|
||||
AuthenticationError,
|
||||
ValidationError,
|
||||
@@ -105,13 +106,15 @@ def run_list(config: taiga_pycli.config.Config, args):
|
||||
return 0
|
||||
|
||||
# Print header
|
||||
print(f"{'ID':<10} {'Timestamp':<20} {'Created':<20}")
|
||||
print("-" * 50)
|
||||
print(f"{'ID':<5} {'Timestamp':<20} {'Created':<20}")
|
||||
print("-" * 45)
|
||||
|
||||
# Print each activity file
|
||||
for activity_file in activity_files:
|
||||
timestamp_formatted = taiga_pycli.workout_utils.format_activity_timestamp(activity_file.timestamp)
|
||||
created_formatted = taiga_pycli.workout_utils.format_activity_timestamp(activity_file.created_at)
|
||||
print(
|
||||
f"{activity_file.id:<10} {activity_file.timestamp:<20} {activity_file.created_at:<20}"
|
||||
f"{activity_file.id:<5} {timestamp_formatted:<20} {created_formatted:<20}"
|
||||
)
|
||||
|
||||
except AuthenticationError as e:
|
||||
|
||||
299
src/taiga_pycli/cli/workouts.py
Normal file
299
src/taiga_pycli/cli/workouts.py
Normal file
@@ -0,0 +1,299 @@
|
||||
import argparse
|
||||
import logging
|
||||
import typing
|
||||
import pathlib
|
||||
|
||||
import taiga_pycli.config
|
||||
import taiga_pycli.service
|
||||
import taiga_pycli.workout_utils
|
||||
from taiga_pycli.models import CreateWorkoutRequest
|
||||
from taiga_pycli.exceptions import (
|
||||
AuthenticationError,
|
||||
ValidationError,
|
||||
NetworkError,
|
||||
TryGoAPIError,
|
||||
)
|
||||
|
||||
_logger = logging.getLogger(__name__)
|
||||
|
||||
|
||||
if typing.TYPE_CHECKING:
|
||||
_SubparserType = argparse._SubParsersAction[argparse.ArgumentParser]
|
||||
else:
|
||||
_SubparserType = typing.Any
|
||||
|
||||
|
||||
def setup_parser(subparsers: _SubparserType) -> None:
|
||||
"""Setup the workouts command group with its subcommands"""
|
||||
parser = subparsers.add_parser("workouts", help="Manage workouts")
|
||||
|
||||
workouts_subparsers = parser.add_subparsers(dest="workouts_cmd", required=True)
|
||||
|
||||
# Upload subcommand - primary workflow
|
||||
upload_parser = workouts_subparsers.add_parser(
|
||||
"upload", help="Upload a FIT file and create workout automatically"
|
||||
)
|
||||
upload_parser.add_argument(
|
||||
"file_path", type=pathlib.Path, help="Path to the FIT file to upload"
|
||||
)
|
||||
upload_parser.set_defaults(func=run_upload)
|
||||
|
||||
# List subcommand
|
||||
list_parser = workouts_subparsers.add_parser(
|
||||
"ls", help="List your workouts"
|
||||
)
|
||||
list_parser.set_defaults(func=run_list)
|
||||
|
||||
# Show subcommand
|
||||
show_parser = workouts_subparsers.add_parser(
|
||||
"show", help="Show detailed workout information"
|
||||
)
|
||||
show_parser.add_argument("id", type=int, help="Workout ID to show")
|
||||
show_parser.set_defaults(func=run_show)
|
||||
|
||||
# Create subcommand for manual workout entry
|
||||
create_parser = workouts_subparsers.add_parser(
|
||||
"create", help="Create a workout manually"
|
||||
)
|
||||
create_parser.add_argument("--distance", type=float, help="Distance in miles")
|
||||
create_parser.add_argument("--time", type=int, help="Duration in seconds")
|
||||
create_parser.add_argument("--pace", type=float, help="Pace in minutes per mile")
|
||||
create_parser.add_argument("--speed", type=float, help="Speed in mph")
|
||||
create_parser.add_argument("--start-time", type=str, help="Start time (ISO format)")
|
||||
create_parser.add_argument("--end-time", type=str, help="End time (ISO format)")
|
||||
create_parser.set_defaults(func=run_create)
|
||||
|
||||
# Delete subcommand
|
||||
delete_parser = workouts_subparsers.add_parser(
|
||||
"delete", help="Delete a workout"
|
||||
)
|
||||
delete_parser.add_argument("id", type=int, help="Workout ID to delete")
|
||||
delete_parser.add_argument("--yes", "-y", action="store_true", help="Skip confirmation")
|
||||
delete_parser.set_defaults(func=run_delete)
|
||||
|
||||
|
||||
def run_upload(config: taiga_pycli.config.Config, args):
|
||||
"""Run the upload command - primary workflow"""
|
||||
try:
|
||||
clnt = taiga_pycli.service.BackendService(config)
|
||||
|
||||
_logger.info(f"Uploading FIT file and creating workout: {args.file_path}")
|
||||
|
||||
# Check file extension
|
||||
if not args.file_path.name.lower().endswith('.fit'):
|
||||
print(f"Warning: File does not have .fit extension: {args.file_path}")
|
||||
|
||||
# Upload and create workout in one step
|
||||
workout = clnt.upload_and_create_workout(args.file_path)
|
||||
|
||||
print("Workout created successfully!")
|
||||
print(f"Workout ID: {workout.id}")
|
||||
|
||||
# Print formatted workout summary
|
||||
summary = taiga_pycli.workout_utils.format_workout_summary(workout)
|
||||
if summary != "Workout data":
|
||||
print(f"Summary: {summary}")
|
||||
|
||||
# Show additional details if available
|
||||
if workout.distance_miles or workout.time_seconds:
|
||||
print("\nWorkout Details:")
|
||||
taiga_pycli.workout_utils.print_workout_details(workout)
|
||||
|
||||
except AuthenticationError as e:
|
||||
print(f"Authentication error: {e}")
|
||||
print("Please run 'taiga login' first.")
|
||||
return 1
|
||||
except ValidationError as e:
|
||||
print(f"Validation error: {e}")
|
||||
return 1
|
||||
except NetworkError as e:
|
||||
print(f"Network error: {e}")
|
||||
return 1
|
||||
except TryGoAPIError as e:
|
||||
print(f"API error: {e}")
|
||||
return 1
|
||||
except Exception as e:
|
||||
_logger.error(f"Unexpected error during workout upload: {e}")
|
||||
print(f"Unexpected error: {e}")
|
||||
return 1
|
||||
|
||||
return 0
|
||||
|
||||
|
||||
def run_list(config: taiga_pycli.config.Config, args):
|
||||
"""Run the list command"""
|
||||
try:
|
||||
clnt = taiga_pycli.service.BackendService(config)
|
||||
|
||||
_logger.info("Listing workouts")
|
||||
|
||||
workouts = clnt.get_workouts()
|
||||
|
||||
if not workouts:
|
||||
print("No workouts found.")
|
||||
print("Upload a FIT file with 'taiga workouts upload <file.fit>' to get started!")
|
||||
return 0
|
||||
|
||||
# Print table header
|
||||
taiga_pycli.workout_utils.print_workout_table_header()
|
||||
|
||||
# Print each workout
|
||||
for workout in workouts:
|
||||
taiga_pycli.workout_utils.print_workout_row(workout)
|
||||
|
||||
print(f"\nTotal: {len(workouts)} workouts")
|
||||
|
||||
except AuthenticationError as e:
|
||||
print(f"Authentication error: {e}")
|
||||
print("Please run 'taiga login' first.")
|
||||
return 1
|
||||
except NetworkError as e:
|
||||
print(f"Network error: {e}")
|
||||
return 1
|
||||
except TryGoAPIError as e:
|
||||
print(f"API error: {e}")
|
||||
return 1
|
||||
except Exception as e:
|
||||
_logger.error(f"Unexpected error during list: {e}")
|
||||
print(f"Unexpected error: {e}")
|
||||
return 1
|
||||
|
||||
return 0
|
||||
|
||||
|
||||
def run_show(config: taiga_pycli.config.Config, args):
|
||||
"""Run the show command"""
|
||||
try:
|
||||
clnt = taiga_pycli.service.BackendService(config)
|
||||
|
||||
_logger.info(f"Showing workout {args.id}")
|
||||
|
||||
workout = clnt.get_workout(args.id)
|
||||
|
||||
# Print detailed workout information
|
||||
taiga_pycli.workout_utils.print_workout_details(workout)
|
||||
|
||||
except AuthenticationError as e:
|
||||
print(f"Authentication error: {e}")
|
||||
print("Please run 'taiga login' first.")
|
||||
return 1
|
||||
except ValidationError as e:
|
||||
print(f"Workout not found: {e}")
|
||||
return 1
|
||||
except NetworkError as e:
|
||||
print(f"Network error: {e}")
|
||||
return 1
|
||||
except TryGoAPIError as e:
|
||||
print(f"API error: {e}")
|
||||
return 1
|
||||
except Exception as e:
|
||||
_logger.error(f"Unexpected error during show: {e}")
|
||||
print(f"Unexpected error: {e}")
|
||||
return 1
|
||||
|
||||
return 0
|
||||
|
||||
|
||||
def run_create(config: taiga_pycli.config.Config, args):
|
||||
"""Run the create command for manual workout entry"""
|
||||
try:
|
||||
clnt = taiga_pycli.service.BackendService(config)
|
||||
|
||||
_logger.info("Creating manual workout")
|
||||
|
||||
# Build request from args
|
||||
request = CreateWorkoutRequest(
|
||||
distance_miles=args.distance,
|
||||
time_seconds=args.time,
|
||||
pace_min_per_mile=args.pace,
|
||||
speed_mph=args.speed,
|
||||
start_time=args.start_time,
|
||||
end_time=args.end_time,
|
||||
)
|
||||
|
||||
# Validate that at least some data is provided
|
||||
if not any([args.distance, args.time, args.pace, args.speed, args.start_time]):
|
||||
print("Error: At least one workout parameter must be provided")
|
||||
print("Use --distance, --time, --pace, --speed, or --start-time")
|
||||
return 1
|
||||
|
||||
workout = clnt.create_workout(request)
|
||||
|
||||
print("Workout created successfully!")
|
||||
print(f"Workout ID: {workout.id}")
|
||||
|
||||
# Show workout details
|
||||
print("\nWorkout Details:")
|
||||
taiga_pycli.workout_utils.print_workout_details(workout)
|
||||
|
||||
except AuthenticationError as e:
|
||||
print(f"Authentication error: {e}")
|
||||
print("Please run 'taiga login' first.")
|
||||
return 1
|
||||
except ValidationError as e:
|
||||
print(f"Validation error: {e}")
|
||||
return 1
|
||||
except NetworkError as e:
|
||||
print(f"Network error: {e}")
|
||||
return 1
|
||||
except TryGoAPIError as e:
|
||||
print(f"API error: {e}")
|
||||
return 1
|
||||
except Exception as e:
|
||||
_logger.error(f"Unexpected error during create: {e}")
|
||||
print(f"Unexpected error: {e}")
|
||||
return 1
|
||||
|
||||
return 0
|
||||
|
||||
|
||||
def run_delete(config: taiga_pycli.config.Config, args):
|
||||
"""Run the delete command"""
|
||||
try:
|
||||
clnt = taiga_pycli.service.BackendService(config)
|
||||
|
||||
# Get workout details first to show what we're deleting
|
||||
try:
|
||||
workout = clnt.get_workout(args.id)
|
||||
except (ValidationError, TryGoAPIError):
|
||||
print(f"Workout {args.id} not found.")
|
||||
return 1
|
||||
|
||||
# Show what we're about to delete
|
||||
print(f"About to delete workout #{workout.id}:")
|
||||
summary = taiga_pycli.workout_utils.format_workout_summary(workout)
|
||||
date = taiga_pycli.workout_utils.format_workout_date(workout.start_time or workout.created_at)
|
||||
print(f" {date}: {summary}")
|
||||
|
||||
# Confirm deletion unless --yes flag is used
|
||||
if not args.yes:
|
||||
response = input("\nAre you sure you want to delete this workout? (y/N): ")
|
||||
if response.lower() not in ['y', 'yes']:
|
||||
print("Deletion cancelled.")
|
||||
return 0
|
||||
|
||||
_logger.info(f"Deleting workout {args.id}")
|
||||
|
||||
clnt.delete_workout(args.id)
|
||||
|
||||
print(f"Workout {args.id} deleted successfully.")
|
||||
|
||||
except AuthenticationError as e:
|
||||
print(f"Authentication error: {e}")
|
||||
print("Please run 'taiga login' first.")
|
||||
return 1
|
||||
except ValidationError as e:
|
||||
print(f"Workout not found: {e}")
|
||||
return 1
|
||||
except NetworkError as e:
|
||||
print(f"Network error: {e}")
|
||||
return 1
|
||||
except TryGoAPIError as e:
|
||||
print(f"API error: {e}")
|
||||
return 1
|
||||
except Exception as e:
|
||||
_logger.error(f"Unexpected error during delete: {e}")
|
||||
print(f"Unexpected error: {e}")
|
||||
return 1
|
||||
|
||||
return 0
|
||||
@@ -57,3 +57,46 @@ class ActivityFile:
|
||||
created_at: str
|
||||
updated_at: str
|
||||
user_id: int
|
||||
|
||||
|
||||
@dataclass
|
||||
class Workout:
|
||||
"""Workout model for API responses"""
|
||||
|
||||
id: int
|
||||
distance_miles: Optional[float]
|
||||
time_seconds: Optional[float]
|
||||
speed_mph: Optional[float]
|
||||
pace_min_per_mile: Optional[float]
|
||||
start_time: Optional[str]
|
||||
end_time: Optional[str]
|
||||
activity_file_id: Optional[int]
|
||||
created_at: str
|
||||
updated_at: str
|
||||
user_id: int
|
||||
|
||||
|
||||
@dataclass
|
||||
class CreateWorkoutRequest:
|
||||
"""Request payload for creating a new workout"""
|
||||
|
||||
distance_miles: Optional[float] = None
|
||||
time_seconds: Optional[float] = None
|
||||
speed_mph: Optional[float] = None
|
||||
pace_min_per_mile: Optional[float] = None
|
||||
start_time: Optional[str] = None
|
||||
end_time: Optional[str] = None
|
||||
activity_file_id: Optional[int] = None
|
||||
|
||||
|
||||
@dataclass
|
||||
class UpdateWorkoutRequest:
|
||||
"""Request payload for updating an existing workout"""
|
||||
|
||||
distance_miles: Optional[float] = None
|
||||
time_seconds: Optional[float] = None
|
||||
speed_mph: Optional[float] = None
|
||||
pace_min_per_mile: Optional[float] = None
|
||||
start_time: Optional[str] = None
|
||||
end_time: Optional[str] = None
|
||||
activity_file_id: Optional[int] = None
|
||||
|
||||
@@ -6,7 +6,15 @@ from pathlib import Path
|
||||
import requests
|
||||
|
||||
|
||||
from taiga_pycli.models import RegisterRequest, LoginRequest, AuthResponse, ActivityFile
|
||||
from taiga_pycli.models import (
|
||||
RegisterRequest,
|
||||
LoginRequest,
|
||||
AuthResponse,
|
||||
ActivityFile,
|
||||
Workout,
|
||||
CreateWorkoutRequest,
|
||||
UpdateWorkoutRequest,
|
||||
)
|
||||
from taiga_pycli.config import Config
|
||||
from taiga_pycli.exceptions import (
|
||||
TryGoAPIError,
|
||||
@@ -80,7 +88,9 @@ class BackendService:
|
||||
elif not response.ok:
|
||||
raise TryGoAPIError(f"API error: {response.status_code}")
|
||||
|
||||
# Parse JSON response
|
||||
# Parse JSON response (skip for 204 No Content)
|
||||
if response.status_code == 204:
|
||||
return response
|
||||
try:
|
||||
_logger.debug(response.json())
|
||||
return response
|
||||
@@ -393,3 +403,316 @@ class BackendService:
|
||||
raise NetworkError(f"Network error during download: {e}")
|
||||
except OSError as e:
|
||||
raise ValidationError(f"Error writing file {output_path}: {e}")
|
||||
|
||||
def upload_and_create_workout(self, file_path: Path) -> Workout:
|
||||
"""
|
||||
Upload FIT file and automatically create workout in one operation
|
||||
|
||||
Args:
|
||||
file_path: Path to the FIT file to upload
|
||||
|
||||
Returns:
|
||||
Workout object created from the FIT file
|
||||
|
||||
Raises:
|
||||
AuthenticationError: If not authenticated
|
||||
ValidationError: If file is invalid or FIT processing fails
|
||||
NetworkError: For connection issues
|
||||
TryGoAPIError: For API errors
|
||||
"""
|
||||
_logger.info(f"Uploading FIT file and creating workout: {file_path}")
|
||||
|
||||
activity_file = self.upload_activity_file(file_path)
|
||||
_logger.info(f"Activity file uploaded with ID: {activity_file.id}")
|
||||
|
||||
try:
|
||||
workout = self.create_workout_from_activity_file(activity_file.id)
|
||||
_logger.info(f"Workout created with ID: {workout.id}")
|
||||
return workout
|
||||
except Exception as e:
|
||||
_logger.error(f"Failed to create workout from activity file {activity_file.id}: {e}")
|
||||
raise
|
||||
|
||||
def get_workouts(self) -> List[Workout]:
|
||||
"""
|
||||
Get all workouts for the authenticated user
|
||||
|
||||
Returns:
|
||||
List of Workout objects
|
||||
|
||||
Raises:
|
||||
AuthenticationError: If not authenticated
|
||||
NetworkError: For connection issues
|
||||
TryGoAPIError: For API errors
|
||||
"""
|
||||
cred = self._read_credential()
|
||||
if cred is None:
|
||||
raise AuthenticationError("Not authenticated. Please login first.")
|
||||
|
||||
_logger.info("Fetching workouts list")
|
||||
|
||||
try:
|
||||
response = self._make_request("GET", "/workouts")
|
||||
response_data = response.json()
|
||||
|
||||
workouts: List[Workout] = []
|
||||
if isinstance(response_data, list):
|
||||
for item in response_data:
|
||||
if isinstance(item, dict):
|
||||
workout = Workout(
|
||||
id=item["id"],
|
||||
distance_miles=item.get("distance_miles"),
|
||||
time_seconds=item.get("time_seconds"),
|
||||
speed_mph=item.get("speed_mph"),
|
||||
pace_min_per_mile=item.get("pace_min_per_mile"),
|
||||
start_time=item.get("start_time"),
|
||||
end_time=item.get("end_time"),
|
||||
activity_file_id=item.get("activity_file_id"),
|
||||
created_at=item["created_at"],
|
||||
updated_at=item["updated_at"],
|
||||
user_id=item["user_id"],
|
||||
)
|
||||
workouts.append(workout)
|
||||
|
||||
_logger.info(f"Retrieved {len(workouts)} workouts")
|
||||
return workouts
|
||||
|
||||
except requests.exceptions.Timeout:
|
||||
raise NetworkError("Request timeout")
|
||||
except requests.exceptions.ConnectionError:
|
||||
raise NetworkError("Connection error")
|
||||
except requests.exceptions.RequestException as e:
|
||||
raise NetworkError(f"Network error: {e}")
|
||||
|
||||
def get_workout(self, workout_id: int) -> Workout:
|
||||
"""
|
||||
Get a specific workout by ID
|
||||
|
||||
Args:
|
||||
workout_id: The ID of the workout to retrieve
|
||||
|
||||
Returns:
|
||||
Workout object
|
||||
|
||||
Raises:
|
||||
AuthenticationError: If not authenticated
|
||||
ValidationError: If workout not found
|
||||
NetworkError: For connection issues
|
||||
TryGoAPIError: For API errors
|
||||
"""
|
||||
cred = self._read_credential()
|
||||
if cred is None:
|
||||
raise AuthenticationError("Not authenticated. Please login first.")
|
||||
|
||||
_logger.info(f"Fetching workout {workout_id}")
|
||||
|
||||
try:
|
||||
response = self._make_request("GET", f"/workouts/{workout_id}")
|
||||
item = response.json()
|
||||
|
||||
workout = Workout(
|
||||
id=item["id"],
|
||||
distance_miles=item.get("distance_miles"),
|
||||
time_seconds=item.get("time_seconds"),
|
||||
speed_mph=item.get("speed_mph"),
|
||||
pace_min_per_mile=item.get("pace_min_per_mile"),
|
||||
start_time=item.get("start_time"),
|
||||
end_time=item.get("end_time"),
|
||||
activity_file_id=item.get("activity_file_id"),
|
||||
created_at=item["created_at"],
|
||||
updated_at=item["updated_at"],
|
||||
user_id=item["user_id"],
|
||||
)
|
||||
|
||||
return workout
|
||||
|
||||
except requests.exceptions.Timeout:
|
||||
raise NetworkError("Request timeout")
|
||||
except requests.exceptions.ConnectionError:
|
||||
raise NetworkError("Connection error")
|
||||
except requests.exceptions.RequestException as e:
|
||||
raise NetworkError(f"Network error: {e}")
|
||||
|
||||
def create_workout(self, request: CreateWorkoutRequest) -> Workout:
|
||||
"""
|
||||
Create a new workout manually
|
||||
|
||||
Args:
|
||||
request: CreateWorkoutRequest with workout data
|
||||
|
||||
Returns:
|
||||
Created Workout object
|
||||
|
||||
Raises:
|
||||
AuthenticationError: If not authenticated
|
||||
ValidationError: If request data is invalid
|
||||
NetworkError: For connection issues
|
||||
TryGoAPIError: For API errors
|
||||
"""
|
||||
cred = self._read_credential()
|
||||
if cred is None:
|
||||
raise AuthenticationError("Not authenticated. Please login first.")
|
||||
|
||||
_logger.info("Creating new workout")
|
||||
|
||||
try:
|
||||
response = self._make_request("POST", "/workouts", asdict(request))
|
||||
item = response.json()
|
||||
|
||||
workout = Workout(
|
||||
id=item["id"],
|
||||
distance_miles=item.get("distance_miles"),
|
||||
time_seconds=item.get("time_seconds"),
|
||||
speed_mph=item.get("speed_mph"),
|
||||
pace_min_per_mile=item.get("pace_min_per_mile"),
|
||||
start_time=item.get("start_time"),
|
||||
end_time=item.get("end_time"),
|
||||
activity_file_id=item.get("activity_file_id"),
|
||||
created_at=item["created_at"],
|
||||
updated_at=item["updated_at"],
|
||||
user_id=item["user_id"],
|
||||
)
|
||||
|
||||
_logger.info(f"Created workout with ID: {workout.id}")
|
||||
return workout
|
||||
|
||||
except requests.exceptions.Timeout:
|
||||
raise NetworkError("Request timeout")
|
||||
except requests.exceptions.ConnectionError:
|
||||
raise NetworkError("Connection error")
|
||||
except requests.exceptions.RequestException as e:
|
||||
raise NetworkError(f"Network error: {e}")
|
||||
|
||||
def update_workout(self, workout_id: int, request: UpdateWorkoutRequest) -> Workout:
|
||||
"""
|
||||
Update an existing workout
|
||||
|
||||
Args:
|
||||
workout_id: The ID of the workout to update
|
||||
request: UpdateWorkoutRequest with fields to update
|
||||
|
||||
Returns:
|
||||
Updated Workout object
|
||||
|
||||
Raises:
|
||||
AuthenticationError: If not authenticated
|
||||
ValidationError: If workout not found or request data is invalid
|
||||
NetworkError: For connection issues
|
||||
TryGoAPIError: For API errors
|
||||
"""
|
||||
cred = self._read_credential()
|
||||
if cred is None:
|
||||
raise AuthenticationError("Not authenticated. Please login first.")
|
||||
|
||||
_logger.info(f"Updating workout {workout_id}")
|
||||
|
||||
try:
|
||||
response = self._make_request("PUT", f"/workouts/{workout_id}", asdict(request))
|
||||
item = response.json()
|
||||
|
||||
workout = Workout(
|
||||
id=item["id"],
|
||||
distance_miles=item.get("distance_miles"),
|
||||
time_seconds=item.get("time_seconds"),
|
||||
speed_mph=item.get("speed_mph"),
|
||||
pace_min_per_mile=item.get("pace_min_per_mile"),
|
||||
start_time=item.get("start_time"),
|
||||
end_time=item.get("end_time"),
|
||||
activity_file_id=item.get("activity_file_id"),
|
||||
created_at=item["created_at"],
|
||||
updated_at=item["updated_at"],
|
||||
user_id=item["user_id"],
|
||||
)
|
||||
|
||||
_logger.info(f"Updated workout {workout_id}")
|
||||
return workout
|
||||
|
||||
except requests.exceptions.Timeout:
|
||||
raise NetworkError("Request timeout")
|
||||
except requests.exceptions.ConnectionError:
|
||||
raise NetworkError("Connection error")
|
||||
except requests.exceptions.RequestException as e:
|
||||
raise NetworkError(f"Network error: {e}")
|
||||
|
||||
def delete_workout(self, workout_id: int) -> None:
|
||||
"""
|
||||
Delete a workout
|
||||
|
||||
Args:
|
||||
workout_id: The ID of the workout to delete
|
||||
|
||||
Raises:
|
||||
AuthenticationError: If not authenticated
|
||||
ValidationError: If workout not found
|
||||
NetworkError: For connection issues
|
||||
TryGoAPIError: For API errors
|
||||
"""
|
||||
cred = self._read_credential()
|
||||
if cred is None:
|
||||
raise AuthenticationError("Not authenticated. Please login first.")
|
||||
|
||||
_logger.info(f"Deleting workout {workout_id}")
|
||||
|
||||
try:
|
||||
self._make_request("DELETE", f"/workouts/{workout_id}")
|
||||
# DELETE returns 204 No Content with empty body, so don't try to parse JSON
|
||||
_logger.info(f"Deleted workout {workout_id}")
|
||||
|
||||
except requests.exceptions.Timeout:
|
||||
raise NetworkError("Request timeout")
|
||||
except requests.exceptions.ConnectionError:
|
||||
raise NetworkError("Connection error")
|
||||
except requests.exceptions.RequestException as e:
|
||||
raise NetworkError(f"Network error: {e}")
|
||||
|
||||
def create_workout_from_activity_file(self, activity_file_id: int) -> Workout:
|
||||
"""
|
||||
Create workout from an existing activity file
|
||||
|
||||
Args:
|
||||
activity_file_id: The ID of the activity file to process
|
||||
|
||||
Returns:
|
||||
Created Workout object
|
||||
|
||||
Raises:
|
||||
AuthenticationError: If not authenticated
|
||||
ValidationError: If activity file not found or FIT processing fails
|
||||
NetworkError: For connection issues
|
||||
TryGoAPIError: For API errors
|
||||
"""
|
||||
cred = self._read_credential()
|
||||
if cred is None:
|
||||
raise AuthenticationError("Not authenticated. Please login first.")
|
||||
|
||||
_logger.info(f"Creating workout from activity file {activity_file_id}")
|
||||
|
||||
try:
|
||||
response = self._make_request(
|
||||
"POST", f"/workouts/from-activity-file/{activity_file_id}"
|
||||
)
|
||||
item = response.json()
|
||||
|
||||
workout = Workout(
|
||||
id=item["id"],
|
||||
distance_miles=item.get("distance_miles"),
|
||||
time_seconds=item.get("time_seconds"),
|
||||
speed_mph=item.get("speed_mph"),
|
||||
pace_min_per_mile=item.get("pace_min_per_mile"),
|
||||
start_time=item.get("start_time"),
|
||||
end_time=item.get("end_time"),
|
||||
activity_file_id=item.get("activity_file_id"),
|
||||
created_at=item["created_at"],
|
||||
updated_at=item["updated_at"],
|
||||
user_id=item["user_id"],
|
||||
)
|
||||
|
||||
_logger.info(f"Created workout {workout.id} from activity file {activity_file_id}")
|
||||
return workout
|
||||
|
||||
except requests.exceptions.Timeout:
|
||||
raise NetworkError("Request timeout")
|
||||
except requests.exceptions.ConnectionError:
|
||||
raise NetworkError("Connection error")
|
||||
except requests.exceptions.RequestException as e:
|
||||
raise NetworkError(f"Network error: {e}")
|
||||
|
||||
199
src/taiga_pycli/workout_utils.py
Normal file
199
src/taiga_pycli/workout_utils.py
Normal file
@@ -0,0 +1,199 @@
|
||||
"""Utility functions for workout data formatting and display"""
|
||||
|
||||
from typing import Optional, Union
|
||||
from datetime import datetime
|
||||
from taiga_pycli.models import Workout
|
||||
|
||||
|
||||
def format_time_duration(seconds: Optional[Union[int, float]]) -> str:
|
||||
"""Format time duration in seconds to HH:MM:SS or MM:SS"""
|
||||
if seconds is None:
|
||||
return "--:--"
|
||||
|
||||
# Convert to int to handle float values from API
|
||||
total_seconds = int(seconds)
|
||||
hours = total_seconds // 3600
|
||||
minutes = (total_seconds % 3600) // 60
|
||||
secs = total_seconds % 60
|
||||
|
||||
if hours > 0:
|
||||
return f"{hours}:{minutes:02d}:{secs:02d}"
|
||||
else:
|
||||
return f"{minutes}:{secs:02d}"
|
||||
|
||||
|
||||
def format_pace(pace_min_per_mile: Optional[float]) -> str:
|
||||
"""Format pace in minutes per mile to MM:SS/mi"""
|
||||
if pace_min_per_mile is None:
|
||||
return "--:--/mi"
|
||||
|
||||
minutes = int(pace_min_per_mile)
|
||||
seconds = int((pace_min_per_mile - minutes) * 60)
|
||||
|
||||
return f"{minutes}:{seconds:02d}/mi"
|
||||
|
||||
|
||||
def format_speed(speed_mph: Optional[float]) -> str:
|
||||
"""Format speed to mph with one decimal place"""
|
||||
if speed_mph is None:
|
||||
return "-- mph"
|
||||
return f"{speed_mph:.1f} mph"
|
||||
|
||||
|
||||
def format_distance(distance_miles: Optional[float]) -> str:
|
||||
"""Format distance to miles with two decimal places"""
|
||||
if distance_miles is None:
|
||||
return "-- mi"
|
||||
return f"{distance_miles:.2f} mi"
|
||||
|
||||
|
||||
def format_workout_date(timestamp: Optional[str]) -> str:
|
||||
"""Format workout timestamp to human-readable date"""
|
||||
if timestamp is None:
|
||||
return "Unknown date"
|
||||
|
||||
try:
|
||||
dt = datetime.fromisoformat(timestamp.replace('Z', '+00:00'))
|
||||
return dt.strftime("%b %d, %Y %I:%M %p")
|
||||
except (ValueError, AttributeError):
|
||||
return "Invalid date"
|
||||
|
||||
|
||||
def format_workout_date_short(timestamp: Optional[str]) -> str:
|
||||
"""Format workout timestamp for tabular display (shorter format)"""
|
||||
if timestamp is None:
|
||||
return "Unknown"
|
||||
|
||||
try:
|
||||
dt = datetime.fromisoformat(timestamp.replace('Z', '+00:00'))
|
||||
return dt.strftime("%m/%d/%Y %H:%M")
|
||||
except (ValueError, AttributeError):
|
||||
return "Invalid"
|
||||
|
||||
|
||||
def format_activity_timestamp(timestamp: Optional[str]) -> str:
|
||||
"""Format activity file timestamp for tabular display"""
|
||||
if timestamp is None:
|
||||
return "Unknown"
|
||||
|
||||
try:
|
||||
# Handle both ISO formats with and without timezone info
|
||||
if timestamp.endswith('Z'):
|
||||
dt = datetime.fromisoformat(timestamp.replace('Z', '+00:00'))
|
||||
elif '+' in timestamp or '-' in timestamp[-6:]:
|
||||
# Handle timezone offset like -05:00
|
||||
dt = datetime.fromisoformat(timestamp)
|
||||
else:
|
||||
dt = datetime.fromisoformat(timestamp)
|
||||
|
||||
return dt.strftime("%Y-%m-%d %H:%M:%S")
|
||||
except (ValueError, AttributeError):
|
||||
return "Invalid date"
|
||||
|
||||
|
||||
def format_workout_summary(workout: Workout) -> str:
|
||||
"""Create a one-line summary of a workout"""
|
||||
distance = format_distance(workout.distance_miles)
|
||||
duration = format_time_duration(workout.time_seconds)
|
||||
pace = format_pace(workout.pace_min_per_mile)
|
||||
|
||||
parts = []
|
||||
if workout.distance_miles:
|
||||
parts.append(distance)
|
||||
if workout.time_seconds:
|
||||
parts.append(f"in {duration}")
|
||||
if workout.pace_min_per_mile:
|
||||
parts.append(f"({pace} pace)")
|
||||
|
||||
if parts:
|
||||
return " ".join(parts)
|
||||
else:
|
||||
return "Workout data"
|
||||
|
||||
|
||||
def estimate_calories(distance_miles: Optional[float], time_seconds: Optional[Union[int, float]], user_weight: float = 150) -> Optional[int]:
|
||||
"""Estimate calories burned based on distance and time"""
|
||||
if not distance_miles or not time_seconds:
|
||||
return None
|
||||
|
||||
pace = (time_seconds / 60) / distance_miles
|
||||
met = 8.0
|
||||
|
||||
if pace < 6:
|
||||
met = 15.0
|
||||
elif pace < 7:
|
||||
met = 12.0
|
||||
elif pace < 8:
|
||||
met = 10.0
|
||||
elif pace < 9:
|
||||
met = 8.5
|
||||
elif pace < 10:
|
||||
met = 8.0
|
||||
else:
|
||||
met = 7.0
|
||||
|
||||
return int(met * (user_weight / 2.2) * (time_seconds / 3600))
|
||||
|
||||
|
||||
# Column widths for workout table
|
||||
_COL_ID = 3
|
||||
_COL_DATE = 16
|
||||
_COL_DISTANCE = 10
|
||||
_COL_DURATION = 9
|
||||
_COL_PACE = 10
|
||||
_COL_SPEED = 9
|
||||
_COL_SOURCE = 7
|
||||
|
||||
|
||||
def print_workout_table_header():
|
||||
"""Print the header for a workout table"""
|
||||
print(f"{'ID':<{_COL_ID}} {'Date':<{_COL_DATE}} {'Distance':<{_COL_DISTANCE}} {'Duration':<{_COL_DURATION}} {'Pace':<{_COL_PACE}} {'Speed':<{_COL_SPEED}} {'Source':<{_COL_SOURCE}}")
|
||||
print("-" * (_COL_ID + _COL_DATE + _COL_DISTANCE + _COL_DURATION + _COL_PACE + _COL_SPEED + _COL_SOURCE + 6)) # +6 for spaces between columns
|
||||
|
||||
|
||||
def print_workout_row(workout: Workout):
|
||||
"""Print a single workout as a table row"""
|
||||
date = format_workout_date_short(workout.start_time or workout.created_at)
|
||||
distance = format_distance(workout.distance_miles)
|
||||
duration = format_time_duration(workout.time_seconds)
|
||||
pace = format_pace(workout.pace_min_per_mile)
|
||||
speed = format_speed(workout.speed_mph)
|
||||
source = "FIT" if workout.activity_file_id else "Manual"
|
||||
|
||||
print(f"{workout.id:<{_COL_ID}} {date:<{_COL_DATE}} {distance:<{_COL_DISTANCE}} {duration:<{_COL_DURATION}} {pace:<{_COL_PACE}} {speed:<{_COL_SPEED}} {source:<{_COL_SOURCE}}")
|
||||
|
||||
|
||||
def print_workout_details(workout: Workout):
|
||||
"""Print detailed workout information"""
|
||||
print(f"Workout #{workout.id}")
|
||||
print(f" Date: {format_workout_date(workout.start_time or workout.created_at)}")
|
||||
|
||||
if workout.distance_miles:
|
||||
print(f" Distance: {format_distance(workout.distance_miles)}")
|
||||
|
||||
if workout.time_seconds:
|
||||
print(f" Duration: {format_time_duration(workout.time_seconds)}")
|
||||
|
||||
if workout.pace_min_per_mile:
|
||||
print(f" Pace: {format_pace(workout.pace_min_per_mile)}")
|
||||
|
||||
if workout.speed_mph:
|
||||
print(f" Speed: {format_speed(workout.speed_mph)}")
|
||||
|
||||
if workout.start_time and workout.end_time:
|
||||
start = format_workout_date(workout.start_time)
|
||||
end = format_workout_date(workout.end_time)
|
||||
print(f" Start: {start}")
|
||||
print(f" End: {end}")
|
||||
|
||||
estimated_calories = estimate_calories(workout.distance_miles, workout.time_seconds)
|
||||
if estimated_calories:
|
||||
print(f" Estimated Calories: {estimated_calories}")
|
||||
|
||||
source = "FIT file" if workout.activity_file_id else "Manual entry"
|
||||
print(f" Source: {source}")
|
||||
if workout.activity_file_id:
|
||||
print(f" Activity File ID: {workout.activity_file_id}")
|
||||
|
||||
print(f" Created: {format_workout_date(workout.created_at)}")
|
||||
print(f" Updated: {format_workout_date(workout.updated_at)}")
|
||||
Reference in New Issue
Block a user