example_python_importer.md

Setup client

# !pip install farmOS==1.0.0b3
from farmOS import farmOS
hostname="http://localhost"
client = farmOS(hostname, client_id="farm", scope="farm_manager")
current_token = client.authorize()

Helper functions

# Define a couple helper functions to use in the importer script below.
def _get_existing_asset_by_name(asset_type, asset_name):
    """
    Helper function to laod an existing asset by name.
    :param asset_type: The asset type to lookup, eg: plant.
    :param asset_name: The asset name to lookup.
    :return: The asset data or None if not found.
    """

    asset_filter = client.filter('name', asset_name)
    assets = list(client.asset.iterate(asset_type, asset_filter))

    if (len(assets) == 0):
        return None
    else:
        return assets[0]

def _get_existing_log_by_name_timestamp(log_type, log_name, timestamp):
    """
    Helper function to load an existing log by name.
    """

    log_filter = {
        **client.filter('name', log_name),
        **client.filter('timestamp', int(timestamp))
    }
    logs = list(client.log.iterate(log_type, log_filter))

    if (len(logs) == 0):
        return None
    else:
        return logs[0]

def _get_or_create_taxonomy_term(vocabulary, term_name):
    """
    Helper function to load an existing or create a new taxonomy term by name.
    :param vocabulary: The taxonomy vocabulary.
    :param term_name: The term name.
    :return: The term data or None if not found.
    """

    term_filter = client.filter('name', term_name)
    terms = list(client.resource.iterate('taxonomy_term', vocabulary, term_filter))

    # Return the first matching term.
    if (len(terms) > 0):
        return terms[0]
    # Else create a new term.
    else:
        term = {
            "attributes": {
                "name": term_name,
            },
        }
        new_term = client.term.send(vocabulary, term)
        return new_term["data"]

Crop importer (creates plant assets, then creates a transplanting log assigning the plant to a location)

import logging
import re
import csv
from datetime import datetime
from time import time

# Specify the file path.
file_path = '/content/farmOS/content/data/uploaded/crop-plan-20220601-AssetsData.csv'


# Prompt the user to avoid running multiple times.
res = input(f"Import crop data from file: {file_path}")

if res is 'y' or res is 'yes':
    print('importing')

    # Open the CSV file.
    with open(file_path, newline='') as csv_file:
        # Iterate through each row of the CSV file.
        reader = csv.DictReader(csv_file, delimiter=',')
        for row in reader:

            # Required values.
            asset_name = row['Name']
            existing_asset = _get_existing_asset_by_name('plant', asset_name)
            if existing_asset is not None:
                print(f"Row {reader.line_num}: Asset already exists: '{asset_name}'")
                continue
            if (asset_name is None or len(asset_name) == 0):
                print(f"Row {reader.line_num}: No asset name")
                continue
            crop_variety = row['Crop/variety']
            if (crop_variety is None or len(crop_variety) == 0):
                print(f"Row {reader.line_num}: No Crop/variety")
                continue
            asset_location_name = row['Area/Location']
            if (asset_location_name is None or len(asset_location_name) == 0):
                print(f"Row {reader.line_num}: No asset location")
                continue

            asset_location = _get_existing_asset_by_name('land', asset_location_name)
            if (asset_location is None):
                print(f"Row {reader.line_num}: Could not find existing location: '{asset_location_name}'")
                continue

            new_asset_data = {
                "attributes": {
                    "type": "plant",
                    "status": 'active',
                    "name": asset_name,
                    "notes": row['Description'],
                },
                "relationships": {
                }
            }

            plant_type = _get_or_create_taxonomy_term("plant_type", crop_variety)
            new_asset_data["relationships"]["plant_type"] = {
                "data": [
                    {
                        "type": "taxonomy_term--plant_type",
                        "id": plant_type["id"],
                    }
                ]
            }

            season_value = row['Season']
            if len(season_value) != 0:
                season = _get_or_create_taxonomy_term("season", season_value)
                new_asset_data["relationships"]["season"] = {
                    "data": [
                        {
                            "type": "taxonomy_term--season",
                            "id": season["id"]
                        }
                    ]
                }

            # Create the asset.
            new_asset = client.asset.send('plant', new_asset_data)

            # Load log values.

            # Convert date string to timestamp.
            now = time()
            pattern = r"\d{4}[-][W]\d{1,2}"
            year_week = re.match(pattern, asset_name)
            timestamp = datetime.strptime(year_week.group(0) + '-1', '%Y-W%W-%w').timestamp()

            # Log status
            #raw_status = int(row['Done'])
            #status = "done" if raw_status == 1 else "pending"
            # TODO Calculate status from the date.
            status = "done"

            # Notes
            # TODO build notes from quantities.
            #notes = row['Notes']

            # Create the transplanting log referencing the plant asset.
            log = {
                "attributes": {
                    "timestamp": timestamp,
                    "status": status,
                    #"notes": notes,
                    "is_movement": True,
                },
                "relationships": {
                    "asset": {
                        "data": [
                            {
                                "type": "asset--plant",
                                "id": new_asset["data"]["id"],
                            }
                        ]
                    },
                    "location": {
                        "data": [
                            {
                                "type": "asset--land",
                                "id": asset_location["id"],
                            }
                        ]
                    }
                }
            }

            # Create a quantity for the log if length is provided.
            if 'length FB' in row:
                length_fb = row['length FB']
                nr_plants_fb = row['Nr plants FB']
                if all(len(x) != 0 for x in [length_fb, nr_plants_fb]):
                    length_quantity_data = {
                        "attributes": {
                            "type": "standard",
                            "measure": "length",
                            "value": {
                                "decimal": length_fb,
                            },
                            "label": "Length FB",
                        },
                        "relationships": {
                        }
                    }
                    # Create the quantity.
                    length_quantity = client.resource.send('quantity', 'standard', length_quantity_data)

                    count_quantity_data = {
                        "attributes": {
                            "type": "standard",
                            "measure": "count",
                            "value": {
                                "decimal": nr_plants_fb,
                            },
                            "label": "Nr plants FB",
                        },
                        "relationships": {
                        }
                    }
                    count_quantity = client.resource.send('quantity', 'standard', count_quantity_data)

                    # Add quantity to log.
                    log["relationships"]["quantity"] = {
                        "data": [
                            {
                                "type": "quantity--standard",
                                "id": length_quantity["data"]["id"],
                                "meta": {
                                    "target_revision_id": length_quantity["data"]["attributes"]["drupal_internal__revision_id"]
                                }
                            },
                            {
                                "type": "quantity--standard",
                                "id": count_quantity["data"]["id"],
                                "meta": {
                                    "target_revision_id": count_quantity["data"]["attributes"]["drupal_internal__revision_id"]
                                }
                            }
                        ]
                    }

            new_log = client.log.send("transplanting", log)

            link = "{hostname}/asset/{id}".format(hostname=hostname,
                                                  id=new_asset["data"]["attributes"]["drupal_internal__id"])
            print(
                "Imported log for asset: {name} - {link}".format(name=new_asset["data"]["attributes"]["name"],
                                                                  link=link))

Harvest importer

import logging
import re
import csv
from datetime import datetime
from time import time

# Specify the file path.
file_path ='/content/farmOS/content/data/uploaded/log_farm_harvest_weekX - week21_2022.csv'


# Prompt the user to avoid running multiple times.
res = input(f"Import harvest data from file: {file_path}")

if res is 'y' or res is 'yes':
    print('importing')

    # Open the CSV file.
    with open(file_path, newline='') as csv_file:
        # Iterate through each row of the CSV file.
        reader = csv.DictReader(csv_file, delimiter=',')
        for row in reader:

            # Required values.
            log_name = row['Log name']
            if (log_name is None or len(log_name) == 0):
                print(f"Row {reader.line_num}: No log name")
                continue
            date = row['Date']
            if (date is None or len(date) == 0):
                print(f"Row {reader.line_num}: No date")
                continue
            # Convert date string to timestamp.
            timestamp = datetime.strptime(date, '%Y-%m-%d').timestamp()

            # Ensure the log doesn't exist.
            existing_log = _get_existing_log_by_name_timestamp('harvest', log_name, timestamp)
            if existing_log is not None:
                print(f"Row {reader.line_num}: Log already exists: '{log_name}'")
                continue

            asset_name = row['Asset names']
            if (asset_name is None or len(asset_name) == 0):
                print(f"Row {reader.line_num}: No asset name")
                continue
            plant_asset = _get_existing_asset_by_name('plant', asset_name)
            if (plant_asset is None):
                print(f"Row {reader.line_num}: Could not find existing plant asset: '{asset_name}'")
                continue
            
            # Quantity value is required.
            quantity_value = row['Quantity value']
            if (quantity_value is None or len(quantity_value) == 0):
                print(f"Row {reader.line_num}: No quantity value.")
                continue

            # Load log values.

            # Log status
            raw_status = int(bool(row['Done']))
            status = "done" if raw_status == 1 else "pending"

            # Notes
            notes = row['Notes']

            # Create the transplanting log referencing the plant asset.
            log = {
                "attributes": {
                    "type": "harvest",
                    "timestamp": timestamp,
                    "status": status,
                    "name": log_name,
                    "notes": notes,
                },
                "relationships": {
                    "asset": {
                        "data": [
                            {
                                "type": "asset--plant",
                                "id": plant_asset["id"],
                            }
                        ]
                    },
                }
            }

            # Add category.
            category = row["Categories"]
            if len(category) != 0:
                category_term = _get_or_create_taxonomy_term("log_category", category)
                log["relationships"]["category"] = {
                    "data": {
                        "type": "taxonomy_term--log_category",
                        "id": category_term["id"],
                    }
                }

            # Add lot number.
            lot_number = row["Lot number"]
            if len(lot_number) != 0:
                log["attributes"]["lot_number"] = lot_number

            # Create a quantity for the log.
            quantity_value = row['Quantity value']
            if len(quantity_value) != 0:
                decimal = float(quantity_value.replace(",", "."))
                harvest_quantity_data = {
                    "attributes": {
                        "type": "standard",
                        "measure": "weight",
                        "value": {
                            "decimal": decimal,
                        },
                    },
                    "relationships": {
                    }
                }

                unit = row["Quantity unit"]
                if len(unit) != 0:
                    unit_term = _get_or_create_taxonomy_term("unit", unit)
                    harvest_quantity_data["relationships"]["units"] = {
                        "data": {
                            "type": "taxonomy_term--unit",
                            "id": unit_term["id"],
                        }
                    }

                # Create the quantity.
                harvest_quantity = client.resource.send('quantity', 'standard', harvest_quantity_data)

                # Add quantity to log.
                log["relationships"]["quantity"] = {
                    "data": [
                        {
                            "type": "quantity--standard",
                            "id": harvest_quantity["data"]["id"],
                            "meta": {
                                "target_revision_id": harvest_quantity["data"]["attributes"]["drupal_internal__revision_id"]
                            }
                        },
                    ]
                }

            new_log = client.log.send("harvest", log)

            link = "{hostname}/log/{id}".format(hostname=hostname,
                                                  id=new_log["data"]["attributes"]["drupal_internal__id"])
            print(
                "Imported log for asset: {name} - {link}".format(name=asset_name,
                                                                  link=link))