partsdb.py (15440B)
1 #!/usr/bin/env python3 2 # -*- coding: utf-8 -*- 3 # vim:fenc=utf-8 4 5 import argparse 6 import os 7 import subprocess 8 import sys 9 import urllib.request 10 import yaml 11 12 from partsdb import database as pdb 13 from partsdb import exports 14 from partsdb import helpers 15 from partsdb import octopart as oc 16 17 __version__ = "v2.0.dev0" 18 octo = oc.OctopartClient(os.getenv("OCTOPART_TOKEN", None)) 19 db = pdb.PartsDB( 20 os.getenv("PARTSDB_FILE", f"{os.getenv('HOME')}/.local/share/partsdb/parts.db") 21 ) 22 23 24 def add_part_from_octo(mpn, quantity, category, storage, part_type): 25 result = octo.get_part(mpn)["data"]["search"]["results"] 26 if result is None: 27 print(f"Can't find results for {sys.argv[1]} on Octopart") 28 sys.exit(0) 29 30 # list results from Octopart and pick one 31 for i, r in enumerate(result): 32 print("-" * 79) 33 print(f"{i}\t{r['part']['manufacturer']['name']}" f"\t{r['part']['mpn']}") 34 print(f"\t{r['part']['short_description']}") 35 print("Sold by:") 36 for s in r["part"]["sellers"]: 37 print(s["company"]["name"]) 38 print(f"\t{r['part']['octopart_url']}") 39 40 pick = int(input("Which one seems better ? ")) 41 p = result[pick]["part"] 42 43 if quantity is None: 44 quantity = int(input("How many of them ? ")) 45 46 # if this exists we increment stock 47 spart = db.get_part_by_mpn(p["mpn"]) 48 if spart: 49 db.update_part_qty(spart["id"], quantity) 50 db.new_part_history_event(spart["id"], quantity, "new buy") 51 return 52 53 # only ask for categories if we do not have one already 54 # list categories to choose from 55 if category is None: 56 for c in db.get_categories(): 57 print(f"{c['id']}) {c['name']}") 58 category = int(input("In which category do you want it in ? ")) 59 60 # only ask for storage if we do not have one already 61 # list storages to choose from 62 if storage is None: 63 for s in db.get_storages(): 64 print(f"{s['id']}) {s['name']}") 65 storage = int(input("Where will you store it ? ")) 66 67 # only ask for part type if we do not have one already 68 if part_type == "none": 69 smd = input("Is this an SMD part (y/n, default yes) ? ") 70 if smd == "n" or smd == "N": 71 part_type = "th" 72 else: 73 part_type = "smd" 74 75 footprint = None 76 datasheet = None 77 image = None 78 specs = "" 79 80 if "specs" in p: 81 for s in p["specs"]: 82 specs += f"{s['attribute']['name']}: {s['display_value']}\n" 83 if s["attribute"]["shortname"] == "case_package": 84 footprint = s["display_value"] 85 if not specs: 86 specs = None 87 88 if "best_datasheet" in p and p["best_datasheet"]: 89 if p["best_datasheet"]["mime_type"] == "application/pdf": 90 datasheet = p["best_datasheet"]["url"] 91 elif "document_collections" in p: 92 for d in p["document_collections"][0]["documents"]: 93 if d["mime_type"] == "application/pdf" and d["name"] == "Datasheet": 94 datasheet = d["url"] 95 headers = {} 96 headers[ 97 "User-Agent" 98 ] = "Mozilla/5.0 (X11; Ubuntu; Linux i686; rv:48.0) Gecko/20100101 Firefox/48.0" 99 if datasheet is not None: 100 req = urllib.request.Request(datasheet, headers=headers) 101 datasheet = urllib.request.urlopen(req).read() 102 103 if "best_image" in p and p["best_image"]: 104 image = p["best_image"]["url"] 105 if image is not None: 106 req = urllib.request.Request(image, headers=headers) 107 image = urllib.request.urlopen(req).read() 108 109 part = [ 110 p["mpn"], 111 p["manufacturer"]["name"], 112 p["short_description"], 113 specs, 114 footprint, 115 category, 116 storage, 117 quantity, 118 datasheet, 119 image, 120 part_type, 121 ] 122 new_id = db.new_part(part) 123 db.new_part_history_event(new_id, quantity, "first purchase") 124 125 if new_id: 126 print(f"New part created with ID: {new_id} and PN: {new_id:04X}") 127 128 129 def add_part_from_template(): 130 new_part_fields = { 131 "mpn": None, 132 "manufacturer": "Unknown", 133 "description": None, 134 "specs": None, 135 "footprint": None, 136 "part_type": None, 137 "cat": None, 138 "storage": None, 139 "quantity": None, 140 } 141 part_tmp_file = f"/tmp/partsdb_new_part.yaml" 142 with open(part_tmp_file, mode="w+") as fp: 143 yaml.dump(new_part_fields, fp) 144 145 # call vim to edit, load the resulting yaml and update db 146 with open(part_tmp_file, mode="r") as fp: 147 editor = os.environ.get("EDITOR", "vim") 148 subprocess.call([editor, fp.name]) 149 try: 150 new_part = yaml.safe_load(fp) 151 new_id = db.new_part_from_template(new_part) 152 if new_id: 153 db.new_part_history_event( 154 new_id, new_part["quantity"], "first purchase" 155 ) 156 print(f"New part created with ID: {new_id} and PN: {new_id:04X}") 157 except yaml.YAMLError as exc: 158 print(exc) 159 160 # clean up 161 os.remove(part_tmp_file) 162 163 164 def list_parts(category, short): 165 if category == "all": 166 parts = db.list_parts() 167 else: 168 parts = db.list_parts_by_category(category) 169 170 if not parts: 171 print("There are no parts in this category") 172 return 173 174 helpers.print_parts_list(parts, short) 175 176 177 def search_part(search_term, output): 178 parts = db.search_parts(search_term) 179 180 if not parts: 181 print("No parts found") 182 return 183 184 helpers.print_parts_list(parts, output) 185 186 187 def get_part(part_id, output): 188 part = db.get_part(part_id) 189 history = db.get_part_history(part_id) 190 191 if not part: 192 print("No part with that ID or PN") 193 return 194 195 helpers.print_part(part, history, output) 196 197 198 def open_image(part_id): 199 image = db.get_image(part_id) 200 if not image["image"]: 201 print(f"There's no image for this part ID ({part_id})") 202 return 203 helpers.open_file(image["image"], ".jpg") 204 205 206 def open_datasheet(part_id): 207 datasheet = db.get_datasheet(part_id) 208 if not datasheet["datasheet"]: 209 print(f"There's no datasheet for this part ID ({part_id})") 210 return 211 helpers.open_file(datasheet["datasheet"], ".pdf") 212 213 214 def delete_part(part_id): 215 db.delete_part(part_id) 216 217 218 def update_part(part_id): 219 # load part into a tmp yaml 220 part = dict(db.get_part(part_id)) 221 222 # ignore some fields. 223 # Binaries We use other functions for this. 224 # also the id fields. We have the text representation of those 225 # insert date should not be modifiable. In fact is ingored on the db update query 226 # and update_date is auto generated 227 ignore_list = [ 228 "datasheet", 229 "image", 230 "storage_id", 231 "category_id", 232 "insert_date", 233 "update_date", 234 ] 235 for i in ignore_list: 236 del part[i] 237 238 part_tmp_file = f"/tmp/partsdb_edit_part{part_id}.yaml" 239 with open(part_tmp_file, mode="w+") as fp: 240 yaml.dump(part, fp) 241 242 # call vim to edit, load the resulting yaml and update db 243 with open(part_tmp_file, mode="r") as fp: 244 editor = os.environ.get("EDITOR", "vim") 245 subprocess.call([editor, fp.name]) 246 try: 247 edited_part = yaml.safe_load(fp) 248 db.update_part(edited_part) 249 except yaml.YAMLError as exc: 250 print(exc) 251 252 # clean up 253 os.remove(part_tmp_file) 254 255 256 def update_datasheet(part_id, datasheet_file): 257 with open(datasheet_file, mode="rb") as f: 258 db.update_datasheet(part_id, f.read()) 259 260 261 def update_image(part_id, image_file): 262 with open(image_file, mode="rb") as f: 263 db.update_image(part_id, f.read()) 264 265 266 def adjust_stock(part_id, stock_mod, comment): 267 db.new_part_history_event(part_id, stock_mod, comment) 268 db.update_part_qty(part_id, stock_mod) 269 270 271 def list_categories(): 272 categories = db.get_categories() 273 print("ID\tName") 274 print("-" * 40) 275 for c in categories: 276 print(f"{c['id']}\t{c['name']}") 277 278 279 def list_storages(): 280 storages = db.get_storages() 281 print("ID\tName") 282 print("-" * 40) 283 for c in storages: 284 print(f"{c['id']}\t{c['name']}") 285 286 287 def main(): 288 ap = argparse.ArgumentParser() 289 # global options here 290 ap.add_argument( 291 "--version", "-v", action="version", version="%(prog)s " + __version__ 292 ) 293 # commands 294 asp = ap.add_subparsers(dest="command") 295 ## part 296 ap_part = asp.add_parser("part", help="Interact with parts.") 297 part_sp = ap_part.add_subparsers(dest="subcommand") 298 ### part -> add 299 ap_part_add = part_sp.add_parser("add", help="Add new part from Octopart") 300 ap_part_add.add_argument( 301 "-T", 302 "--from-template", 303 action="store_true", 304 dest="from_template", 305 help="Add part from YAML template", 306 ) 307 ap_part_add.add_argument("-m", "--mpn", dest="mpn", help="Manufacturer part number") 308 ap_part_add.add_argument( 309 "-q", dest="quantity", help="Quantity of new items", type=int 310 ) 311 ap_part_add.add_argument( 312 "-c", dest="category", help="Which categoryId it belongs to", type=int 313 ) 314 ap_part_add.add_argument( 315 "-s", dest="storage", help="Which storageId it belongs to", type=int 316 ) 317 ap_part_add.add_argument( 318 "-t", 319 dest="type", 320 choices=["smd", "th"], 321 default="none", 322 help="Trhough-hole of smd ?", 323 ) 324 ### part -> del (delete) 325 ap_part_del = part_sp.add_parser("del", help="Delete a part") 326 ap_part_del.add_argument("part_id", help="Part Id") 327 ### part -> get 328 ap_part_get = part_sp.add_parser("get", help="Get all details of a part") 329 ap_part_get.add_argument("part_id", help="Part Id") 330 ap_part_get.add_argument( 331 "-d", dest="datasheet", action="store_true", help="Open datasheet if available." 332 ) 333 ap_part_get.add_argument( 334 "-i", dest="image", action="store_true", help="Open image if available." 335 ) 336 ap_part_get.add_argument( 337 "-o", 338 dest="output", 339 choices=["full", "json"], 340 default="full", 341 help="Short output", 342 ) 343 ### part -> list 344 ap_part_list = part_sp.add_parser("ls", help="List parts") 345 ap_part_list.add_argument( 346 "-c", "--category", default="all", help="Category Name or ID" 347 ) 348 ap_part_list.add_argument( 349 "-o", 350 dest="output", 351 choices=["full", "short", "json"], 352 default="full", 353 help="Short output", 354 ) 355 ### part -> search 356 ap_part_search = part_sp.add_parser("search", help="Search parts") 357 ap_part_search.add_argument("search_term", help="Term to search for") 358 ap_part_search.add_argument( 359 "-o", 360 dest="output", 361 choices=["full", "short", "json"], 362 default="full", 363 help="Short output", 364 ) 365 ### part -> stock 366 ap_part_stock = part_sp.add_parser("stock", help="Modifies a part stock") 367 ap_part_stock.add_argument("part_id", help="Part Id") 368 ap_part_stock.add_argument( 369 "stock_mod", help="Stock modifier (+ or -) int", type=int 370 ) 371 ap_part_stock.add_argument("comment", help="Reason for the stock mod") 372 ### part -> update 373 ap_part_update = part_sp.add_parser( 374 "update", 375 help="update a part. Will launch $EDITOR with a YAML dump of the part", 376 ) 377 ap_part_update.add_argument("-d", dest="datasheet", help="Datasheet file to update") 378 ap_part_update.add_argument("-i", dest="image", help="Image file to update") 379 ap_part_update.add_argument("part_id", help="Part Id") 380 ## cat (categories) 381 ap_cat = asp.add_parser("cat", help="Interact with categories.") 382 cat_sp = ap_cat.add_subparsers(dest="subcommand") 383 ### cat -> list 384 cat_sp.add_parser("ls", help="List categories") 385 ### cat -> new 386 ap_cat_new = cat_sp.add_parser("new", help="Add a new category") 387 ap_cat_new.add_argument("cat_name", help="New category name") 388 ## sto (storages) 389 ap_sto = asp.add_parser("sto", help="Interact with storages.") 390 sto_sp = ap_sto.add_subparsers(dest="subcommand") 391 ### sto -> del 392 ap_sto_del = sto_sp.add_parser( 393 "del", help="Delete a storage. All affected parts will be put into Unsorted" 394 ) 395 ap_sto_del.add_argument("sto_id", help="Storage ID to delete name") 396 ### sto -> list 397 sto_sp.add_parser("ls", help="List storages") 398 ### sto -> new 399 ap_sto_new = sto_sp.add_parser("new", help="Add a new storage") 400 ap_sto_new.add_argument("sto_name", help="New storage name") 401 ## export 402 ap_export = asp.add_parser("export", help="Exports DB to HTML") 403 ap_export.add_argument( 404 "-x", 405 dest="to_excel", 406 action="store_true", 407 help="Export to Excel. Only parts and storages (default is to HTML)", 408 ) 409 ap_export.add_argument( 410 "-s", 411 dest="sto_id", 412 type=int, 413 default=0, 414 help="Export only parts on a given storage ID (default to all)", 415 ) 416 ap_export.add_argument("dest_folder", default=".", help="Destination folder") 417 418 args = ap.parse_args() 419 if not args.command: 420 ap.print_help() 421 sys.exit(0) 422 423 version_major = int(__version__.split(".")[0][1:]) 424 if not db.validate_major_version(version_major): 425 print("Database does not match software version") 426 sys.exit(1) 427 428 if "part_id" in vars(args).keys(): 429 args.part_id = int(args.part_id, 0) 430 431 if args.command == "part": 432 if args.subcommand == "add": 433 if args.from_template: 434 add_part_from_template() 435 else: 436 if not args.mpn: 437 print("MPN is required if not adding from template") 438 sys.exit(1) 439 add_part_from_octo( 440 args.mpn, args.quantity, args.category, args.storage, args.type 441 ) 442 elif args.subcommand == "del": 443 delete_part(args.part_id) 444 elif args.subcommand == "get": 445 get_part(args.part_id, args.output) 446 if args.datasheet and args.output != "json": 447 open_datasheet(args.part_id) 448 if args.image and args.output != "json": 449 open_image(args.part_id) 450 elif args.subcommand == "ls": 451 list_parts(args.category, args.output) 452 elif args.subcommand == "search": 453 search_part(args.search_term, args.output) 454 elif args.subcommand == "stock": 455 adjust_stock(args.part_id, args.stock_mod, args.comment) 456 elif args.subcommand == "update": 457 if args.datasheet: 458 update_datasheet(args.part_id, args.datasheet) 459 elif args.image: 460 update_image(args.part_id, args.image) 461 else: 462 update_part(args.part_id) 463 464 if args.command == "cat": 465 if args.subcommand == "ls": 466 list_categories() 467 elif args.subcommand == "new": 468 print("Not implemented") 469 470 if args.command == "sto": 471 if args.subcommand == "ls": 472 list_storages() 473 elif args.subcommand == "del": 474 print("Not implemented") 475 elif args.subcommand == "new": 476 print("Not implemented") 477 478 if args.command == "export": 479 if args.to_excel: 480 exports.export_to_excel(args.dest_folder, args.sto_id) 481 else: 482 exports.export_to_html(args.dest_folder) 483 484 db.close() 485 486 487 if __name__ == "__main__": 488 main()