Script to import quantity units & associated conversion factors in to grocy, pythonically

quantity_unit.py

### This script adds metric (UK) units and unit conversions to grocy
### inspired by https://gist.github.com/randompherret/f772d43cb618b55997f203e023631d5f
### this script is by no means perfect however it can handle failed runs and restarting midway through
### usage: python quantity_units.py

### make sure to set the environment variables in a .env file in the same directory as this script
### GROCY_API_KEY and GROCY_BASE_URL are required
### LOG_TO_FILE and DELETE_EXISTING_UNITS are optional
### LOG_TO_FILE will log the added units and conversions to a file (not used for anythning just interesting)
### DELETE_EXISTING_UNITS will delete all existing units in grocy before adding new ones

### I don't recommend using this on a live instance of grocy, create a clone and test it on there.
### on existing instances without the DELETE_EXISTING_UNITS flag set to true, it will not add duplicates or conversions
### however if the names do not match exactly it will add them as new units / conversions

### Example .env file
# GROCY_BASE_URL=http://grocy.opvl/api
# GROCY_API_KEY=YOUR_GROCY_API_KEY
# LOG_TO_FILE=false
# DELETE_EXISTING_UNITS=false

### Requirements
# python-dotenv
# requests


import os
from typing import Any

import requests
import json

import dotenv

dotenv.load_dotenv()

grocy_api_key = os.getenv("GROCY_API_KEY")
grocy_base_url = os.getenv("GROCY_BASE_URL")
log_to_file = bool(os.getenv("LOG_TO_FILE"))
delete_existing_units = bool(os.getenv("DELETE_EXISTING_UNITS"))

class Unit:
    id: int | None
    name: str
    name_plural: str
    description: str

    def __init__(self, name: str, name_plural: str, description: str):
        self.id = None
        self.name = name
        self.name_plural = name_plural
        self.description = description

    def toJson(self):
        return {
            "name": self.name,
            "name_plural": self.name_plural,
            "description": self.description,
        }


class UnitConversion:
    from_unit_name: str
    to_unit_name: str
    factor: float
    from_unit_id: int | None
    to_unit_id: int | None

    def __init__(self, from_unit_name: str, to_unit_name: str, factor: float):
        self.from_unit_name = from_unit_name
        self.to_unit_name = to_unit_name
        self.factor = factor

    def toJson(self):
        if self.from_unit_name is None or self.to_unit_name is None:
            raise ValueError("Unit ids must be set before converting to json")

        return {
            "from_qu_id": self.from_unit_id,
            "to_qu_id": self.to_unit_id,
            "factor": self.factor,
        }


units = [
    Unit(name="g", name_plural="g", description="Gram"),
    Unit(name="kg", name_plural="kg", description="Kilogram"),
    Unit(name="ml", name_plural="ml", description="Milliliter"),
    Unit(name="l", name_plural="l", description="Liter"),
    Unit(name="tsp", name_plural="tsp", description="Teaspoon"),
    Unit(name="tbsp", name_plural="tbsp", description="Tablespoon"),
    Unit(name="cup", name_plural="cups", description="Cup"),
    Unit(name="floz", name_plural="floz", description="Fluid Ounce"),
    Unit(name="pt", name_plural="pt", description="Pint"),
    Unit(name="qt", name_plural="qt", description="Quart"),
    Unit(name="gal (UK)", name_plural="gal (UK)", description="Gallon (UK)"),
    Unit(name="gal", name_plural="gal", description="Gallon (US)"),
    Unit(name="oz", name_plural="oz", description="Ounce"),
    Unit(name="lb", name_plural="lb", description="Pound"),
    Unit(name="pack", name_plural="packs", description="Pack"),
    Unit(name="piece", name_plural="pieces", description="Piece"),
]

unit_conversions = [
    UnitConversion(from_unit_name="g", to_unit_name="kg", factor=0.001),
    UnitConversion(from_unit_name="kg", to_unit_name="lb", factor=2.20462),
    UnitConversion(from_unit_name="kg", to_unit_name="oz", factor=35.274),
    UnitConversion(from_unit_name="kg", to_unit_name="cup", factor=8.511),
    UnitConversion(from_unit_name="cup", to_unit_name="ml", factor=236.588),
    UnitConversion(from_unit_name="cup", to_unit_name="floz", factor=8),
    UnitConversion(from_unit_name="cup", to_unit_name="tbsp", factor=16),
    UnitConversion(from_unit_name="cup", to_unit_name="tsp", factor=48),
    UnitConversion(from_unit_name="cup", to_unit_name="g", factor=201.6),
    UnitConversion(from_unit_name="l", to_unit_name="floz", factor=33.814),
    UnitConversion(from_unit_name="l", to_unit_name="ml", factor=1000),
    UnitConversion(from_unit_name="tsp", to_unit_name="ml", factor=5),
    UnitConversion(from_unit_name="tsp", to_unit_name="floz", factor=0.166667),
    UnitConversion(from_unit_name="tsp", to_unit_name="tbsp", factor=3),
    UnitConversion(from_unit_name="tsp", to_unit_name="g", factor=4.92892),
    UnitConversion(from_unit_name="tbsp", to_unit_name="ml", factor=15),
    UnitConversion(from_unit_name="tbsp", to_unit_name="floz", factor=0.5),
    UnitConversion(from_unit_name="tbsp", to_unit_name="g", factor=14.7868),
    UnitConversion(from_unit_name="floz", to_unit_name="l", factor=0.0295735),
    UnitConversion(from_unit_name="floz", to_unit_name="cup", factor=0.125),
    UnitConversion(from_unit_name="floz", to_unit_name="ml", factor=28.4130625),
    UnitConversion(from_unit_name="pt", to_unit_name="floz", factor=16),
    UnitConversion(from_unit_name="pt", to_unit_name="ml", factor=473.176),
    UnitConversion(from_unit_name="pt", to_unit_name="cup", factor=2),
    UnitConversion(from_unit_name="pt", to_unit_name="l", factor=0.473176),
    UnitConversion(from_unit_name="qt", to_unit_name="pt", factor=2),
    UnitConversion(from_unit_name="qt", to_unit_name="floz", factor=32),
    UnitConversion(from_unit_name="qt", to_unit_name="ml", factor=946.353),
    UnitConversion(from_unit_name="gal", to_unit_name="qt", factor=4),
    UnitConversion(from_unit_name="gal", to_unit_name="pt", factor=8),
    UnitConversion(from_unit_name="gal", to_unit_name="floz", factor=128),
    UnitConversion(from_unit_name="gal", to_unit_name="ml", factor=3785.41),
    UnitConversion(from_unit_name="gal", to_unit_name="cup", factor=16),
    UnitConversion(from_unit_name="gal", to_unit_name="l", factor=3.78541),
    UnitConversion(from_unit_name="gal (UK)", to_unit_name="l", factor=4.54609),
    UnitConversion(from_unit_name="gal (UK)", to_unit_name="floz", factor=160),
    UnitConversion(from_unit_name="gal (UK)", to_unit_name="ml", factor=4546.09),
    UnitConversion(from_unit_name="oz", to_unit_name="lb", factor=0.0625),
    UnitConversion(from_unit_name="oz", to_unit_name="g", factor=28.3495),
    UnitConversion(from_unit_name="lb", to_unit_name="g", factor=453.592),
]


def process_units(existing_units) -> None:
    # add new units to grocy
    for unit in units:
        if any(grocy_unit["name"] == unit.name for grocy_unit in existing_units):
            print(f"Unit {unit.name} already exists in grocy. Skipping...")
            # set the id of the unit to the one in grocy
            unit.id = next(
                grocy_unit["id"]
                for grocy_unit in existing_units
                if grocy_unit["name"] == unit.name
            )
            continue

        # remove id from unit before sending to grocy
        response = requests.post(
            grocy_base_url + "/objects/quantity_units",
            headers={"GROCY-API-KEY": grocy_api_key},
            json=unit.toJson(),
        )
        if response.status_code != 200:
            print(f"Error adding unit {unit.name} to grocy")
            response.raise_for_status()
            continue

        unit.id = response.json()["created_object_id"]
        print(f"Added unit {unit.name} - {unit.id} to grocy")

    if log_to_file:
        with open("units/added_units.json", "w+") as f:
            json.dump([unit.toJson() for unit in units], f)


def add_conversion_for_unit(unit: Unit) -> None:

    # get all unit conversions currently stored in grocy
    response = requests.get(
        grocy_base_url + "/objects/quantity_unit_conversions",
        headers={"GROCY-API-KEY": grocy_api_key},
    )
    if response.status_code != 200:
        print("Error getting unit conversions from grocy")

    grocy_unit_conversions = response.json()
    if log_to_file:
        with open(f"units/conversions/{unit.name}.json", "w+") as f:
            json.dump(grocy_unit_conversions, f)

    # get all unit conversions for current unit by name
    unit_conversions_for_unit = [
        conversion
        for conversion in unit_conversions
        if conversion.from_unit_name == unit.name
    ]
    if len(unit_conversions_for_unit) == 0:
        print(f"No conversions found for unit {unit.name}")

        return

    # set the unit ids for the conversions
    for conversion in unit_conversions_for_unit:
        conversion.from_unit_id = next(
            grocy_unit.id
            for grocy_unit in units
            if grocy_unit.name == conversion.from_unit_name
        )
        conversion.to_unit_id = next(
            grocy_unit.id
            for grocy_unit in units
            if grocy_unit.name == conversion.to_unit_name
        )

        # check if the conversion already exists in grocy
        if any(
            existing_conversion["from_qu_id"] == conversion.from_unit_id
            and existing_conversion["to_qu_id"] == conversion.to_unit_id
            for existing_conversion in grocy_unit_conversions
        ):
            print(
                f"Conversion from {conversion.from_unit_name} to {conversion.to_unit_name} already exists in grocy. Skipping..."
            )
            continue

        response = requests.post(
            grocy_base_url + "/objects/quantity_unit_conversions",
            headers={"GROCY-API-KEY": grocy_api_key},
            json=conversion.toJson(),
        )
        if response.status_code != 200:
            if "already exists" in response.text:
                print(
                    f"Conversion from {conversion.from_unit_name} to {conversion.to_unit_name} already exists in grocy. Skipping..."
                )
                continue

            response.raise_for_status()

        print(
            f"Added conversion from {conversion.from_unit_name} to {conversion.to_unit_name} to grocy"
        )

    if log_to_file:
        with open(f"units/conversions/{unit.name}_added.json", "w+") as f:
            json.dump(
                [conversion.toJson() for conversion in unit_conversions_for_unit], f
            )


def get_existing_units() -> list[dict[str, Any]]:
    response = requests.get(
        grocy_base_url + "/objects/quantity_units",
        headers={"GROCY-API-KEY": grocy_api_key},
    )
    if response.status_code != 200:
        print("Error getting units from grocy")
        response.raise_for_status()

    if log_to_file:
        with open("units/existing_units.json", "w+") as f:
            json.dump(response.json(), f)
    return response.json()


def clear_all_units(unit_ids: list[int]) -> None:
    for unit_id in unit_ids:
        response = requests.delete(
            grocy_base_url + f"/objects/quantity_units/{unit_id}",
            headers={"GROCY-API-KEY": grocy_api_key},
        )
        if response.status_code != 204:
            print(f"Error deleting unit {unit_id} from grocy")
            response.raise_for_status()
            continue

        print(f"Deleted unit {unit_id} from grocy")

    if log_to_file:
        with open("units/deleted_units.json", "w+") as f:
            json.dump(unit_ids, f)


if __name__ == "__main__":

    if not grocy_api_key or not grocy_base_url:
        raise ValueError("GROCY_API_KEY and GROCY_BASE_URL must be set in .env")

    if log_to_file:
        os.makedirs("units", exist_ok=True)
        os.makedirs("units/conversions", exist_ok=True)

    if delete_existing_units:
        existing_units = get_existing_units()
        clear_all_units([unit["id"] for unit in existing_units])

    process_units(get_existing_units())
    for unit in units:
        add_conversion_for_unit(unit)