fitnessCog/activities/climb.json

@@ -6,8 +6,6 @@
             "aliases": [
                 "meter",
                 "meters",
-                "metre",
-                "metres",
                 "m",
                 "m."
             ]
@@ -17,8 +15,6 @@
             "aliases": [
                 "kilometer",
                 "kilometers",
-                "kilometre",
-                "kilometres",
                 "km",
                 "km."
             ]

fitnessCog/activity_logger.py

@@ -1,105 +1,119 @@
-import csv
 import json
-from pathlib import Path
+import csv
-from typing import Optional
+import os
 
 
-def activity_log_path(activity: str, logs_dir: Optional[Path] = None) -> Path:
+def activity_log_path(activity: str) -> str:
     """
-    Returns the path of the activity log file in .csv format.
+    Returns the path of the activity log file in .csv format
+    :param activity:
+    :return:
     """
-    logs_dir = logs_dir or (Path.cwd() / "activity_logs")
+    filename = os.path.join("activity_logs", f"{activity}.csv")
-    return logs_dir / f"{activity}.csv"
+    return filename
 
 
-def activity_units_path(activity: str, units_dir: Optional[Path] = None) -> Path:
+def activity_units_path(activity: str) -> str:
     """
-    Returns the path of the activity units file in .json format.
+    Returns the path of the activity activities file in .json format
+    :param activity:
+    :return:
     """
-    units_dir = units_dir or (Path.cwd() / "activities")
+    filename = os.path.join("activities", f"{activity}.json")
-    return units_dir / f"{activity}.json"
+    return filename
 
 
-def is_convertable(activity: str, units_dir: Optional[Path] = None) -> bool:
+def is_convertable(activity: str) -> bool:
-    filename = activity_units_path(activity, units_dir)
+    """
-    with filename.open("r", encoding="utf-8") as f:
+    Returns True if the activity has multiple units that needed to be converted.
-        raw_data = json.load(f)
+    i.e. distance may be sent in feet, but would require to be converted in meters.
-    return bool(raw_data.get("convertable"))
+    :param activity:
+    :return:
+    """
+    filename = activity_units_path(activity)
+    raw_data = json.load(open(filename))
+    return raw_data.get("transformable")
 
 
-def create_activity_log_file(
+
-    activity: str,
+def create_activity_log_file(activity: str) -> None:
-    *,
+    """
-    units_dir: Optional[Path] = None,
+    Creates the activity log file in .csv format
-    logs_dir: Optional[Path] = None,
+    :param activity:
-) -> None:
+    :return:
-    if not activity_units_path(activity, units_dir).exists():
+    """
+    if not os.path.exists(activity_units_path(activity)):
         raise ValueError(f"{activity} is not a valid activity")
-
+    if not os.path.exists("activity_logs"):
-    logs_dir = logs_dir or (Path.cwd() / "activity_logs")
+        os.makedirs("activity_logs")
-    logs_dir.mkdir(parents=True, exist_ok=True)
+    filename = activity_log_path(activity)
-
+    with open(filename, 'a') as f:
-    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,
+def convert_units(activity: str, value: int | float, unit: str) -> float:
-    value: int | float,
+    """
-    unit: str | None,
+    Returns the value of activity in a factor of 1, i.e. converting feet to meters
-    *,
+    :param activity:
-    units_dir: Optional[Path] = None,
+    :param value:
-) -> float:
+    :param unit:
-    filename = activity_units_path(activity, units_dir)
+    :return:
-    if not filename.exists():
+    """
+    filename = activity_units_path(activity)
+    if not os.path.exists(filename):
         raise ValueError(f"{activity} is not a valid activity")
 
     # Consider None unit value as factor of 1
     if unit is None:
-        return float(value)
+        return value
 
-    with filename.open("r", encoding="utf-8") as f:
+    raw_data = json.load(open(filename))
-        raw_data = json.load(f)
+    units_data = raw_data.get("units")
 
-    units_data = raw_data.get("units") or {}
+    for unit_name, data in units_data.items():
-    unit_l = unit.lower()
+        if unit in data["aliases"]:
-
+            return value * data["factor"]
-    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,
+        timestamp: int,
-    username: str,
+        username: str,
-    activity: str,
+        activity: str,
-    value: int | float,
+        value: int | float,
-    unit: str | None,
+        unit: str | None
-    *,
-    units_dir: Optional[Path] = None,
-    logs_dir: Optional[Path] = None,
 ) -> None:
-    
+    """
-    if not isinstance(value, (int, float)) or value <= 0:
+    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):
         raise ValueError(f"{value} is not a valid value")
-
+    filename = activity_log_path(activity)
-    filename = activity_log_path(activity, logs_dir)
+    if not os.path.exists(filename):
-    if not filename.exists():
+        create_activity_log_file(activity)
-        create_activity_log_file(activity, units_dir=units_dir, logs_dir=logs_dir)
+    if is_convertable(activity):
-
+        converted_value = round(convert_units(activity, value, unit), 2)
-    if is_convertable(activity, units_dir):
-        converted_value = round(convert_units(activity, value, unit, units_dir=units_dir), 2)
     else:
-        converted_value = round(float(value), 2)
+        converted_value = round(value, 2)
-
+    with open(filename, 'a') as f:
-    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"
+    )

fitnessCog/fitness.py

@@ -1,143 +1,78 @@
 import os
-from pathlib import Path
-
 import discord
-from redbot.core import commands
+from redbot.core import Config, commands
-from redbot.core.data_manager import cog_data_path
+from .activity_logger import log_activity, activity_log_path, activity_units_path
 
-from .activity_logger import log_activity
 
 
 class FitnessCog(commands.Cog):
     def __init__(self, bot):
         self.bot = bot
-
+        # Some hardcoded values, this probably should be changed with redbot config. IDK how to use it or what's in it.
         default_threads = "1457402451530350727,1328363409648910356"
-        threads_raw = os.getenv("THREADS_ID", default_threads)
+        self.threads_id = list(map(int, os.getenv("THREADS_ID", default_threads).split(",")))
-
-        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", "🐈💨")
 
-        # Units are shipped with the cog in: fitnessCog/activities/*.json
-        self.units_dir: Path = Path(__file__).parent / "activities"
-
-        # 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):
+    async def on_message(self, message):
         if message.author.bot:
             return
         if message.channel.id not in self.threads_id:
             return
-
+        if not message.content.startswith("!"):
-        content = (message.content or "").strip()
-        if not content.startswith("!"):
             return
 
-        cmdline = content[1:].strip().lower()
+        # region Fitness commands
-        if not cmdline:
+        if message.content.startswith("!getunits"):
+            await self.get_units_file(message)
             return
-
+        if message.content.startswith("!getlog"):
-        cmd, *rest = cmdline.split(" ", 1)
+            await self.get_log_file(message)
-        arg = rest[0].strip() if rest else ""
-
-        if cmd == "getunits":
-            await self.get_units_file(message, arg)
             return
-
+        # Make sure this one is past every other ! commands
-        if cmd == "getlog":
-            await self.get_log_file(message, arg)
-            return
-
         await self.log_fitness(message)
+        return
+        # endregion Fitness commands
 
-    async def get_units_file(self, message: discord.Message, activity: str):
+    # region Fitness commands functions
+    @staticmethod
+    async def get_units_file(message: discord.Message):
         try:
-            if not activity:
+            raw = message.content[1:].lower().strip().split(" ", 1)
-                await message.channel.send(
+            filename = activity_units_path(raw[1])
-                    "Usage: `!getunits <activity>` (example: `!getunits pushups`)"
+            await message.channel.send(file=discord.File(fp=filename))
-                )
-                return
-
-            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}`.")
-                return
-
-            await message.channel.send(file=discord.File(fp=str(path)))
-
         except Exception as e:
             await message.channel.send(str(e))
 
-    async def get_log_file(self, message: discord.Message, activity: str):
+    @staticmethod
+    async def get_log_file(message: discord.Message):
         try:
-            if not activity:
+            raw = message.content[1:].lower().strip().split(" ", 1)
-                await message.channel.send(
+            filename = activity_log_path(raw[1])
-                    "Usage: `!getlog <activity>` (example: `!getlog pushups`)"
+            await message.channel.send(file=discord.File(fp=filename))
-                )
-                return
-
-            activity = activity.lower().strip()
-            path = self.logs_dir / f"{activity}.csv"
-
-            if not path.exists():
-                await message.channel.send(f"No log CSV found for `{activity}`.")
-                return
-
-            await message.channel.send(file=discord.File(fp=str(path)))
-
         except Exception as e:
             await message.channel.send(str(e))
 
-    async def log_fitness(self, message: discord.Message):
+    async def log_fitness(self, message):
         try:
-            parts = message.content[1:].strip().split()
+            raw = message.content[1:].lower().strip().split(" ", 2)
-            if len(parts) < 2:
-                await message.reply(
-                    "Usage: `!<activity> <value> [unit]` (example: `!pushups 20` or `!run 5 km`)"
-                )
-                return
-
-            activity = parts[0]
-
-            try:
-                value = float(parts[1])
-            except ValueError:
-                await message.reply(
-                    "Invalid number. Usage: `!<activity> <value> [unit]` (example: `!run 5 km`)"
-                )
-                return
-
-            unit = parts[2] if len(parts) > 2 else "m"
 
             timestamp = int(message.created_at.timestamp())
             username = message.author.name
+            activity = raw[0]
+            value = float(raw[1])
+            unit = raw[2] if len(raw) > 2 else "m"
 
-            log_activity(
+            log_activity(timestamp, username, activity, value, unit)
-                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)
 
         except Exception as e:
             await message.reply(str(e))
+    # endregion Fitness commands functions
 
 
 def setup(bot):
     bot.add_cog(FitnessCog(bot))
+