openspoolman/spoolman_client.py at main · Xento/openspoolman · GitHub

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
import requests
from config import SPOOLMAN_API_URL, SPOOL_SORTING
import json
from logger import append_to_rotating_file, log

SPOOLMAN_LOG_FILE = "/home/app/logs/spoolman.log"


def _log_spoolman_change(action, spool_id=None, payload=None, status=None):
  parts = [action]

  if spool_id is not None:
    parts.append(f"spool_id={spool_id}")

  if status is not None:
    parts.append(f"status={status}")

  if payload is not None:
    try:
      payload_str = json.dumps(payload)
    except TypeError:
      payload_str = str(payload)
    parts.append(f"payload={payload_str}")

  try:
    append_to_rotating_file(SPOOLMAN_LOG_FILE, " | ".join(parts))
  except Exception:
    pass

def patchExtraTags(spool_id, old_extras, new_extras):
  for key, value in new_extras.items():
    old_extras[key] = value

  resp = requests.patch(f"{SPOOLMAN_API_URL}/spool/{spool_id}", json={
    "extra": old_extras
  })
  _log_spoolman_change(
    "patch_extra_tags",
    spool_id=spool_id,
    payload={"extra": old_extras},
    status=resp.status_code,
  )
  #print(resp.text)
  #print(resp.status_code)


def getSpoolById(spool_id):
  response = requests.get(f"{SPOOLMAN_API_URL}/spool/{spool_id}")
  #print(response.status_code)
  #print(response.text)
  return response.json()


def fetchSpoolList():
  if SPOOL_SORTING:
    response = requests.get(f"{SPOOLMAN_API_URL}/spool?sort={SPOOL_SORTING}")
  else:
    response = requests.get(f"{SPOOLMAN_API_URL}/spool")

  #print(response.status_code)
  #print(response.text)
  return response.json()

def consumeSpool(spool_id, use_weight=None, use_length=None):
  if use_weight is None and use_length is None:
    raise ValueError("use_weight or use_length is required")

  payload = {}
  if use_weight is not None:
    payload["use_weight"] = use_weight
  if use_length is not None:
    payload["use_length"] = use_length

  log(f'Consuming {payload} from spool {spool_id}')

  response = requests.put(f"{SPOOLMAN_API_URL}/spool/{spool_id}/use", json=payload)
  _log_spoolman_change(
    "consume_spool",
    spool_id=spool_id,
    payload=payload,
    status=response.status_code,
  )
  #print(response.status_code)
  #print(response.text)

def fetchSettings():
  response = requests.get(f"{SPOOLMAN_API_URL}/setting/")
  #print(response.status_code)
  #print(response.text)

  # JSON in ein Python-Dictionary laden
  data = response.json()

  # Extrahiere die Werte aus den relevanten Feldern
  extra_fields_spool = json.loads(data["extra_fields_spool"]["value"])
  extra_fields_filament = json.loads(data["extra_fields_filament"]["value"])
  base_url = data["base_url"]["value"]
  currency = data["currency"]["value"]

  settings = {}
  settings["extra_fields_spool"] = extra_fields_spool
  settings["extra_fields_filament"] = extra_fields_filament
  settings["base_url"] = base_url.replace('"', '')
  settings["currency"] = currency.replace('"', '')

  return settings