diff --git a/fitnessCog/activity_logger.py b/fitnessCog/activity_logger.py index 51143c8..04debca 100644 --- a/fitnessCog/activity_logger.py +++ b/fitnessCog/activity_logger.py @@ -1,119 +1,105 @@ -import json import csv -import os +import json +from pathlib import Path +from typing import Optional -def activity_log_path(activity: str) -> str: +def activity_log_path(activity: str, logs_dir: Optional[Path] = None) -> Path: """ - Returns the path of the activity log file in .csv format - :param activity: - :return: + Returns the path of the activity log file in .csv format. """ - filename = os.path.join("activity_logs", f"{activity}.csv") - return filename + logs_dir = logs_dir or (Path.cwd() / "activity_logs") + return logs_dir / f"{activity}.csv" -def activity_units_path(activity: str) -> str: +def activity_units_path(activity: str, units_dir: Optional[Path] = None) -> Path: """ - Returns the path of the activity activities file in .json format - :param activity: - :return: + Returns the path of the activity units file in .json format. """ - filename = os.path.join("activities", f"{activity}.json") - return filename + units_dir = units_dir or (Path.cwd() / "activities") + return units_dir / f"{activity}.json" -def is_convertable(activity: str) -> bool: - """ - Returns True if the activity has multiple units that needed to be converted. - i.e. distance may be sent in feet, but would require to be converted in meters. - :param activity: - :return: - """ - filename = activity_units_path(activity) - raw_data = json.load(open(filename)) - return raw_data.get("convertable") +def is_convertable(activity: str, units_dir: Optional[Path] = None) -> bool: + filename = activity_units_path(activity, units_dir) + with filename.open("r", encoding="utf-8") as f: + raw_data = json.load(f) + return bool(raw_data.get("convertable")) - -def create_activity_log_file(activity: str) -> None: - """ - Creates the activity log file in .csv format - :param activity: - :return: - """ - if not os.path.exists(activity_units_path(activity)): +def create_activity_log_file( + activity: str, + *, + units_dir: Optional[Path] = None, + logs_dir: Optional[Path] = None, +) -> None: + if not activity_units_path(activity, units_dir).exists(): raise ValueError(f"{activity} is not a valid activity") - if not os.path.exists("activity_logs"): - os.makedirs("activity_logs") - filename = activity_log_path(activity) - with open(filename, 'a') as f: + + logs_dir = logs_dir or (Path.cwd() / "activity_logs") + logs_dir.mkdir(parents=True, exist_ok=True) + + filename = activity_log_path(activity, logs_dir) + if filename.exists(): + return # don't re-add headers + + with filename.open("w", newline="", encoding="utf-8") as f: writer = csv.writer(f) writer.writerow(["timestamp", "username", "value"]) - -def convert_units(activity: str, value: int | float, unit: str) -> float: - """ - Returns the value of activity in a factor of 1, i.e. converting feet to meters - :param activity: - :param value: - :param unit: - :return: - """ - filename = activity_units_path(activity) - if not os.path.exists(filename): +def convert_units( + activity: str, + value: int | float, + unit: str | None, + *, + units_dir: Optional[Path] = None, +) -> float: + filename = activity_units_path(activity, units_dir) + if not filename.exists(): raise ValueError(f"{activity} is not a valid activity") # Consider None unit value as factor of 1 if unit is None: - return value + return float(value) - raw_data = json.load(open(filename)) - units_data = raw_data.get("units") + with filename.open("r", encoding="utf-8") as f: + raw_data = json.load(f) - for unit_name, data in units_data.items(): - if unit in data["aliases"]: - return value * data["factor"] + units_data = raw_data.get("units") or {} + unit_l = unit.lower() + + for _, data in units_data.items(): + aliases = [a.lower() for a in (data.get("aliases") or [])] + if unit_l in aliases: + return float(value) * float(data["factor"]) raise ValueError(f"{unit} is not a valid unit") def log_activity( - timestamp: int, - username: str, - activity: str, - value: int | float, - unit: str | None + timestamp: int, + username: str, + activity: str, + value: int | float, + unit: str | None, + *, + units_dir: Optional[Path] = None, + logs_dir: Optional[Path] = None, ) -> None: - """ - Logs the activity in .csv file - :param timestamp: - :param username: - :param activity: - :param value: - :param unit: - :return: - """ - if value <= 0 or not isinstance(value, int | float): + + if not isinstance(value, (int, float)) or value <= 0: raise ValueError(f"{value} is not a valid value") - filename = activity_log_path(activity) - if not os.path.exists(filename): - create_activity_log_file(activity) - if is_convertable(activity): - converted_value = round(convert_units(activity, value, unit), 2) + + filename = activity_log_path(activity, logs_dir) + if not filename.exists(): + create_activity_log_file(activity, units_dir=units_dir, logs_dir=logs_dir) + + if is_convertable(activity, units_dir): + converted_value = round(convert_units(activity, value, unit, units_dir=units_dir), 2) else: - converted_value = round(value, 2) - with open(filename, 'a') as f: + converted_value = round(float(value), 2) + + with filename.open("a", newline="", encoding="utf-8") as f: writer = csv.writer(f) writer.writerow([timestamp, username, converted_value]) - - -if __name__ == "__main__": - log_activity( - timestamp=0, - username="test", - activity="climb", - value=1, - unit="m" - ) diff --git a/fitnessCog/fitness.py b/fitnessCog/fitness.py index a95b134..ccdea4e 100644 --- a/fitnessCog/fitness.py +++ b/fitnessCog/fitness.py @@ -3,7 +3,7 @@ from pathlib import Path import discord from redbot.core import commands -from redbot.core.data_manager import bundled_data_path +from redbot.core.data_manager import cog_data_path from .activity_logger import log_activity @@ -15,21 +15,20 @@ class FitnessCog(commands.Cog): default_threads = "1457402451530350727,1328363409648910356" threads_raw = os.getenv("THREADS_ID", default_threads) - self.threads_id = [] + self.threads_id: list[int] = [] for x in threads_raw.split(","): x = x.strip() if x.isdigit(): self.threads_id.append(int(x)) self.confirm_reactions = os.getenv("CONFIRMATION_REACTIONS", "🐈💨") - self.bundled_activities_dir: Path = bundled_data_path(self) / "activities" - # ---- path helpers ---- - def activity_units_path(self, activity: str) -> Path: - return self.bundled_activities_dir / f"{activity}.json" + # Units are shipped with the cog in: fitnessCog/activities/*.json + self.units_dir: Path = Path(__file__).parent / "activities" - def activity_log_path(self, activity: str) -> Path: - return self.bundled_activities_dir / f"{activity}_log.json" + # Logs are writable and belong in Red's data dir + self.logs_dir: Path = cog_data_path(self) / "activity_logs" + self.logs_dir.mkdir(parents=True, exist_ok=True) @commands.Cog.listener() async def on_message(self, message: discord.Message): @@ -68,8 +67,8 @@ class FitnessCog(commands.Cog): ) return - activity = activity.lower() - path = self.activity_units_path(activity) + activity = activity.lower().strip() + path = self.units_dir / f"{activity}.json" if not path.exists(): await message.channel.send(f"No units JSON found for `{activity}`.") @@ -88,11 +87,11 @@ class FitnessCog(commands.Cog): ) return - activity = activity.lower() - path = self.activity_log_path(activity) + activity = activity.lower().strip() + path = self.logs_dir / f"{activity}.csv" if not path.exists(): - await message.channel.send(f"No log JSON found for `{activity}`.") + await message.channel.send(f"No log CSV found for `{activity}`.") return await message.channel.send(file=discord.File(fp=str(path))) @@ -124,7 +123,15 @@ class FitnessCog(commands.Cog): timestamp = int(message.created_at.timestamp()) username = message.author.name - log_activity(timestamp, username, activity, value, unit) + log_activity( + timestamp, + username, + activity, + value, + unit, + units_dir=self.units_dir, + logs_dir=self.logs_dir, + ) for r in self.confirm_reactions: await message.add_reaction(r)