106 lines
3.1 KiB
Python
106 lines
3.1 KiB
Python
import csv
|
|
import json
|
|
from pathlib import Path
|
|
from typing import Optional
|
|
|
|
|
|
def activity_log_path(activity: str, logs_dir: Optional[Path] = None) -> Path:
|
|
"""
|
|
Returns the path of the activity log file in .csv format.
|
|
"""
|
|
logs_dir = logs_dir or (Path.cwd() / "activity_logs")
|
|
return logs_dir / f"{activity}.csv"
|
|
|
|
|
|
def activity_units_path(activity: str, units_dir: Optional[Path] = None) -> Path:
|
|
"""
|
|
Returns the path of the activity units file in .json format.
|
|
"""
|
|
units_dir = units_dir or (Path.cwd() / "activities")
|
|
return units_dir / f"{activity}.json"
|
|
|
|
|
|
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,
|
|
*,
|
|
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")
|
|
|
|
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 | 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 float(value)
|
|
|
|
with filename.open("r", encoding="utf-8") as f:
|
|
raw_data = json.load(f)
|
|
|
|
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,
|
|
*,
|
|
units_dir: Optional[Path] = None,
|
|
logs_dir: Optional[Path] = None,
|
|
) -> None:
|
|
|
|
if not isinstance(value, (int, float)) or value <= 0:
|
|
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)
|
|
else:
|
|
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])
|