Skip to content

Instantly share code, notes, and snippets.

@OPVL
Created July 13, 2024 15:29

Revisions

  1. OPVL created this gist Jul 13, 2024.
    303 changes: 303 additions & 0 deletions quantity_unit.py
    Original file line number Diff line number Diff line change
    @@ -0,0 +1,303 @@
    ### 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)