Python API

Overview

V-Portal offers standard functions to interface with its features in Python.

Development Environment

The recommended development environment is VSCode with Remote Development using SSH . This allows you to directly program in the same environment as the NUC. The python programs are read from the /home/vikaso/programs directory, so all python programs should be written in there.

Features

Data Fields

Interact with Data Fields programmatically: read values, write values (when allowed), reset to defaults, and subscribe to value changes.

Quick Start

import core.data_fields
 
# Lookup (raises if not found)
field = core.data_fields.get("field_name")
 
# Read
current = field.get()
 
# Write (raises if not modifiable from Python or the incorrect type is provided)
field.set(1)
 
# Reset to default
field.reset()
 
# Subscribe to changes
def on_change(value):
    print(f"Changed to {value}")
 
# sub must be maintained and remain in-scope, otherwise the subscription is lost
sub = field.subscribe(on_change)
 
# Unsubscribe when done
sub.unsubscribe()

Module: core.data_fields

get(name: str) -> DataField

Fetch a Data Field by its unique name.

• Parameters
  • name: The data field identifier (string).
• Returns
  • DataField instance if found.
• Raises
  • DataFieldNotFoundError if the field does not exist.

Example

import core.data_fields as df
field = df.get("robotCycleTime")

Class: DataField

Represents a single data field with read/write/reset and change subscription.

get() -> Any

Read the current value.

• Returns: The field value (type depends on Data Field definition).
• Raises
  • RuntimeError if the field is not readable.

Example

value = field.get()
 
if isinstance(value, int) and value > 0:
  print(value)

set(value: Any) -> bool

Write a new value.

• Returns: bool. True if the value was successfully set, False otherwise.
• Validation: Value must match the field’s type.
• Raises
  • RuntimeError if field is not writable from Python.
  • RuntimeError Type mismatch: attempting to set value of wrong type. If the value fails type validation.

Example

field.set(1)  #sets the value to 1 if writable

reset() -> None

Reset the field to its default value.

• Raises
  • RuntimeError if field is not writable from Python.

Example

field.reset()

last_updated_ms() -> int

Get the epoch time the data field was last updated in milliseconds.

• Returns: int. The epoch time in milliseconds the data field was updated at.
• Validation: The Data Field must be persistent.
• Raises
  • RuntimeError if the Data Field is not persistent.

Example

import time
 
if field.last_updated_ms() > time.time_ns() - 1000:
  print("Data Field updated within 1 second")

subscribe(callback: Callable[[Any], None]) -> Subscription

Receive notifications when the field value changes.

• Parameters
  • callback(value): Function invoked with the new value.
• Returns
  • Subscription object. Call unsubscribe() to stop receiving updates.

The returned subscription object must remain within scope otherwise the subscription is lost!

• Delivery semantics
  • Callbacks are invoked after the field’s value changes.
  • Callbacks run on the internal dispatch loop (see Threading & Concurrency).

Example

def on_change(value):
 print(f"Changed to {value}")
 
sub1 = field.subscribe(on_change)
 
# Lambda functions can also be used
sub2 = field.subscribe(lambda v: print(f"Changed to {v}"))
 
# Later…
sub1.unsubscribe()
sub2.unsubscribe()

Calling field.subscribe(…) without assigning its return value to a variable causes the subscription to go out of scope and removed.

Class: Subscription

Represents an active subscription to field changes.

unsubscribe() -> None

Stops further callbacks for this subscription.

Example

sub = field.subscribe(lambda v: print(v))
sub.unsubscribe()

Threading & Concurrency

• Callbacks: Subscription callbacks run on an internal dispatch thread (or event loop). Keep callbacks fast (non‑blocking) and offload heavy work to your own worker thread/pool if needed.
• Reentrancy: Avoid calling set() on the same field from within its own callback unless your application logic explicitly supports it.

Usage Patterns

Safe read‑modify‑write

f = core.data_fields.get("threshold")
curr = f.get()
new = max(curr, 10) # business rule
f.set(new) # may raise RuntimeError

Guarded write with fallback

try:
 core.data_fields.get("mode").set("auto")
except Exception as E:
 # fallback behavior or log
 pass

Managed Subscriptions

cache = {"field" : 0}
 
def add_subscription(dfield:DataField,
 callback: Callable,
 error_callback: Callable,
 *args, **kwargs):
  def wrapper(dfield_value):
      try:
          return callback(dfield_value, dfield, *args, **kwargs)
      except Exception as e:
          if error_callback is not None:
              error_callback(e)
  return dfield.subscribe(wrapper)
 
def update_internal_map(value, dfield, name):
  global cache
  cache[name] = value
 
field = core.data_fields.get("field")
sub = add_subscription(field, update_internal_map, None, name="field")
# ... later
sub.unsubscribe()

Complex Data Types

Data Fields can be created with a string type with a max limit of 1000 characters. This allows use for more complex data that can be stored in various forms, e.g. JSON, XML.

import json
from dataclasses import dataclass, asdict
from typing import List, Optional
import core.data_fields as df
 
# Example target object
@dataclass
class SensorConfig:
    name: str
    threshold: float
    enabled: bool = True
    tags: List[str] = None
 
    @staticmethod
    def from_dict(d: dict) -> "SensorConfig":
        return SensorConfig(
            name=d["name"],
            threshold=float(d["threshold"]),
            enabled=bool(d.get("enabled", True)),
            tags=d.get("tags") or [],
        )
 
# 1) Get the field (string, max 1000 chars, storing JSON)
field = df.get("sensorConfig")
 
# 2) Parse JSON and convert to object
raw = field.get()  # expects a JSON string, e.g. {"name":"A1","threshold":12.5,"enabled":true,"tags":["lab"]}
try:
    data = json.loads(raw)
except (TypeError, json.JSONDecodeError) as e:
    raise ValueError(f"Invalid JSON in Data Field 'sensor.config': {e}")
 
cfg = SensorConfig.from_dict(data)
 
# --- Use cfg --- #
print(cfg)
print(cfg.name, cfg.threshold, cfg.enabled, cfg.tags)
 
# Optional: modify and write back (re‑serialize to JSON)
cfg.threshold = 15.0
field.set(json.dumps(asdict(cfg)))

Notes & Best Practices

• Lifecycle: Always call unsubscribe() before shutdown to avoid dangling callbacks.
• Defaults: reset() sets the value to the field’s configured default, not None or zero (unless that is the configured default).

Popups

Display a popup from Python, block until the user responds (like input()), and receive both the pressed action and any data entered/selected in the popup.

Quick Start

import core.popups
 
# Blocks until user submits; returns PopupResult
 
result = popups.invoke_generic(
 "Get Batch Amount", 
 "info",
 json.dumps({"text_question": "Do you want to proceed with the operation?",}), # Initial values / UI inputs
 json.dumps([{"buttonTitle": "Submit", "colour": "#00FF00", "returnValue": "proceed"}]), # Button row
 0
)
 
print(result.wait())
# -> {"actionPressed":"proceed","returnData":{"text_question":""},"sessionId":10,"success":true}