farmOS.py 2.x Import example

README.md

Instructions

Install farmOS.py using pip. Rather than installing system wide, it is recommended to install packages into a virtual environment. For more information see Installing Python Packages.

# Dedicate a directory for farmOS scripts.
cd farmos_scripts

# Create a virtual environment in a directory called "venv".
python3 -m venv venv
source venv/bin/activate

# Install farmOS.py
python3 -m pip install farmOS~=1.0.0b

# Copy the script and CSV file into this directory.
# Edit the script to specify the correct hostname.
wget "https://gist.githubusercontent.com/paul121/7ec9685b6ca216b3b0173d8a16f1de04/raw/0d8ece3443b704d290e942f092ea1ce6e7265eaa/import_plantings.py"

# Run the script.
python3 import_plantings.py -f plantings.csv
Enter username: 
Enter password: 
INFO:root:Imported planting asset: Big Corn planing - http://localhost/asset/24
INFO:root:Imported planting asset: Pumpkins - http://localhost/asset/25
INFO:root:Imported planting asset: Green beans #1 - http://localhost/asset/26
INFO:root:Imported planting asset: Green beans #2 (future) - http://localhost/asset/27

import_plantings.py

import os
import csv
import argparse
import logging
from datetime import datetime
from time import time

from farmOS import farmOS

# Uncomment this line to allow HTTP requests.
# os.environ['OAUTHLIB_INSECURE_TRANSPORT'] = '1'

# Allow INFO level logs.
logging.getLogger().setLevel(logging.INFO)

# Configure a file command line argument.
parser = argparse.ArgumentParser()                                               
parser.add_argument("--file", "-f", type=str, required=True)
args = parser.parse_args()

# The path to the CSV file.
#file_path = "plantings.csv"
file_path = args.file

# farmOS server hostname. EDIT THIS!
hostname = "http://localhost"

# Create and authorize the client.
client = farmOS(hostname=hostname, scope="farm_manager", version=2)
client.authorize()

# Alternatively hard code username and password.
#username = "admin"
#password = "admin"
# client.authorize(username, password)

def _get_plant_asset_plant_type(crop_name, plant_type_name):
    """
    Helper function to load an existing plant_type term or create a new one.
    
    :param crop_name: Name of the crop_family term.
    :param plant_type_name: Name of the plant_type term.
    :return: The ID of the plant_type term.
    """
    
    # Request plant_types of the specified crop_family to see if a plant_type term already exists.
    crop_filter = client.filter('crop_family.entity.name', crop_name)
    name_filter = client.filter('name', plant_type_name)
    plant_types = list(client.term.iterate('plant_type', {**crop_filter, **name_filter}))
    
    # Return ID of the first one found.
    if len(plant_types) > 0:
        plant_type_id = plant_types[0]["id"]
        
    # Else create a new plant_type term.
    else:
        
        # Request all crop_family terms to see if the crop_family exists.
        crop_filter = client.filter('name', crop_name)
        crop_families = list(client.term.iterate('crop_family', crop_filter))
        
        # Return ID of the first one found.
        if len(crop_families) > 0:
            family_id = crop_families[0]["id"]
            
        # Else create a new crop_family term.
        else:
            family = {"attributes": {"name": crop_name}}
            new_family = client.term.send('crop_family', family)
            family_id = new_family["data"]["id"]

        # Create a new plant_type with the crop_family term relationship.
        plant_type = {
            "attributes": {
                "name": plant_type_name,
            },
            "relationships": {
                'crop_family': {
                    "data": [
                        {
                            "type": "taxonomy_term--crop_family",
                            "id": family_id
                        }
                    ]
                },
            }
        }
        new_plant_type = client.term.send('plant_type', plant_type)
        plant_type_id = new_plant_type["data"]["id"]

    return plant_type_id

# 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:
        
        # Load the crop and variety columns.
        crop_name = row['crop']
        variety_name = row['variety']
        
        # Use the helper function to get the plant_type term ID.
        plant_type_id = _get_plant_asset_plant_type(crop_name, variety_name)

        # Create a new plant asset.
        name = row['name']
        plant_asset = {
            "attributes": {
                "type": "plant",
                "name": name,
            },
            "relationships": {
                "plant_type": {
                    "data": [
                        {
                            "type": "taxonomy_term--plant_type",
                            "id": plant_type_id
                        }
                    ]
                },
            }
        }
        new_plant_asset = client.asset.send('plant', plant_asset)

        # Create a seeding log at the specified date.
        # Convert date string to timestamp.
        now = time()
        timestamp = datetime.strptime(row['date'], '%m/%d/%Y').timestamp()
        
        # The status is done if the timestamp is in the past, pending if in the future.
        status = "done" if timestamp <= now else "pending"

        # Create the seeding log referencing the plant asset.
        seeding_log = {
            "attributes": {
                "type": "seeding",
                "status": status
            },
            "relationships": {
                "asset": {
                    "data": [
                        {
                            "type": "asset--plant",
                            "id": new_plant_asset["data"]["id"],
                        }
                    ]
                }
            }
        }
        new_seeding_log = client.log.send("seeding", seeding_log)
        
        link = "{hostname}/asset/{id}".format(hostname=hostname, id=new_plant_asset["data"]["attributes"]["drupal_internal__id"])
        logging.info("Imported planting asset: {name} - {link}".format(name=new_plant_asset["data"]["attributes"]["name"], link=link))
        

import_plantings_1x.py

import os
import csv
import argparse
import logging
from datetime import datetime
from time import time

from farmOS import farmOS

# Uncomment this line to allow HTTP requests.
# os.environ['OAUTHLIB_INSECURE_TRANSPORT'] = '1'

# Allow INFO level logs.
logging.getLogger().setLevel(logging.INFO)

# Configure a file command line argument.
parser = argparse.ArgumentParser()                                               
parser.add_argument("--file", "-f", type=str, required=True)
args = parser.parse_args()

# The path to the CSV file.
#file_path = "plantings.csv"
file_path = args.file

# farmOS server hostname. EDIT THIS!
hostname = "http://localhost"

# Vocabulary IDs. EDIT THIS!
FARM_CROP_VOCABULARY=7
FARM_CROP_FAMILIES_VOCABULARY=6

# Create and authorize the client.
client = farmOS(hostname=hostname, scope="user_access", version=1)
token = client.authorize()
logging.info(token)

# Alternatively hard code username and password.
#username = "admin"
#password = "admin"
# client.authorize(username, password)

def _get_plant_asset_plant_type(crop_name, plant_type_name):
    """
    Helper function to load an existing plant_type term or create a new one.
    
    :param crop_name: Name of the crop_family term.
    :param plant_type_name: Name of the plant_type term.
    :return: The ID of the plant_type term.
    """
    
    # Request plant_types of the specified crop_family to see if a plant_type term already exists.
    plant_types = client.term.get({"name": plant_type_name, "vocabulary": FARM_CROP_VOCABULARY})["list"]

    # Return ID of the first one found.
    if len(plant_types) > 0:
        plant_type_id = plant_types[0]["tid"]
        
    # Else create a new plant_type term.
    else:
        
        # Request all crop_family terms to see if the crop_family exists.
        crop_families = client.term.get({"name": crop_name, "vocabulary": FARM_CROP_FAMILIES_VOCABULARY})["list"]
        
        # Return ID of the first one found.
        if len(crop_families) > 0:
            family_id = crop_families[0]["tid"]
            
        # Else create a new crop_family term.
        else:
            family = {
                "name": crop_name,
                "vocabulary": FARM_CROP_FAMILIES_VOCABULARY,
            }
            new_family = client.term.send(family)
            family_id = new_family["id"]

        # Create a new plant_type with the crop_family term relationship.
        plant_type = {
            "name": plant_type_name,
            "vocabulary": FARM_CROP_VOCABULARY,
            'crop_family': family_id,
        }
        new_plant_type = client.term.send(plant_type)
        plant_type_id = new_plant_type["id"]

    return plant_type_id

# 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:
        
        # Load the crop and variety columns.
        crop_name = row['crop']
        variety_name = row['variety']
        
        # Use the helper function to get the plant_type term ID.
        plant_type_id = _get_plant_asset_plant_type(crop_name, variety_name)

        # Create a new plant asset.
        name = row['name']
        plant_asset = {
            "type": "planting",
            "name": name,
            "crop": [
                plant_type_id
            ],
        }
        new_plant_asset = client.asset.send(plant_asset)

        # Create a seeding log at the specified date.
        # Convert date string to timestamp.
        now = time()
        timestamp = datetime.strptime(row['date'], '%m/%d/%Y').timestamp()
        
        # The status is done if the timestamp is in the past, pending if in the future.
        status = timestamp <= now

        # Create the seeding log referencing the plant asset.
        seeding_log = {
            "type": "farm_seeding",
            "done": status,
            "asset": [
                new_plant_asset["id"],
            ]
        }
        new_seeding_log = client.log.send(seeding_log)
        logging.info("Created asset: {id}".format(id=new_plant_asset["id"]))

plantings.csv

Hey @ludwa6, you should go ahead and install farmOS.py with conda rather than use pip + virtual environments. It does sound like a virtual environment could mess things up. I haven't used those myself so I can only stick to the Python recommendations I shared above. I know you already know, but to share for others, here are some instructions for installing via conda: https://github.com/conda-forge/farmos-feedstock/