Actions/cofactr-cogs
6
Fork 0
mirror of https://github.com/AllSpiceIO/cofactr-cogs.git synced 2025-04-14 07:19:13 +00:00
Files Issues Releases Activity

Merge pull request from AllSpiceIO/jt/search-strategy

Browse Source 
feat: Change search strategy and add manufacturer to query
This commit is contained in:
Jonathan Tran 2024-05-28 14:31:35 -04:00 committed by GitHub
commit c8c33be354
No known key found for this signature in database
GPG Key ID: B5690EEEBB952194
7 changed files with 344 additions and 208 deletions

1
Dockerfile
View File

@ -1,6 +1,7 @@
FROM python:3.12-bookworm
COPY entrypoint.py /entrypoint.py
COPY cofactr_cogs /cofactr_cogs
RUN pip install requests

8
README.md
View File

@ -1,6 +1,8 @@
# Cofactr COGS
Generate cost of goods sold (COGS) using Cofactr.
Generate cost of goods sold (COGS) in AllSpice Actions using Cofactr.
This uses the Cofactr API. See the [Cofactr API docs](https://help.cofactr.com/en/articles/8868930-cofactr-component-cloud-api-documentation) for more information.
## Usage
@ -11,7 +13,11 @@ Add the following step to your actions:
uses: https://hub.allspice.io/Actions/cofactr-cogs@v1
with:
bom_file: bom.csv
bom_part_number_column: Part Number
bom_manufacturer_column: Manufacturer
bom_quantity_column: Quantity
quantities: "1,10,100,1000"
search_strategy: mpn_sku_mfr
client_id: YOUR_COFACTR_CLIENT_ID
api_key: ${{ secrets.COFACTR_API_KEY }}
output_file: cogs.csv

30
action.yml
View File

@ -12,13 +12,24 @@ inputs:
default: "1,10,100,1000"
bom_part_number_column:
description: >
The name of the part number column in the BOM file. Defaults to 'Part
Number'.
default: Part Number
The name of the part number column in the BOM file. You must provide
this.
default: ''
bom_manufacturer_column:
description: >
The name of the manufacturer column in the BOM file. Defaults to ''. If
you use a search strategy that uses manufacturer, you must provide this.
default: ''
bom_quantity_column:
description: >
The name of the quantity column in the BOM file. Defaults to 'Quantity'.
default: Quantity
The name of the quantity column in the BOM file. You must provide this.
default: ''
search_strategy:
description: >
The Cofactr search strategy. Can be: "mpn_sku_mfr" or "fuzzy" (uses mpn).
Defaults to "mpn_sku_mfr". See Cofactr API documentation for more
information on search strategies.
default: mpn_sku_mfr
output_file:
description: >
The path to the output file. Defaults to stdout, i.e. printing to the
@ -30,6 +41,9 @@ inputs:
client_id:
description: "Cofactr Client ID"
required: true
log_level:
description: Set log level for debugging
default: info
runs:
using: "docker"
image: "Dockerfile"
@ -38,10 +52,16 @@ runs:
- "${{ inputs.quantities }}"
- "--bom-part-number-column"
- ${{ inputs.bom_part_number_column }}
- "--bom-manufacturer-column"
- ${{ inputs.bom_manufacturer_column }}
- "--bom-quantity-column"
- ${{ inputs.bom_quantity_column }}
- "--search-strategy"
- ${{ inputs.search_strategy }}
- "--output-file"
- ${{ inputs.output_file }}
- "--log-level"
- ${{ inputs.log_level }}
- ${{ inputs.bom_file }}
env:
ALLSPICE_AUTH_TOKEN: ${{ github.token }}

0
cofactr_cogs/__init__.py Normal file
View File

111
cofactr_cogs/api.py Normal file
View File

@ -0,0 +1,111 @@
from dataclasses import dataclass
from enum import Enum
import os
import sys
import requests
class SearchStrategy(Enum):
MPN_SKU_MFR = "mpn_sku_mfr"
FUZZY = "fuzzy"
def to_query_value(self) -> str:
# Cofactr's default search strategy is designed for search results.
if self == SearchStrategy.FUZZY:
return "default"
return self.value
def query_needs_manufacturer(self) -> bool:
return self == SearchStrategy.MPN_SKU_MFR
@dataclass
class PartPrices:
cofactr_id: str
prices: dict[int, float]
def fetch_price_for_part(
part_number: str, manufacturer: str, search_strategy: SearchStrategy
) -> PartPrices | None:
"""
Get the price of a component per n units.
The return value of this function is a mapping of number of units to the
price in dollars per unit if you purchase that many units. For example::
{
1: 0.5,
10: 0.45,
500: 0.4,
100: 0.35,
}
In this case, the price per unit is 0.5, the price per unit if you buy 10
or more is 0.45, the price per unit if you buy 50 or more is 0.4, and so on.
Your breakpoints can be any positive integer.
The implementation of this function depends on the API you are using to get
pricing data. This is an example implementation that uses the cofactr API,
and will not work unless you have a cofactr API key. You will need to
replace this function with your own implementation if you use some other
API, such as Octopart or TrustedParts. You have access to the `requests`
python library to perform HTTP requests.
:param part_number: A part number by which to search for the component.
:returns: A mapping of price breakpoints to the price at that breakpoint.
"""
if part_number.startswith("NOTAPART"):
return None
api_key = os.environ.get("COFACTR_API_KEY")
client_id = os.environ.get("COFACTR_CLIENT_ID")
if api_key is None or client_id is None:
raise ValueError(
"Please set the COFACTR_API_KEY and COFACTR_CLIENT_ID environment variables"
)
query = part_number
if search_strategy.query_needs_manufacturer() and manufacturer:
query += f" {manufacturer}"
search_response = requests.get(
"https://graph.cofactr.com/products/",
headers={
"X-API-KEY": api_key,
"X-CLIENT-ID": client_id,
},
params={
"q": query,
"search_strategy": search_strategy.to_query_value(),
"schema": "product-offers-v0",
"external": "true",
"limit": "1",
},
)
if search_response.status_code != 200:
print(
f"Warning: Received status code {search_response.status_code} for {part_number} {manufacturer}",
file=sys.stderr,
)
return None
search_results = search_response.json()
try:
reference_prices = search_results.get("data", [])[0].get("reference_prices")
except IndexError:
print(
f"Warning: No results found for {part_number} {manufacturer}",
file=sys.stderr,
)
return None
prices = {int(price["quantity"]): float(price["price"]) for price in reference_prices}
return PartPrices(
cofactr_id=search_results["data"][0]["id"],
prices=prices,
)

198
cofactr_cogs/cli.py Normal file
View File

@ -0,0 +1,198 @@
# Compute the Cost of Goods Sold for a BOM.
#
# This script doesn't depend on py-allspice, but it requires a BOM CSV file to
# run. You can use https://github.com/AllSpiceIO/generate-bom to generate a BOM
# CSV.
from argparse import ArgumentParser
from contextlib import ExitStack
import csv
import sys
from cofactr_cogs.api import fetch_price_for_part, PartPrices, SearchStrategy
def main() -> None:
parser = ArgumentParser()
parser.add_argument(
"bom_file",
help="The path to the BOM file.",
)
parser.add_argument(
"--quantities",
help=(
"A comma-separated list of quantities of PCBs to compute the COGS "
+ "for. Defaults to '%(default)s'."
),
default="1,10,100,1000",
)
parser.add_argument(
"--bom-part-number-column",
help="The name of the part number column in the BOM file. You must provide this.",
default="",
)
parser.add_argument(
"--bom-manufacturer-column",
help="The name of the manufacturer column in the BOM file. Defaults to '%(default)s'. If "
+ "you use a search strategy that uses manufacturer, you must provide this.",
default="",
)
parser.add_argument(
"--bom-quantity-column",
help="The name of the quantity column in the BOM file. You must provide this.",
default="",
)
parser.add_argument(
"--search-strategy",
help="The Cofactr search strategy. Can be: mpn_sku_mfr or fuzzy (uses mpn). "
+ "Defaults to '%(default)s'. The API also supports mpn_exact and mpn_exact_mfr, "
+ "but they are not recommended.",
default="mpn_sku_mfr",
)
parser.add_argument(
"--output-file",
help="The path to the output file. Defaults to stdout, i.e. printing to the console.",
)
parser.add_argument(
"--log-level",
help="The log level to use. Defaults to '%(default)s'.",
default="info",
)
args = parser.parse_args()
quantities = [int(quantity) for quantity in args.quantities.split(",")]
part_number_column = args.bom_part_number_column
if not part_number_column:
raise ValueError(
"BOM part number column needs to be specified. Please set bom_part_number_column."
)
manufacturer_column = args.bom_manufacturer_column
quantity_column = args.bom_quantity_column
if not quantity_column:
raise ValueError(
"BOM quantity column needs to be specified. Please set bom_quantity_column."
)
search_strategy = SearchStrategy(args.search_strategy)
with open(args.bom_file, "r") as bom_file:
bom_csv = csv.DictReader(bom_file)
parts = [
part for part in bom_csv if part[part_number_column] and part[part_number_column] != ""
]
print(f"Computing COGS for {len(parts)} parts", file=sys.stderr)
print(f"Fetching prices for {len(parts)} parts", file=sys.stderr)
prices_for_parts = {}
use_mfr = bool(manufacturer_column)
if not use_mfr and search_strategy.query_needs_manufacturer():
raise ValueError(
"Search strategy requires manufacturer, but no BOM manufacturer column was provided. Please set bom_manufacturer_column."
)
if args.log_level.lower() == "debug":
print(f"Using part number column: {part_number_column!r}", file=sys.stderr)
print(f"Using manufacturer column: {manufacturer_column!r}", file=sys.stderr)
print(f"Using use_mfr: {use_mfr!r}", file=sys.stderr)
print(f"Using quantity column: {quantity_column!r}", file=sys.stderr)
print(f"Using search strategy: {search_strategy!r}", file=sys.stderr)
for part in parts:
part_number = part[part_number_column]
manufacturer = part[manufacturer_column] if use_mfr else ""
part_prices = fetch_price_for_part(part_number, manufacturer, search_strategy)
if part_prices is not None and len(part_prices.prices) > 0:
prices_for_parts[(part_number, manufacturer)] = part_prices
print(f"Found prices for {len(prices_for_parts)} parts", file=sys.stderr)
if len(prices_for_parts) == 0:
print("No prices found for any parts", file=sys.stderr)
sys.exit(1)
# The number of columns that the output should have.
expected_columns = (4 if use_mfr else 3) + 2 * len(quantities)
headers = ["Part Number"]
if use_mfr:
headers.append("Manufacturer")
headers.append("Cofactr ID")
headers.append("Quantity")
for quantity in quantities:
headers.append(f"Per Unit at {quantity}")
headers.append(f"Total at {quantity}")
assert len(headers) == expected_columns
rows = []
totals: dict[int, float] = {quantity: 0 for quantity in quantities}
for part in parts:
part_number = part[part_number_column]
manufacturer = part[manufacturer_column] if use_mfr else ""
part_quantity = int(part[quantity_column])
part_prices = prices_for_parts.get((part_number, manufacturer))
cofactr_id = part_prices.cofactr_id if part_prices else None
current_row = [part_number]
if use_mfr:
current_row.append(manufacturer)
current_row.append(cofactr_id)
current_row.append(part_quantity)
for quantity in quantities:
breakpoints = price_breakpoints(part_prices, quantity)
if part_prices and breakpoints:
largest_breakpoint_less_than_qty = max(breakpoints)
price_at_breakpoint = part_prices.prices[largest_breakpoint_less_than_qty]
current_row.append(price_at_breakpoint)
total_for_part_at_quantity = price_at_breakpoint * part_quantity
current_row.append(total_for_part_at_quantity)
totals[quantity] += total_for_part_at_quantity
else:
current_row.append(None)
current_row.append(None)
assert len(current_row) == expected_columns
rows.append(current_row)
with ExitStack() as stack:
if args.output_file:
file = stack.enter_context(open(args.output_file, "w"))
writer = csv.writer(file)
else:
writer = csv.writer(sys.stdout)
writer.writerow(headers)
writer.writerows(rows)
totals_row = ["Totals", None, None]
if use_mfr:
totals_row.append(None)
for quantity in quantities:
totals_row.append(None)
totals_row.append(str(totals[quantity]))
assert len(totals_row) == expected_columns
writer.writerow(totals_row)
print("Computed COGS", file=sys.stderr)
def price_breakpoints(part_prices: PartPrices | None, quantity: int) -> list[int] | None:
if part_prices is None:
return None
breakpoints = [breakpoint for breakpoint in part_prices.prices.keys() if breakpoint <= quantity]
return breakpoints if len(breakpoints) > 0 else None
if __name__ == "__main__":
main()

204
entrypoint.py
View File

@ -1,205 +1,5 @@
#! /usr/bin/env python3
# Compute the Cost of Goods Sold for a BOM.
#
# This script doesn't depend on py-allspice, but it requires a BOM CSV file to
# run. You can use https://github.com/AllSpiceIO/generate-bom to generate a BOM
# CSV.
from argparse import ArgumentParser
from contextlib import ExitStack
import csv
import os
import sys
import requests
def fetch_price_for_part(part_number: str) -> dict[int, float]:
"""
Get the price of a component per n units.
The return value of this function is a mapping of number of units to the
price in dollars per unit if you purchase that many units. For example::
{
1: 0.5,
10: 0.45,
500: 0.4,
100: 0.35,
}
In this case, the price per unit is 0.5, the price per unit if you buy 10
or more is 0.45, the price per unit if you buy 50 or more is 0.4, and so on.
Your breakpoints can be any positive integer.
The implementation of this function depends on the API you are using to get
pricing data. This is an example implementation that uses the cofactr API,
and will not work unless you have a cofactr API key. You will need to
replace this function with your own implementation if you use some other
API, such as Octopart or TrustedParts. You have access to the `requests`
python library to perform HTTP requests.
:param part_number: A part number by which to search for the component.
:returns: A mapping of price breakpoints to the price at that breakpoint.
"""
if part_number.startswith("NOTAPART"):
return {}
api_key = os.environ.get("COFACTR_API_KEY")
client_id = os.environ.get("COFACTR_CLIENT_ID")
if api_key is None or client_id is None:
raise ValueError(
"Please set the COFACTR_API_KEY and COFACTR_CLIENT_ID environment variables"
)
search_response = requests.get(
"https://graph.cofactr.com/products/",
headers={
"X-API-KEY": api_key,
"X-CLIENT-ID": client_id,
},
params={
"q": part_number,
"schema": "product-offers-v0",
"external": "true",
"limit": "1",
},
)
if search_response.status_code != 200:
print(
f"Warning: Received status code {search_response.status_code} for {part_number}",
file=sys.stderr,
)
return {}
search_results = search_response.json()
try:
reference_prices = search_results.get("data", [])[0].get("reference_prices")
except IndexError:
print(
f"Warning: No results found for {part_number}",
file=sys.stderr,
)
return {}
prices = {int(price["quantity"]): float(price["price"]) for price in reference_prices}
return prices
from cofactr_cogs.cli import main
if __name__ == "__main__":
parser = ArgumentParser()
parser.add_argument(
"bom_file",
help="The path to the BOM file.",
)
parser.add_argument(
"--quantities",
help=(
"A comma-separated list of quantities of PCBs to compute the COGS "
+ "for. Defaults to '%(default)s'."
),
default="1,10,100,1000",
)
parser.add_argument(
"--bom-part-number-column",
help="The name of the part number column in the BOM file. Defaults to '%(default)s'.",
default="Part Number",
)
parser.add_argument(
"--bom-quantity-column",
help="The name of the quantity column in the BOM file. Defaults to '%(default)s'.",
default="Quantity",
)
parser.add_argument(
"--output-file",
help="The path to the output file. Defaults to stdout, i.e. printing to the console.",
)
args = parser.parse_args()
quantities = [int(quantity) for quantity in args.quantities.split(",")]
part_number_column = args.bom_part_number_column
quantity_column = args.bom_quantity_column
with open(args.bom_file, "r") as bom_file:
bom_csv = csv.DictReader(bom_file)
parts = [
part for part in bom_csv if part[part_number_column] and part[part_number_column] != ""
]
print(f"Computing COGS for {len(parts)} parts", file=sys.stderr)
print(f"Fetching prices for {len(parts)} parts", file=sys.stderr)
prices_for_parts = {}
for part in parts:
part_number = part[part_number_column]
prices = fetch_price_for_part(part_number)
if prices and len(prices) > 0:
prices_for_parts[part_number] = prices
print(f"Found prices for {len(prices_for_parts)} parts", file=sys.stderr)
if len(prices_for_parts) == 0:
print("No prices found for any parts", file=sys.stderr)
sys.exit(1)
headers = [
"Part Number",
"Quantity",
]
for quantity in quantities:
headers.append(f"Per Unit at {quantity}")
headers.append(f"Total at {quantity}")
rows = []
totals: dict[int, float] = {quantity: 0 for quantity in quantities}
for part in parts:
part_number = part[part_number_column]
part_quantity = int(part[quantity_column])
current_row = [part_number, part_quantity]
for quantity in quantities:
try:
prices = prices_for_parts[part_number]
largest_breakpoint_less_than_qty = max(
[breakpoint for breakpoint in prices.keys() if breakpoint <= quantity]
)
price_at_breakpoint = prices[largest_breakpoint_less_than_qty]
current_row.append(price_at_breakpoint)
total_for_part_at_quantity = price_at_breakpoint * part_quantity
current_row.append(total_for_part_at_quantity)
totals[quantity] += total_for_part_at_quantity
except (ValueError, KeyError):
current_row.append(None)
current_row.append(None)
rows.append(current_row)
with ExitStack() as stack:
if args.output_file:
file = stack.enter_context(open(args.output_file, "w"))
writer = csv.writer(file)
else:
writer = csv.writer(sys.stdout)
writer.writerow(headers)
writer.writerows(rows)
totals_row = ["Totals", None]
for quantity in quantities:
totals_row.append(None)
totals_row.append(str(totals[quantity]))
writer.writerow(totals_row)
print("Computed COGS", file=sys.stderr)
main()