Kanium-PUBLIC/KaniumCogs
5
0
Fork 0
Code Issues 6 Pull Requests Packages Projects 1 Releases Wiki Activity

Merge pull request 'Aligning activity logger' (#17) from debug_fitness_more into master

Browse Source 
Reviewed-on: #17
This commit is contained in:
AllfatherHatt 2026-02-04 18:25:08 +01:00
commit fa1be19767
2 changed files with 92 additions and 97 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

156
fitnessCog/activity_logger.py
View File

@ -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"
)

33
fitnessCog/fitness.py
View File

@ -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):
@ -67,7 +66,8 @@ class FitnessCog(commands.Cog):
)
return
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}`.")
@ -86,10 +86,11 @@ class FitnessCog(commands.Cog):
)
return
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)))
@ -121,7 +122,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)