Python Scripting

Last updated

Scripts and script libraries can be written in Python as well as JavaScript. Python scripts have the same triggers, the same run log, History, _STATUS tags, workflow Run Script node and dashboard buttons, and the same Scada.Ink API, in Python style. numpy, pandas, SciPy, statsmodels and scikit-learn are built in, so history analysis, SPC beyond the built-in charts, regression and anomaly detection are a few lines away.

Python scripting is included on every appliance but starts switched off. While it is off, the Scripts list, the script editor and the Script Libraries page look and work exactly as they do for JavaScript only.

Turning it on

Sign in as an administrator.
Open Settings → Feature Toggles.
Switch on Python Scripting (under Control & Logic).

What appears once it is on:

a Language choice (JavaScript or Python) next to the name box when you create a script or a library;
a Python badge on Python scripts and libraries in their lists;
in the editor of a Python script: Python syntax highlighting and autocomplete, a Check button, an Own interpreter option on the Trigger tab, and a note that Python scripts start with network access off;
Python Packages in the Administration part of the sidebar (administrators only).

Switching it off again keeps every Python script and library. They stop running, and each attempt is logged: Python scripting is turned off on this appliance, so this Python script did not run. Existing scripts always stay in the language they were created in.

Writing a Python script

The API is ready as globals, with no import:

python

speed = tags.read("CNC1", "Spindle/Speed")
load = tags.read("CNC1", "Spindle/Load")

if speed and load:
    efficiency = float(load) / float(speed) * 100
    tags.write("computed/efficiency", round(efficiency, 1))
    log("Efficiency:", f"{efficiency:.1f}%")

Every API call is a plain, synchronous call: no await.
Names are Python style: tags.read_qualified, work_orders.add_counts, named_queries.run, alarms.get_active().
log() and print() both write to the run log. Warnings Python prints go to the log with WARN:.
context is a dict of the trigger's data: context["trigger"], context.get("value").
To hand a value back (a message handler's reply, a library Test's answer), assign it to a top-level variable named result.
A failed API call raises ScadaInkError; an HTTP error status raises HttpError, which has .status and .body.
with db.transaction(): makes the db calls inside it commit together, or not at all.
time.sleep(seconds) works and still lets the appliance stop the script.
timers.set_timeout(fn, ms), timers.set_interval(fn, ms) and timers.clear(id) follow the JavaScript timer rules (20 pending, intervals at least 100 ms). Callbacks run after the script's own code has finished.
history.query(..., as_frame=True) returns a pandas DataFrame indexed by time.

In a library, or anywhere you prefer explicit imports: from scadaink import tags, db, log.

Check, errors and Go to line

Check compiles the script without running it and reports the first syntax error with its line and column, with a Go to line button. A failed run's error ends with the line, for example KeyError: 'missing' (line 3); the run log adds a traceback limited to your script and your libraries, and the Log tab's Go to line button puts the cursor on it.

Python libraries

A Python library is a module: scripts import it by its name (import shift_utils). Its name must be a valid module name, must not be a Python keyword, a built-in script name, a standard-library module (json, time…) or an installed package. JavaScript scripts don't see Python libraries, and Python scripts don't see JavaScript ones. Libraries are read fresh for every run, so a saved change applies to the next run. In a library Test, the library is already imported by name; set result to see a value.

Speed, interpreters and limits

Python runs in Pyodide (CPython 3.14 compiled to WebAssembly) inside the same locked-down script host as JavaScript. Starting an interpreter takes a second or two, and importing pandas or SciPy several more, so interpreters are kept and reused: imported modules stay loaded, and each run still starts with fresh variables. An interpreter is only reused for scripts with the same author and the same network switches, so one person's code never runs with another's authority. Tick Own interpreter on a script to give every run a fresh interpreter that is discarded afterwards: slower, but nothing at all is shared.

Limit	Value
Time per run	30 s, as for JavaScript. Python gets a `KeyboardInterrupt` (so `finally` blocks and `db.transaction` roll back); an interpreter that was stopped is never reused
Memory	384 MB of Python memory per run. Past it the run fails with Script exceeded the 384 MB Python memory limit and the interpreter is replaced
Timers	20 pending; each callback at most 5 s
Interpreters	1 on an appliance with under 3 GB of memory, 2 under 6 GB, otherwise 3; further runs wait in a queue of 32

What Python scripts cannot do

Reach the appliance's files, database, licence or operating system: the script host can't see them.
Open network sockets. Web requests go through the API: http.get, or urllib.request, requests and httpx, which are routed through it. See Script Network Access.
Use JavaScript from Python: import js, pyodide, pyodide.ffi and micropip are refused.
Load compiled code of their own: ctypes only opens libraries that came with an installed package.
Install packages. An administrator does that on the Python Packages page.

The full REST API

api.get("/devices"), api.post(...), api.put(...) and api.delete(...) call any Scada.Ink REST route as the script's author, with every permission check the web app gets, and return the route's data. A script last saved by an administrator acts as that administrator; any other script acts as its author. Appliance, licence, feature toggle, software update, certificate, restore, reset, API token, security, Python package and script network routes are refused for every script, however it was saved. Reading settings stays allowed.

API: JavaScript and Python names

Generated from the same definition as the code, so it always matches. Functions marked admin work only in scripts last saved by an administrator.

JavaScript	Python	What it does
`devices.find(name)`	`devices.find(name)`	Find a device ID by name
`devices.localId()`	`devices.local_id()`	The local (memory tag) device ID
`devices.list()`	`devices.list()`	List all devices
`tags.read(device, tagPath)`	`tags.read(device, tag_path)`	Read a raw tag value. Device is an ID or a name
`tags.readQualified(device, tagPath)`	`tags.read_qualified(device, tag_path)`	Read a tag with its quality and timestamp
`tags.readAll(device)`	`tags.read_all(device)`	Read every tag of a device
`tags.readAllQualified(device)`	`tags.read_all_qualified(device)`	Read every tag of a device with quality metadata
`tags.write(tagPath, value)`	`tags.write(tag_path, value)`	Write a local tag
`tags.writeDevice(device, tagPath, value)`	`tags.write_device(device, tag_path, value)`	Write a tag on a physical device (write access must be enabled)
`await tags.readBatch(device, tagPaths)`	`tags.read_batch(device, tag_paths)`	Read several tags of one device in one call
`await tags.writeBatch(entries)`	`tags.write_batch(entries)`	Write several local tags in one call
`await http.request(method, url, body?, headers?)`	`http.request(method, url, body=None, headers=None)`	Any request. Returns { status, ok, headers, body } and never fails on status
`await http.get(url, headers?)`	`http.get(url, headers=None)`	GET (10 s limit). Returns the body; fails on a non-2xx status
`await http.post(url, body?, headers?)`	`http.post(url, body=None, headers=None)`	POST. Returns the body; fails on a non-2xx status
`await http.put(url, body?, headers?)`	`http.put(url, body=None, headers=None)`	PUT. Returns the body; fails on a non-2xx status
`await http.delete(url, headers?)`	`http.delete(url, headers=None)`	DELETE. Returns the body; fails on a non-2xx status
—	`http.raw_request(method, url, body_base64=None, headers=None)`	Bytes in, bytes out (base64). urllib, requests and httpx use it; the same network rules apply
`await namedQueries.run(name, params?)`	`named_queries.run(name, params=None)`	Run a named query; returns its rows
`await namedQueries.runFull(name, params?)`	`named_queries.run_full(name, params=None)`	Run a named query; returns rows plus row count and timing
`namedQueries.list()`	`named_queries.list()`	Named queries this script may run
`db.query(table, options?)`	`db.query(table, options=None)`	Query rows from a Data Table
`db.insert(table, row)`	`db.insert(table, row)`	Insert a row
`db.update(table, rowId, changes)`	`db.update(table, row_id, changes)`	Update a row
`db.delete(table, rowId)`	`db.delete(table, row_id)`	Delete a row
`db.tables()`	`db.tables()`	List all Data Table names
`alarms.getActive()`	`alarms.get_active()`	Active alarms
`alarms.notify(severity, title, message?)`	`alarms.notify(severity, title, message=None)`	Raise an alarm (info, warning or error). Returns the alarm ID
`alarms.acknowledge(alarmId)`	`alarms.acknowledge(alarm_id)`	Acknowledge an active alarm
`alarms.clear(alarmId)`	`alarms.clear(alarm_id)`	Clear an active alarm
`auth.login(email, password)`	`auth.login(email, password)`	Check credentials; returns the user and tokens, or null
`users.list()`	`users.list()`	List all users (admin)
`users.add(user)`	`users.add(user)`	Create a user; returns the ID (admin)
`users.update(id, changes)`	`users.update(id, changes)`	Change a user's email, name or role (admin)
`users.setPassword(id, newPassword)`	`users.set_password(id, new_password)`	Reset a user's password (admin)
`users.remove(id)`	`users.remove(id)`	Permanently delete a user (admin)
`users.disable(id)`	`users.disable(id)`	Block sign-in but keep the account and its records (admin)
`users.enable(id)`	`users.enable(id)`	Re-enable a disabled user (admin)
`users.logout(idOrEmail)`	`users.logout(id_or_email)`	Sign a user out everywhere (admin)
`users.logoutAll()`	`users.logout_all()`	Sign every user out (admin)
`system.backup()`	`system.backup()`	Create a database backup on disk (admin)
`system.info()`	`system.info()`	Runtime information (version, uptime…)
`email.isConfigured()`	`email.is_configured()`	True if email delivery is set up
`await email.send(options)`	`email.send(options)`	Send an email
`await history.query(device, tagPath, start, stop?)`	`history.query(device, tag_path, start, stop=None, as_frame=False)`	Raw historian samples. In Python, pass as_frame=True for a pandas DataFrame
`await history.queryAggregated(device, tagPath, start, stop, window, fn)`	`history.query_aggregated(device, tag_path, start, stop, window, fn, as_frame=False)`	Aggregated historian samples. In Python, pass as_frame=True for a pandas DataFrame
`workflows.find(name)`	`workflows.find(name)`	Find a workflow ID by name
`workflows.list()`	`workflows.list()`	List all workflows
`await workflows.run(idOrName, params?)`	`workflows.run(id_or_name, params=None)`	Run a workflow and wait for it
`workflows.trigger(idOrName, params?)`	`workflows.trigger(id_or_name, params=None)`	Start a workflow without waiting for it
`await workflows.setEnabled(idOrName, enabled)`	`workflows.set_enabled(id_or_name, enabled)`	Enable or disable a workflow
`await workflows.enable(idOrName)`	`workflows.enable(id_or_name)`	Turn a workflow's triggers on
`await workflows.disable(idOrName)`	`workflows.disable(id_or_name)`	Turn a workflow's triggers off
`workflows.status(idOrName)`	`workflows.status(id_or_name)`	Enabled state plus the last run's status and time
`ladder.find(name)`	`ladder.find(name)`	Find a ladder task ID by name
`ladder.list()`	`ladder.list()`	List ladder tasks with their running state
`ladder.start(idOrName)`	`ladder.start(id_or_name)`	Start scanning a ladder task
`ladder.stop(idOrName)`	`ladder.stop(id_or_name)`	Stop scanning a ladder task
`ladder.isRunning(idOrName)`	`ladder.is_running(id_or_name)`	True while the task is scanning
`ladder.status(idOrName)`	`ladder.status(id_or_name)`	Full scan status (scan times, faults, rung states)
`messages.send(name, payload?, options?)`	`messages.send(name, payload=None, options=None)`	Send a named message; returns how many scripts and workflows received it
`await messages.sendRequest(name, payload?, options?)`	`messages.send_request(name, payload=None, options=None)`	Send a message and wait for the first handler's result
`workOrders.list(options?)`	`work_orders.list(options=None)`	Work orders, highest priority first
`workOrders.get(idOrNumber)`	`work_orders.get(id_or_number)`	One work order by ID or number
`workOrders.active(deviceId)`	`work_orders.active(device_id)`	The active work order on a device
`await workOrders.start(idOrNumber)`	`work_orders.start(id_or_number)`	Activate a pending work order (downloads its recipe if set to)
`workOrders.complete(idOrNumber)`	`work_orders.complete(id_or_number)`	Complete an active work order
`workOrders.addCounts(idOrNumber, good, reject?)`	`work_orders.add_counts(id_or_number, good, reject=None)`	Add to the counts; returns the new totals
`workOrders.setCounts(idOrNumber, counts)`	`work_orders.set_counts(id_or_number, counts)`	Set the counts outright
—	`api.request(method, path, body=None, query=None)`	Call any Scada.Ink REST route as this script's author. Returns
`console.log(...args)`	—	Write to the run log
`console.warn(...args)`	—	Write a WARN line to the run log
`console.error(...args)`	—	Write an ERROR line to the run log
`db.transaction(callback)`	`db.transaction()`	with db.transaction(): — the db calls inside commit together or not at all
`expression.eval(expr)`	—	Evaluate an expression string with {tag} references
—	`api.get(path, query=None)`	GET a REST route as this script's author; returns the response data, fails on an error status
—	`api.post(path, body=None)`	POST to a REST route; returns the response data
—	`api.put(path, body=None)`	PUT to a REST route; returns the response data
—	`api.delete(path)`	DELETE a REST route; returns the response data
—	`timers.set_timeout(fn, ms, *args)`	Call fn once after ms milliseconds (saved scripts only)
—	`timers.set_interval(fn, ms, *args)`	Call fn every ms milliseconds, at least 100 (saved scripts only)
—	`timers.clear(timer_id)`	Cancel a timer

Python Scripting ​

Turning it on ​

Writing a Python script ​

Check, errors and Go to line ​

Python libraries ​

Speed, interpreters and limits ​

What Python scripts cannot do ​

The full REST API ​

API: JavaScript and Python names ​

See also ​