ci-server-engineer/.claude/skills/pytools/tasks/event-notification.md

task	version	source	status
Wait for value or record changes with DSS event notification in a Python script	R1.05	IM 36K10P40-01EN §3.10, §3.11 (stop request), §4 (2nd ed., Feb. 2026); `EVENT_Q_SLOTS` in the installed `dss.sup` (R1.03 and R1.05)	from docs, not tested (no Python on the R1.05 test server)

Event notification

Start from setup-first-script.md. The example only reads; it leaves nothing behind when it ends (notifications stopped, dataset and connection closed).

1. How it works

• Request: per record dss.recordEventNotify(conn, ds, <key>, 'u') (u update, d delete), or per dataset dss.datasetEventNotify(conn, ds, 'iud') (also i insert).
• DSS puts each change in one queue per client; the script must poll it with dss.eventRead(conn) → list of dicts {'ds_name', 'event', 'id', 'record', 'set_hnd'} (record = the record after the change, only the fields opened). Empty list = nothing new.
• If the queue overflows, DSS empties it (events lost). Size: EVENT_Q_SLOTS in dss.sup (default 100); poll often and keep the number of watched records reasonable.
• One request per record; to change the mode, stop it first (PY_DSS_EVT_NOTIFY_DEF).
• Stop with recordEventStop / datasetEventStop, then close.

2. Example: print value changes of two items

import os, sys, time, getpass
sys.path.insert(0, os.path.join(os.environ['TLS_ROOT_PATH'], 'tls', 'exe'))
import dss

ITEMS = ['CI-Eng-Sample.TANK1.LEVEL', 'CI-Eng-Sample.PUMP1.RUNNING']   # or <Section>.<Tag>
RUN_SECONDS = 60                                                       # 0 = until Ctrl+C

conn = dss.connect(os.environ.get('CI_USER') or input('User: '),
                   os.environ.get('CI_PASSWORD') if 'CI_PASSWORD' in os.environ
                   else getpass.getpass('Password: '))
watched = []
try:
    ds = dss.openDataset(conn, 'ITEM_VAL', ['NAME', 'ITEM_VALUE', 'STATUS', 'UPDATE_TIME'], 'r')
    try:
        for name in ITEMS:
            dss.recordEventNotify(conn, ds, name, 'u')
            watched.append(name)
        t_end = time.time() + RUN_SECONDS
        try:
            while not RUN_SECONDS or time.time() < t_end:
                time.sleep(1)
                for ev in dss.eventRead(conn):
                    r = ev['record']
                    print(dss.dateString(conn, r['UPDATE_TIME'], 'GENERAL'),
                          r['NAME'], r['ITEM_VALUE'], r['STATUS'])
        except KeyboardInterrupt:
            pass
    finally:
        for name in watched:
            dss.recordEventStop(conn, ds, name)
        dss.closeDataset(conn, ds)
finally:
    dss.close(conn)

Expected: one line per value update of the two items (with the sample's live-data feed running, TANK1.LEVEL changes every few seconds); nothing for unchanged items. Ends after 60 s or Ctrl+C.

3. Long-running scripts

• Give the script a fixed BUS name before connecting: dss.busSetName('PYWATCH'). It can then be stopped cleanly from a command prompt with durstp -p PYWATCH; inside the script this raises dss.error with code PY_DSS_STOP_REQUEST — catch it, stop the notifications, close and exit. Without a name the process is PYDSS_<number> (new each run). Check whether it runs with durwfp -p PYWATCH -t 0 (skill system-management, bus-utilities.md).
• Instead of time.sleep polling, a script that also serves BUS messages can block in dss.busReceive(conn, 'SYNC_WAIT') (manual §3.11); then events are only read between messages.
• Do not leave test watchers running on a shared or production server.

4. Not tested / to verify

On a test Core with Python and the sample feed (user): run §2 (expect changes of TANK1.LEVEL); start it with a BUS name and stop it with durstp (expect a clean exit via PY_DSS_STOP_REQUEST); check whether the R1.05 module still names unnamed scripts PYDSS_<n> (the dss.sup routing PYDSS* → DSSSDAPY relies on it, while a script with its own BUS name such as PYWATCH may be routed to a general DSSSDA instead, ISS-180).