Compare commits
No commits in common. "master" and "json-key-fix" have entirely different histories.
master
...
json-key-f
@ -6,8 +6,6 @@
|
||||
"aliases": [
|
||||
"meter",
|
||||
"meters",
|
||||
"metre",
|
||||
"metres",
|
||||
"m",
|
||||
"m."
|
||||
]
|
||||
@ -17,8 +15,6 @@
|
||||
"aliases": [
|
||||
"kilometer",
|
||||
"kilometers",
|
||||
"kilometre",
|
||||
"kilometres",
|
||||
"km",
|
||||
"km."
|
||||
]
|
||||
|
||||
@ -1,78 +1,80 @@
|
||||
import csv
|
||||
import json
|
||||
from pathlib import Path
|
||||
from typing import Optional
|
||||
import csv
|
||||
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")
|
||||
return logs_dir / f"{activity}.csv"
|
||||
filename = os.path.join("activity_logs", 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")
|
||||
return units_dir / f"{activity}.json"
|
||||
filename = os.path.join("activities", f"{activity}.json")
|
||||
return filename
|
||||
|
||||
|
||||
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 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 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():
|
||||
|
||||
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)):
|
||||
raise ValueError(f"{activity} is not a valid activity")
|
||||
|
||||
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:
|
||||
if not os.path.exists("activity_logs"):
|
||||
os.makedirs("activity_logs")
|
||||
filename = activity_log_path(activity)
|
||||
with open(filename, 'a') as f:
|
||||
writer = csv.writer(f)
|
||||
writer.writerow(["timestamp", "username", "value"])
|
||||
|
||||
|
||||
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():
|
||||
|
||||
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):
|
||||
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(f)
|
||||
raw_data = json.load(open(filename))
|
||||
units_data = raw_data.get("units")
|
||||
|
||||
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"])
|
||||
for unit_name, data in units_data.items():
|
||||
if unit in data["aliases"]:
|
||||
return value * data["factor"]
|
||||
|
||||
raise ValueError(f"{unit} is not a valid unit")
|
||||
|
||||
@ -82,24 +84,36 @@ def log_activity(
|
||||
username: str,
|
||||
activity: str,
|
||||
value: int | float,
|
||||
unit: str | None,
|
||||
*,
|
||||
units_dir: Optional[Path] = None,
|
||||
logs_dir: Optional[Path] = None,
|
||||
unit: str | 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, 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)
|
||||
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)
|
||||
else:
|
||||
converted_value = round(float(value), 2)
|
||||
|
||||
with filename.open("a", newline="", encoding="utf-8") as f:
|
||||
converted_value = round(value, 2)
|
||||
with open(filename, 'a') 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"
|
||||
)
|
||||
|
||||
@ -1,143 +1,78 @@
|
||||
import os
|
||||
from pathlib import Path
|
||||
|
||||
import discord
|
||||
from redbot.core import commands
|
||||
from redbot.core.data_manager import cog_data_path
|
||||
from redbot.core import Config, commands
|
||||
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[int] = []
|
||||
for x in threads_raw.split(","):
|
||||
x = x.strip()
|
||||
if x.isdigit():
|
||||
self.threads_id.append(int(x))
|
||||
|
||||
self.threads_id = list(map(int, os.getenv("THREADS_ID", default_threads).split(",")))
|
||||
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
|
||||
|
||||
content = (message.content or "").strip()
|
||||
if not content.startswith("!"):
|
||||
if not message.content.startswith("!"):
|
||||
return
|
||||
|
||||
cmdline = content[1:].strip().lower()
|
||||
if not cmdline:
|
||||
# region Fitness commands
|
||||
if message.content.startswith("!getunits"):
|
||||
await self.get_units_file(message)
|
||||
return
|
||||
|
||||
cmd, *rest = cmdline.split(" ", 1)
|
||||
arg = rest[0].strip() if rest else ""
|
||||
|
||||
if cmd == "getunits":
|
||||
await self.get_units_file(message, arg)
|
||||
if message.content.startswith("!getlog"):
|
||||
await self.get_log_file(message)
|
||||
return
|
||||
|
||||
if cmd == "getlog":
|
||||
await self.get_log_file(message, arg)
|
||||
# Make sure this one is past every other ! commands
|
||||
await self.fitness_activity_log(message)
|
||||
return
|
||||
# endregion Fitness commands
|
||||
|
||||
await self.log_fitness(message)
|
||||
|
||||
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:
|
||||
await message.channel.send(
|
||||
"Usage: `!getunits <activity>` (example: `!getunits pushups`)"
|
||||
)
|
||||
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)))
|
||||
|
||||
raw = message.content[1:].lower().strip().split(" ", 1)
|
||||
filename = activity_units_path(raw[1])
|
||||
await message.channel.send(file=discord.File(fp=filename))
|
||||
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:
|
||||
await message.channel.send(
|
||||
"Usage: `!getlog <activity>` (example: `!getlog pushups`)"
|
||||
)
|
||||
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)))
|
||||
|
||||
raw = message.content[1:].lower().strip().split(" ", 1)
|
||||
filename = activity_log_path(raw[1])
|
||||
await message.channel.send(file=discord.File(fp=filename))
|
||||
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()
|
||||
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"
|
||||
raw = message.content[1:].lower().strip().split(" ", 2)
|
||||
|
||||
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(
|
||||
timestamp,
|
||||
username,
|
||||
activity,
|
||||
value,
|
||||
unit,
|
||||
units_dir=self.units_dir,
|
||||
logs_dir=self.logs_dir,
|
||||
)
|
||||
log_activity(timestamp, username, activity, value, unit)
|
||||
|
||||
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))
|
||||
|
||||
|
||||
Loading…
x
Reference in New Issue
Block a user