ci-server-engineer/.claude/skills/pytools/tasks/read-items-history.md

task	version	source	status
Read item values and definitions, and item history / aggregates, from a Python script	R1.05	IM 36K10P40-01EN §3.3, §3.5, §3.12, §3.13, §4 (2nd ed., Feb. 2026); field names from the R1.05 data dictionary (copied from the R1.05 test server 2026-10-08); `aggr_period` and its values checked in the installed R1.05 `dss.pyd` (2026-10-09)	from docs, not tested (no Python on the R1.05 test server)

Read items and item history

Start from setup-first-script.md (setup, logon). Only read mode ('r') is used here.

1. Read all items of a section (ITEM_VAL)

ITEM_VAL = current values (NAME, ITEM_VALUE, ITEM_VALUE_STR, STATUS, QUALITY, UPDATE_TIME, limits, IN_ALARM, ENG_UNIT, SCALE_HIGH_LIMIT, SCALE_LOW_LIMIT …); ITEM_DF = definitions (description, type, limits, history and alarm settings). See all fields with dss.getFieldNames(conn, 'ITEM_VAL') and their type/length with dss.getFieldProperties(conn, 'ITEM_VAL', 'NAME').

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

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: '))
try:
    ds = dss.openDataset(conn, 'ITEM_VAL',
                         ['NAME', 'ITEM_VALUE', 'ITEM_VALUE_STR', 'STATUS', 'ENG_UNIT'], 'r')
    try:
        dss.createFilter(conn, ds, '"CI-Eng-Sample.*"')     # or '"<Section>.*"'
        dss.activateFilter(conn, ds)
        while True:
            try:
                r = dss.readNext(conn, ds)
            except dss.error as e:
                if e.args[0] == 'DSS-E-EOS':                # end of the dataset
                    break
                raise
            print('%-40s %12s %-8s %s' % (r['NAME'], r['ITEM_VALUE_STR'] or r['ITEM_VALUE'],
                                           r['ENG_UNIT'], r['STATUS']))
        dss.deactivateFilter(conn, ds)
        dss.deleteFilter(conn, ds)
    finally:
        dss.closeDataset(conn, ds)
finally:
    dss.close(conn)

• One item: dss.readEqual(conn, ds, 'CI-Eng-Sample.TANK1.LEVEL') (no filter needed).
• Small result sets: dss.readRecords(conn, ds) returns a list at once (the filter applies).
• Time fields are floats (seconds since 1970, local time unless the dataset was opened with 'rt' = UTC); format with dss.dateString(conn, t, 'GENERAL').

2. Item history (fast path)

getItemHistory asks ITEM Service directly (faster than ITEM_HISTORY through DSS). The item must be in the storage group you name (CI-Eng-Sample: CIENG_EVENT and CIENG_SCAN10S hold TANK1.LEVEL and TANK1.INFLOW; CIENG_AGG holds hourly and 30-minute max/min/average of TANK1.LEVEL; skill history-management).

# inside the try: after dss.connect(...)
print(dss.getHisGroupItems(conn, 'CIENG_EVENT'))           # items in the group

start = dss.dateRelDSS(conn, '-02:00')                       # 2 hours ago
end = dss.dateRelDSS(conn, '-00:00')                         # now
item, group = 'CI-Eng-Sample.TANK1.LEVEL', 'CIENG_EVENT'
while True:
    samples = dss.getItemHistory(conn, item, group, start, end, 500)
    done = samples and samples[-1] is None
    rows = [s for s in samples if s is not None]
    for (t, value, status, option_bits, quality) in rows:
        print(dss.dateString(conn, t, 'GENERAL'), value, status, quality)
    if done or not rows:
        break
    start = rows[-1][0]          # next block starts at the last time read (may repeat it once)

Optional named arguments:

• interval=60000 — interpolated values every 60 s (milliseconds) instead of the stored samples.
• attribute='HIGH_LIMIT' (also DEADBAND, LOW_LIMIT, LOW_LOW_LIMIT, HIGH_HIGH_LIMIT, SCALE_LOW, SCALE_HIGH) where the group type stores them.
• Aggregation group: attribute='AGGR_MAX' (AGGR_MIN, AGGR_AVERAGE, AGGR_INTG, AGGR_COUNT, AGGR_STDV, AGGR_DFMS) with aggr_period='HOUR' (30_MINUTES, SHIFT, DAY, WEEK, MONTH, YEAR; only periods configured for the group, and only aggregates stored for the item), e.g. dss.getItemHistory(conn, item, 'CIENG_AGG', start, end, 100, attribute='AGGR_MAX', aggr_period='HOUR').
• node=<number> — group on another node of a distributed system.

The status and quality numbers are the same codes as in ITEM_HISTORY / STATUS_DF (read STATUS_DF to translate status numbers, task filter-index.md).

Expected: a list of tuples per block; the last block ends with None. An item that is not in the group, or a group that does not exist, raises dss.error (code from ITEM Service).

3. Not tested / to verify

On a test Core with Python (user): run §1 against the sample section and compare with itmval -g (skill system-management, read-write-item.md); run §2 for CIENG_EVENT and compare with an ITEM_HISTORY report or rptgen query for the same period (skill reporting). Check whether the repeated boundary sample appears when a block is continued.