Read and change instrument settings as JSON

"""Read, inspect and change instrument settings through SvanLINK's setup JSON.

    from svanlink_setup import SvanLink

    svanlink = SvanLink("http://192.168.1.10:8000", token="userToken")
    setup = svanlink.read("123456")                      # getSetupJSON
    print(setup.get("Measurement/Profile 1/Filter"))     # A

    change = {"Measurement/Profile 1/Filter": "C"}
    print(svanlink.change(setup, change, check=True))    # what would change
    svanlink.change(setup, change)                       # write it

Needs Python 3.8+ and `requests`.
"""
import requests


class SvanLinkError(Exception):
    """SvanLINK answered with Status "error". The message says why."""


def _path(path):
    """'Measurement/Profile 1/Filter' or a tuple of names -> a tuple of names."""
    return tuple(path.split("/")) if isinstance(path, str) else tuple(path)


def _is_setting(node):
    # In the values tree a setting is text. With "options": true it is an
    # object that has a code and a type. A group is any other object.
    return not isinstance(node, dict) or ("code" in node and "type" in node)


def patch(changes):
    """{"Measurement/Profile 1/Filter": "C", ...} -> the nested tree setSetupJSON takes.

    A key can also be a tuple of names, which you need for names that contain
    "/" (such as "1/3 Octave Source"); find() returns paths in that form."""
    tree = {}
    for path, value in changes.items():
        *groups, name = _path(path)
        node = tree
        for group in groups:
            node = node.setdefault(group, {})
        node[name] = value
    return tree


def holds(condition, raw):
    """Whether a visibility `when` or a choice's `allowedIf` holds for these raw values."""
    if isinstance(condition, bool):          # already worked out by SvanLINK
        return condition
    if "and" in condition:
        return all(holds(c, raw) for c in condition["and"])
    if "or" in condition:
        return any(holds(c, raw) for c in condition["or"])
    current = raw.get(condition["param"])
    if current is None:
        return False
    op, value = condition["op"], condition["value"]
    if op == "==":
        return str(current).strip() == value
    if op == "!=":
        return str(current).strip() != value
    try:                                     # < > <= >= compare whole numbers
        left, right = int(current), int(value)
    except (TypeError, ValueError):
        return False
    return {"<": left < right, ">": left > right, "<=": left <= right, ">=": left >= right}[op]


class Setup:
    """One getSetupJSON answer."""

    def __init__(self, response, serial=None):
        self.serial = serial
        self.device_type = response["deviceType"]
        self.fingerprint = response["fingerprint"]
        self.tree = response["settings"]

    def items(self):
        """(path, setting) for every setting. A path is a tuple of names."""
        def walk(node, path):
            for name, child in node.items():
                if _is_setting(child):
                    yield path + (name,), child
                else:
                    yield from walk(child, path + (name,))
        return walk(self.tree, ())

    @property
    def detailed(self):
        """True when read with "options": true."""
        return any(isinstance(setting, dict) for _, setting in self.items())

    def get(self, path):
        """The setting at a path: text, or an object when detailed."""
        node = self.tree
        for name in _path(path):
            node = node[name]
        return node

    def value(self, path):
        """The readable value at a path, in either form."""
        setting = self.get(path)
        return setting["value"] if isinstance(setting, dict) else setting

    def find(self, name):
        """Paths of every setting with this name, or with this code when detailed."""
        return [path for path, setting in self.items()
                if path[-1] == name or (isinstance(setting, dict) and setting.get("code") == name)]

    def raw_values(self):
        """{code: raw} for every setting, the values conditions compare. Detailed only."""
        raw = {}
        for _, setting in self.items():
            if "raw" in setting:
                raw[setting["code"]] = setting["raw"]
            elif "codes" in setting:         # a time of day: hour, minute, second
                for code, part in zip(setting["codes"], setting["value"].split(":")):
                    raw[code] = str(int(part))
        return raw

    def is_visible(self, setting, raw=None):
        """Whether the editor shows a setting, for these raw values (default: as read)."""
        if "visibility" not in setting:
            return setting["visible"]
        raw = self.raw_values() if raw is None else raw
        shown = setting["baseVisible"]
        for rule in setting["visibility"]:   # in order; the last rule that holds wins
            if holds(rule["when"], raw):
                shown = rule["show"]
        return shown

    def allowed_choices(self, setting, raw=None):
        """The choices that are legal now (a choice with allowedIf only while it holds)."""
        raw = self.raw_values() if raw is None else raw
        return [c for c in setting.get("choices", []) if holds(c.get("allowedIf", True), raw)]


class SvanLink:
    """A SvanLINK host. `url` is its API address, e.g. http://192.168.1.10:8000"""

    def __init__(self, url, token=None, timeout=120):
        self.url = url.rstrip("/") + "/"
        self.token = token
        self.timeout = timeout    # reading and writing a setup takes seconds

    def call(self, request, **params):
        body = {"Request": request, "Params": params}
        if self.token:
            body["token"] = self.token
        reply = requests.post(self.url, json=body, timeout=self.timeout)
        reply.raise_for_status()
        data = reply.json()
        if data.get("Status") != "ok":
            raise SvanLinkError(data.get("StatusMessage") or request + " failed")
        return data["Response"]

    def read(self, serial, options=False):
        """The connected instrument's current setup."""
        return Setup(self.call("getSetupJSON", device=serial, options=options), serial)

    def change(self, setup, changes, check=False):
        """Write {path: value} to the instrument `setup` was read from.

        Refused if its setup changed since that read. check=True only reports
        what would change. A running measurement is stopped for the write and
        started again ("restarted": true)."""
        return self.call("setSetupJSON", device=setup.serial, fingerprint=setup.fingerprint,
                         settings=patch(changes), check=check)

// svanlink-setup.js: read, inspect and change instrument settings through SvanLINK's setup JSON.
//
//   import { SvanLink } from './svanlink-setup.js';
//   const svanlink = new SvanLink('http://192.168.1.10:8000', 'userToken');
//   const setup = await svanlink.read('123456');                // getSetupJSON
//   setup.value('Measurement/Profile 1/Filter');                // 'A'
//   const change = { 'Measurement/Profile 1/Filter': 'C' };
//   await svanlink.change(setup, change, { check: true });      // what would change
//   await svanlink.change(setup, change);                       // write it
//
// Works in Node 18+ and in browsers (both have fetch).

export class SvanLinkError extends Error {}

const toPath = (path) => (typeof path === 'string' ? path.split('/') : [...path]);

// In the values tree a setting is text. With "options": true it is an object
// that has a code and a type. A group is any other object.
const isSetting = (node) =>
  typeof node !== 'object' || node === null || ('code' in node && 'type' in node);

/**
 * { 'Measurement/Profile 1/Filter': 'C', ... } -> the nested tree setSetupJSON takes.
 * Also takes [[path, value], ...], where a path can be an array of names
 * (needed for names that contain '/', such as '1/3 Octave Source').
 */
export function patch(changes) {
  const tree = {};
  for (const [path, value] of Array.isArray(changes) ? changes : Object.entries(changes)) {
    const names = toPath(path);
    const name = names.pop();
    let node = tree;
    for (const group of names) node = node[group] ??= {};
    node[name] = value;
  }
  return tree;
}

/** Whether a visibility `when` or a choice's `allowedIf` holds for these raw values. */
export function holds(condition, raw) {
  if (typeof condition === 'boolean') return condition;   // already worked out by SvanLINK
  if (condition.and) return condition.and.every((c) => holds(c, raw));
  if (condition.or) return condition.or.some((c) => holds(c, raw));
  const current = raw[condition.param];
  if (current === undefined || current === null) return false;
  const { op, value } = condition;
  if (op === '==') return String(current).trim() === value;
  if (op === '!=') return String(current).trim() !== value;
  const left = Number(current), right = Number(value);      // < > <= >= compare whole numbers
  if (!Number.isInteger(left) || !Number.isInteger(right)) return false;
  return { '<': left < right, '>': left > right, '<=': left <= right, '>=': left >= right }[op];
}

/** One getSetupJSON answer. */
export class Setup {
  constructor(response, serial) {
    this.serial = serial;
    this.deviceType = response.deviceType;
    this.fingerprint = response.fingerprint;
    this.tree = response.settings;
  }

  /** [path, setting] for every setting. A path is an array of names. */
  *items(node = this.tree, path = []) {
    for (const [name, child] of Object.entries(node)) {
      if (isSetting(child)) yield [[...path, name], child];
      else yield* this.items(child, [...path, name]);
    }
  }

  /** True when read with "options": true. */
  get detailed() {
    for (const [, setting] of this.items()) return typeof setting === 'object';
    return false;
  }

  /** The setting at a path: text, or an object when detailed. */
  get(path) {
    return toPath(path).reduce((node, name) => node[name], this.tree);
  }

  /** The readable value at a path, in either form. */
  value(path) {
    const setting = this.get(path);
    return typeof setting === 'object' ? setting.value : setting;
  }

  /** Paths (arrays of names) of every setting with this name, or with this code when detailed. */
  find(name) {
    const found = [];
    for (const [path, setting] of this.items()) {
      if (path.at(-1) === name || setting?.code === name) found.push(path);
    }
    return found;
  }

  /** { code: raw } for every setting, the values conditions compare. Detailed only. */
  rawValues() {
    const raw = {};
    for (const [, setting] of this.items()) {
      if ('raw' in setting) raw[setting.code] = setting.raw;
      else if (setting.codes) {                               // a time of day: hour, minute, second
        setting.value.split(':').forEach((part, i) => {
          if (setting.codes[i]) raw[setting.codes[i]] = String(parseInt(part, 10));
        });
      }
    }
    return raw;
  }

  /** Whether the editor shows a setting, for these raw values (default: as read). */
  isVisible(setting, raw = this.rawValues()) {
    if (!setting.visibility) return setting.visible;
    let shown = setting.baseVisible;
    for (const rule of setting.visibility) {                 // in order; the last rule that holds wins
      if (holds(rule.when, raw)) shown = rule.show;
    }
    return shown;
  }

  /** The choices that are legal now (a choice with allowedIf only while it holds). */
  allowedChoices(setting, raw = this.rawValues()) {
    return (setting.choices ?? []).filter((c) => holds(c.allowedIf ?? true, raw));
  }
}

/** A SvanLINK host. `url` is its API address, e.g. http://192.168.1.10:8000 */
export class SvanLink {
  constructor(url, token) {
    this.url = url.replace(/\/+$/, '') + '/';
    this.token = token;
  }

  async call(request, params) {
    const body = { Request: request, Params: params };
    if (this.token) body.token = this.token;
    const reply = await fetch(this.url, {
      method: 'POST',
      headers: { 'Content-Type': 'application/json' },
      body: JSON.stringify(body),
    });
    if (!reply.ok) throw new SvanLinkError(`${request}: HTTP ${reply.status}`);
    const data = await reply.json();
    if (data.Status !== 'ok') throw new SvanLinkError(data.StatusMessage || `${request} failed`);
    return data.Response;
  }

  /** The connected instrument's current setup. */
  async read(serial, { options = false } = {}) {
    return new Setup(await this.call('getSetupJSON', { device: serial, options }), serial);
  }

  /** Write { path: value } to the instrument `setup` was read from. See the Python version. */
  change(setup, changes, { check = false } = {}) {
    return this.call('setSetupJSON', {
      device: setup.serial, fingerprint: setup.fingerprint, settings: patch(changes), check,
    });
  }
}