Skip to content

Instantly share code, notes, and snippets.

@paul121
Last active Jul 26, 2021
Embed
What would you like to do?
farmOS.py 2.x Import example

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 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 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"]))
name crop variety date
Big Corn planing corn big 5/5/2021
Pumpkins squash pumpkin 8/5/2021
Green beans #1 beans green beans 6/5/2021
Green beans #2 (future) beans green beans 10/5/2021
@ludwa6
Copy link

ludwa6 commented Jul 23, 2021

Hey, @paul121 : this is great, am keen to try it -only i'm a bit hesitant to do something that might mess up my Python environment, created with Spyder via the Anaconda distro. Spyder warns users NOT to use pip, but rather to stick with conda for installing packages. Fortunately i was able to install farmOS.py via conda-forge channel (thanks for that!), but now i just want to be sure: can i safely follow your instruction to set up a virtual environment in this conda-managed environment of mine?

NB: I did browse the docs you pointed to, including section on Virtual Environments but -tho it mentions other package managers besides pip, which it recommends- there's nothing in there about conda, so...

Do you have any insight about this?

@paul121
Copy link
Author

paul121 commented Jul 26, 2021

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/

Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment