#!/usr/bin/env python3 """Generate the CI Server display XML for the WRPS demo. python build_display.py -> ./out/*.xml Emits three displays: WRPS_Overview.xml the single operator display, 1920 x 1080 WRPS_FP_Pump.xml pump faceplate (opened per pump) WRPS_FP_Setpoints.xml station setpoint faceplate Everything here follows the file format as observed in the live deployment copied to 99-reference/ciserver-hmi-deployment, not a manual: geometry x,y is the CENTRE; left/right/top/bottom are half-extents colours shared paints, referenced as ref="NN" live value + actionConnectTo -> property number.value -> connection {direction 1, itemName, itemAttribute ItemValue} write actionSetItemAttribute + filter (the value) + connection {direction 2, itemName, itemAttribute ItemValue} navigation actionActivateDisplay + NAME Style tokens and component rules come from component-kit.md. Nothing is hand-placed twice: a value box is one function, so every value box on every display is identical by construction. KNOWN GAP - state colouring is static in this version. Driving a fill colour from a value needs either item alarm limits (deferred by the user along with trends and alarms) or a threshold/transformation set up in the editor, which is not safe to guess at. Numbers, navigation and writes are live. See the README section "What still needs doing in the editor". """ import datetime import pathlib from pathlib import Path import point_format as pf OUT = Path(__file__).resolve().parent / "out" W, H = 1920, 1080 # ------------------------------------------------------------- item ids # A connection needs BOTH itemName and itemId; with the name alone every # value reads 0 - the whole screen looks alive and means nothing. Getting # this wrong is the single failure mode that has cost this project the most # time, so the rules below are enforced by the build, not by memory. # # THE POLICY: EVERY ID IS MEASURED. NOTHING IS INFERRED. # ------------------------------------------------------- # An id comes from CI Server or the build stops. There is no rule, no # default and no fallback, because an id we invented that happens to be # wrong produces exactly the failure that is hardest to see: a screen full # of live-looking zeroes. # # Ids are measured by saving a display in CI View - it resolves every # connection by NAME on save - and reading them back: # # python build_display.py --harvest # # WRPS_TagTest binds every project item, so saving that ONE display # measures all 49 in a single pass. That is the intended route. # # An earlier version derived the ids it had not measured, by arithmetic on # the item NSIDs. It no longer does: an id we invent that happens to be # wrong produces the failure that is hardest to see - a screen full of # live-looking zeroes. Every id is measured or the build stops. # # WHY THEY GO STALE # ----------------- # **Every .qli item import renumbers every item.** On 2026-08-14 the items # were re-imported to change engineering units, keeping every NAME, NSID and # ID_NUMBER byte-identical - and every id still moved. CI Server recreates # items on import rather than updating them in place. Every value on every # screen went dead at once. # # The cure is not arithmetic. It is: VALIDATE the display in CI Server's # Editor Module. That resolves every connection by name and heals the file. # Then re-harvest, so the next build ships the new ids. # # THE GUARDS - all hard failures, none is a warning # ------------------------------------------------ # 1. Every bound item must have a measured id. check_ids() lists all the # gaps at once, since one harvest fixes them together. # 2. item-ids.meta.json pins the SHA of the wrps_item_df.qli the ids were # harvested against. If that file has changed, the items have been or # will be re-imported, so the ids are presumed stale. # 3. --verify compares CI Server's own resolved ids against # what this build emits, and exits 1 naming every item that would read 0. # # The cure for all of them is the same one command, and every error says so. IDS_FILE = Path(__file__).resolve().parent / "item-ids.csv" IDS_META = Path(__file__).resolve().parent / "item-ids.meta.json" ITEM_FILE = Path(__file__).resolve().parents[1] / "modbus_points" / "wrps_item_df.qli" HARVEST_ALL = ( " Fix: get the ids from CI Server - do not infer them.\n" " 1. in CI View open WRPS_TagTest, which binds EVERY project item\n" " 2. link any one value by hand and save; CI View then resolves and\n" " writes the id of every connection in the file\n" " 3. python build_display.py --harvest \n" " That measures all 49 in one pass. Any display works, but only\n" " WRPS_TagTest covers the whole item list." ) REHARVEST = HARVEST_ALL _ids = {} _missing = set() def item_file_sha(): import hashlib if not ITEM_FILE.is_file(): return "" return hashlib.sha256(ITEM_FILE.read_bytes()).hexdigest() def item_nsids(): """{item name: NSID} from the item export - the axis ids are built on.""" import re if not ITEM_FILE.is_file(): return {} q = ITEM_FILE.read_text(encoding="utf-8") return {m.group(1): int(m.group(2)) for m in re.finditer(r'^"(AID\.WRPS\.[A-Z0-9_.]+)",(\d+),', q, re.M)} def item_reps(): import re if not ITEM_FILE.is_file(): return {} q = ITEM_FILE.read_text(encoding="utf-8") return dict(re.findall(r'^"(AID\.WRPS\.[A-Z0-9_.]+)".*?"(Real|Boolean)"', q, re.M | re.S)) def measured_ids(): import csv if not IDS_FILE.is_file(): return {} return {r["item"]: r["itemId"] for r in csv.DictReader(IDS_FILE.open(encoding="utf-8"))} def load_ids(): """The measured ids, checked for internal consistency. No derivation.""" import json import sys ids = measured_ids() if not item_nsids(): return if not ids: sys.exit("BUILD FAILED: no item ids have ever been measured.\n" " Every value would render as 0 on a screen that looks fine.\n" + REHARVEST) if IDS_META.is_file(): meta = json.loads(IDS_META.read_text(encoding="utf-8")) if meta.get("item_file_sha") != item_file_sha(): sys.exit("BUILD FAILED: %s has changed since the ids were harvested.\n" " Harvested against : %s\n" " Now : %s\n" " The items have been regenerated, so they will be (or have\n" " been) re-imported - and every import moves every id.\n" % (ITEM_FILE.name, meta.get("item_file_sha", "?")[:16], item_file_sha()[:16]) + REHARVEST) else: sys.exit("BUILD FAILED: %s is missing, so the ids cannot be shown to\n" " match the current items.\n" % IDS_META.name + REHARVEST) _ids.update(ids) def item_id(item): """The MEASURED id for an item. Nothing is derived. An earlier version filled gaps from `NSID + K`. The rule held every time it was checked, but it is still an inference about someone else's numbering, and an inferred id that is wrong produces a screen that looks alive and reads 0 - the exact failure this project keeps paying for. So: if CI Server has not told us an id, the build stops. Every item can be measured at once - see check_ids(). """ if not _ids: load_ids() if item not in _ids: # Recorded, not raised: check_ids() reports every missing item in # one list, and nothing is written to disk before it runs. _missing.add(item) return "UNMEASURED" return _ids[item] # ---------------------------------------------------------------- style # component-kit.md. Kept as (r,g,b) so they land straight in colorSet. C = { "BG": (237, 240, 244), "SF": (255, 255, 255), "SF2": (245, 248, 251), "BD": (195, 204, 216), "BD2": (223, 229, 236), "T1": (14, 22, 33), "T2": (70, 84, 104), "AC": (11, 125, 168), "AC2": (18, 165, 217), "RUN": (15, 138, 69), "RUNBG": (228, 246, 234), "WRN": (168, 91, 0), "WRNBG": (255, 239, 210), "ALM": (211, 32, 41), "ALMBG": (253, 232, 233), "ACBG": (222, 240, 248), "ALMBD": (150, 19, 26), "PRC": (43, 63, 82), "WTR": (159, 207, 230), "WHITE": (255, 255, 255), } LAB = "Segoe UI" VAL = "Consolas" # item namespace NS = "AID.WRPS" def text_width(value, size, font=LAB): """Rough rendered width of a string, in px. Only an estimate - the components keep their original size and CI View lays them out itself. It is shared rather than inlined because the ladder marks place a live number immediately left of a label, and if the two disagreed about how wide the label is they would drift apart. """ return len(value) * size * (0.62 if font == VAL else 0.55) # ------------------------------------------------------------- plumbing class Display: """Accumulates shared objects and components, then serialises.""" def __init__(self, title, width=W, height=H): self.title = title self.w, self.h = width, height self.refs = [] # (id, xml) self.ref_cache = {} self.parts = [] self.next_ref = 1000 self.next_html = 100 # --- shared objects ------------------------------------------------ def _ref(self, key, xml): if key in self.ref_cache: return self.ref_cache[key] rid = self.next_ref self.next_ref += 1 self.refs.append((rid, xml)) self.ref_cache[key] = rid return rid def paint(self, rgb): r, g, b = rgb return self._ref(("paint", rgb), '' '' '' % (r, g, b)) def conn(self, item, attribute="ItemValue", direction=1): """Write connection, as a shared object - the form the editor saves.""" return self._ref(("conn", item, attribute, direction), '' '%d' '%s' '%s' '%s' '' % (direction, item, item_id(item), attribute)) def read_conn(self, item): """Read connection, inline and without itemAttribute. This is exactly what CI View wrote when the item was linked by hand; the earlier version omitted itemId and every value read 0. """ return ('' "1" "%s" "%s" "" % (item, item_id(item))) def filt(self, value="0.0"): return self._ref(("filter", value), '%s' '' % value) def direct(self): return self._ref(("tdirect",), '' 'java.lang.Double' 'java.lang.Double' '') def named_colour(self, name): """A colour CI View names rather than mixes - `gainsboro` and friends. The rest of the kit gives colours as RGB; the microTrend background is the one place CI View wrote a name, so it is kept as one. """ return self._ref(("colorSet", name), '%s' "" % name) def comp_prop(self, name, value=None): """A componentProperty default, shared. Empty value stays empty.""" return self._ref(("cprop", name, value), '%s' "%s" % (name, "" if value is None else "%s" % value)) def plain_conn(self, item): """Read connection as a SHARED object, with no itemAttribute. `conn()` always writes an itemAttribute and `read_conn()` is inline; the microTrend needs the third shape - shared and bare - which is what CI View wrote for the trend's realTime/fillWithHistory bindings. """ return self._ref(("plainconn", item), '' "1" "%s%s" "" % (item, item_id(item))) def hid(self): self.next_html += 1 return self.next_html # --- state: making the screen show what the plant is doing ---------- # # All three of these were copied from the site's own displays, which # use them heavily - rectangle.fillPaint 276 times, rectangle.visible # 29, line.visible 25. The note in DEPLOY.md that said driving a # shape from a value needed item alarm limits was simply wrong, and # it is why this screen showed a grey pump whether it was running, # stopped or tripped. def parameter(self, name, kind="Boolean", default="false"): """A display parameter: the wire between a function and a shape.""" self.parts.append( '' '%s' '' "%s" % (name, kind, default)) return name def param_conn(self, name, direction=1): return ('%d' "%s%s" % (direction, name, "ItemValue" if direction == 2 else "")) def equals(self, item, value, pname): """item == value -> a boolean parameter, via genericFunctions. The only way to test an enum. STATE 2 means "Pumping" and the operator should read that word, not the number 2. """ self.parameter(pname) n = len([p for p in self.parts if "genericFunctions" in p]) gx, gy = 40 + (n % 30) * 62, 1068 + (n // 30) * 10 self.parts.append( '' "%s" "%.1f%.1f4.04.0" "28.028.0" "0.00.0" "%.1f" '' '' '' '' '%s' '' '' '%s' "" % (pname, gx, gy, float(value), self.filt(), self.read_conn(item), self.filt(), self.param_conn(pname, 2))) return pname def vis_action(self, kind, item=None, param=None): """The action alone, so it can share a data block with another one.""" conn = self.read_conn(item) if item else self.param_conn(param, 1) return ('' '' '%s' % (kind, self.filt(), conn)) def vis(self, kind, item=None, param=None): """Shared data object binding .visible to a bool item/param.""" return self._ref( ("vis", kind, item, param), '%s' % self.vis_action(kind, item, param)) def moves_y(self, kind, item, y_at_0, y_at_full, full=100.0): """Bind .y to an item, so a mark sits at the level it means. A setpoint that can be typed into and a line drawn at a fixed height are two claims about the same number, and only one of them updates. This binds the second to the first. The `transformationLinear` shape is the one CI View wrote when the user moved the STOP ALL mark by hand on 2026-08-17, including the absent `inputValueStart` - the editor omits it because it defaults to 0, and every scale here does start at 0. `x`/`y` are declared properties on Rectangle, Text and Number alike (checked in components/*.xml), so the line, its live number and its label can all ride the same item. All three carry `html="noConnect"` though: in CI Server's WEB client these bindings may not animate. The desktop CI View client is what this project targets. """ return ('' '' '' '' "java.lang.Double" "java.lang.Double" "%.1f" "%.1f" "%.1f" "%s" % (kind, self.filt(), full, y_at_0, y_at_full, self.read_conn(item))) def data(self, actions): """Register actions as a shared data object and return its ref. Actions in globalSection as , referenced with . This is a preference, not a requirement: WRPS_TagTest bound all 49 items both this way and inline as , and both columns updated identically. Shared is kept because it is what CI View writes on save and it de-duplicates the repeated actions. (An earlier note here claimed inline bindings never update and that this was why values read 0. That was wrong - see the mask note on number() for the actual cause.) """ return self._ref(("data", actions), '%s' % actions) def open_display(self, name, event="Clicked b1"): """Shared data object that opens a display AS A POPUP. ActivateSpecific + layout FixedPopup is what the site's own screens use; without it the display replaces the current one full-screen. """ return self._ref(("open", name, event), '' '' '' '' '%s' 'FixedPopup' 'default' '' % (event, name)) def hit_area(self, x, y, w, h, display): """Invisible click target: a fully transparent rectangle. Using a real button here drew a grey button face over the symbol underneath, which is exactly what happened on the first deploy. """ cx, cy = x + w / 2.0, y + h / 2.0 clear = self._ref(("paint", "clear"), '' '' "") s = ['', "%d" % self.hid(), "%.1f%.1f" % (cx, cy), "%.1f%.1f" % (h / 2.0, h / 2.0), "%.1f%.1f" % (w / 2.0, w / 2.0), "0.00.0", '', '' % clear, '' % clear, '' % self.open_display(display), ""] self.parts.append("".join(s)) # --- primitives ---------------------------------------------------- def rect(self, x, y, w, h, fill, stroke=None, width=1.0, show_item=None, show_param=None, move=None): """Rectangle by top-left + size; converted to centre/half-extent. show_item / show_param bind `visible`, so a shape can appear only when a pump is running, a state is active, or a mode is selected. `move` takes a moves_y() action and rides in the SAME data block - a component carries one, so the two cannot be separate. """ cx, cy = x + w / 2.0, y + h / 2.0 s = ['', "%d" % self.hid(), "%.1f%.1f" % (cx, cy), "%.1f%.1f" % (h / 2.0, h / 2.0), "%.1f%.1f" % (w / 2.0, w / 2.0), "0.00.0", '' % width, '' % self.paint(stroke or fill), '' % self.paint(fill)] acts = (self.vis_action("rectangle", show_item, show_param) if (show_item or show_param) else "") + (move or "") if acts: s.append('' % self.data(acts)) s.append("") self.parts.append("".join(s)) def line(self, x1, y1, x2, y2, colour, width=2.0, dash=False, move=None): """Orthogonal line drawn as a thin rectangle. The `line` component positions itself with curvePoints/curveSegments, whose anchor convention could not be determined unambiguously from the one example in the deployment. A filled rectangle is exact, needs no guessing, and is what the site's own screens use for rules and pipes. Diagonals are therefore not available - which is no loss, because a P&ID is drawn orthogonally anyway. """ if dash: # a dashed line is a run of rectangles, so there is no single one # to move; nothing needs a moving dashed line, so this is a bug assert move is None, "a dashed line cannot be moved by an item" # dashed instrument signal: a run of short rectangles if y1 == y2: x = min(x1, x2) while x < max(x1, x2): self.rect(x, y1 - width / 2.0, min(7.0, max(x1, x2) - x), width, colour) x += 12.0 else: y = min(y1, y2) while y < max(y1, y2): self.rect(x1 - width / 2.0, y, width, min(7.0, max(y1, y2) - y), colour) y += 12.0 return if y1 == y2: self.rect(min(x1, x2), y1 - width / 2.0, abs(x2 - x1), width, colour, move=move) elif x1 == x2: self.rect(x1 - width / 2.0, min(y1, y2), width, abs(y2 - y1), colour, move=move) else: assert move is None, "an L-shaped line is two rectangles; move one" # route orthogonally: across, then down self.line(x1, y1, x2, y1, colour, width) self.line(x2, y1, x2, y2, colour, width) def ellipse(self, cx, cy, r, fill, stroke, width=2.0, show_item=None, show_param=None): s = ['', "%d" % self.hid(), "%.1f%.1f" % (cx, cy), "%.1f%.1f" % (r, r), "%.1f%.1f" % (r, r), "0.00.0", '' % width, '' % self.paint(stroke), '' % self.paint(fill)] if show_item or show_param: s.append('' % self.vis("ellipse", show_item, show_param)) s.append("") self.parts.append("".join(s)) def text(self, x, y, value, size=14, colour=None, bold=False, font=LAB, anchor="left", show_item=None, show_param=None, move=None): colour = colour or C["T1"] # width estimate: enough for the editor to place it; it keeps original size tw = text_width(value, size, font) cx = x + tw / 2.0 if anchor == "left" else (x - tw / 2.0 if anchor == "right" else x) s = ['', "%d" % self.hid(), "%s" % esc(value), '' % (font, "true" if bold else "false", size), '' % self.paint(colour), '', "true", "%.1f%.1f" % (cx, y), "%.1f%.1f" % (size * 0.7, size * 0.7), "%.1f%.1f" % (tw / 2.0, tw / 2.0), "0.00.0"] acts = (self.vis_action("text", show_item, show_param) if (show_item or show_param) else "") + (move or "") if acts: s.append('' % self.data(acts)) s.append("") self.parts.append("".join(s)) def number(self, x, y, item, size=22, colour=None, bold=True, decimals=0, width=110, move=None): """A live value. This is the only way a number reaches the screen. is a DIGIT MASK, not a Java DecimalFormat pattern. The first build emitted "0" and "0.0" - masks one digit wide - so nothing wider than a single digit had anywhere to render, and every value read 0. Confirmed by CI View itself: setting the FIT-201 INLET value to two decimals by hand wrote 99.99, immediately after and with no . That element order is copied here. Masks are per point and live in point_format.py, sized to each point's engineering range. `decimals` is ignored - the mask decides - but is kept in the signature because callers pass it. """ colour = colour or C["T1"] cx = x + width / 2.0 s = ['', "%d" % self.hid(), "%s" % pf.mask(item), '' % (VAL, "true" if bold else "false", size), '' % self.paint(colour), '', "true", "%.1f%.1f" % (cx, y), "%.1f%.1f" % (size * 0.7, size * 0.7), "%.1f%.1f" % (width / 2.0, width / 2.0), "0.00.0", '' % self.data( '' '' '%s' % (self.filt(), self.read_conn(item)) + (move or "")), ""] self.parts.append("".join(s)) def number_field(self, x, y, w, h, item, size=14): """An EDITABLE value - the operator types into it and it writes. Copied from what CI View itself wrote when the user added a level setpoint field by hand on 2026-08-14. Two things differ from every other component here and both are load-bearing: - geometry is on the ELEMENT as attributes, not child elements - the connection carries ItemValue and NO ; a numberField is read-write by nature It also takes no : the field shows what the operator can type, and a mask would fight the editing. """ cx, cy = x + w / 2.0, y + h / 2.0 s = ['' % (cx, cy, h / 2.0, h / 2.0, w / 2.0, w / 2.0), "%d" % self.hid(), '' % (LAB, size), '' '' '' '' "%s%s" "ItemValue" "" % (self.filt(), item, item_id(item)), ""] self.parts.append("".join(s)) def data_bar(self, x, y, w, h, item, low=0.0, high=100.0, fill=None, back=None, up=True, opens=None): """A filled bar driven by an item - the live level in the well. The range lives on the COMPONENT (`lowLimit`/`highLimit`), not on the item, so this needs no alarm or scale limits configured in CI Server - which is what kept the well fill static until now. The site's own Straddle_Detail uses exactly this for a fuel level %, with highLimit 100. `opens` puts an actionActivateDisplay in the SAME data block as the value binding, so the bar is both the reading and the way in to the numbers behind it. A hit_area over the well would have done the job too, but this is the mechanism the user built by hand in CI View on 2026-08-17 and proved works, and an invisible rectangle laid over the well would sit on top of every level mark in it. """ cx, cy = x + w / 2.0, y + h / 2.0 s = ['' % (cx, cy, h / 2.0, h / 2.0, w / 2.0, w / 2.0), "%d" % self.hid(), "true", '' "" % (back or C["SF"]), '' "" % (fill or C["WTR"]), "%.1f%.1f" % (low, high), '' % ("Up" if up else "Down"), '' '' '%s%s' % (self.filt(), self.read_conn(item), '' '' '' "%s" "%sdefault" "" % (opens, "FixedPopup" if opens.startswith("WRPS_FP") else "default") if opens else ""), ""] self.parts.append("".join(s)) def micro_trend(self, x, y, w, h, item, scale_max): """A strip trend of one item, under or over the box that reads it. Transcribed from the two the user drew by hand in CI View on 2026-08-17, so the generated ones are the same component and not a near miss. Three things in here are CI View's shape, not a choice: - the item arrives through `microTrend.itemName[0]` with `ItemName` - the trend is told WHICH item to pull history for, so it wants the name, not the value. Every other component here binds ItemValue. - `realTime` and `fillWithHistory` are bound to the same item, as a bare shared connection with no itemAttribute. They read as booleans, which is odd for a flow in m3/h; it is what the editor wrote for a trend that works, so it is what is written here. - `actionInternalServiced` leads the data block and takes no body. `scaleMax` is the only scale set; the minimum stays at the component default of 0, which is right for a level in % and a flow in m3/h. """ cx, cy = x + w / 2.0, y + h / 2.0 src = self.plain_conn(item) bind = ('' '' '') s = ['' % (cx, cy, h / 2.0, h / 2.0, w / 2.0, w / 2.0), "%d" % self.hid(), '' % self.named_colour("gainsboro"), '' % self.comp_prop("itemName[0]"), 'scaleMax[0]' "%.1f" % scale_max, '', '', '' '' '' '1' "%s%s" "ItemName" % (self.filt(), item, item_id(item)), bind % ("realTime", self.filt(), src), bind % ("fillWithHistory", self.filt(), src), "", ""] self.parts.append("".join(s)) def button(self, x, y, w, h, label, item=None, write=None, display=None, size=15, close=False): """WRPS_Button: writes a value, opens a display, or both.""" cx, cy = x + w / 2.0, y + h / 2.0 acts = [] if item is not None and write is not None: acts.append('' '' '' % (self.filt(str(write)), self.conn(item, direction=2))) if close: # closes the popup it lives on, rather than activating another # display over the top of it acts.append('' '') if display: acts.append('' '' '' "%s" "%sdefault" "" % (display, "FixedPopup" if display.startswith("WRPS_FP") else "default")) # Element is "] self.parts.append("".join(s)) # --- kit components ------------------------------------------------- def panel(self, x, y, w, h, header): self.rect(x, y, w, h, C["SF"], C["BD"], 1.0) self.rect(x, y, w, 34, C["SF"], C["BD2"], 1.0) self.text(x + 12, y + 17, header, size=13, colour=C["T2"], font=VAL) def value_box(self, x, y, w, tag, item, units="", decimals=0, big=False, editable=False): """WRPS_Value. Every analog value on every display is this. `editable` swaps the read-only number for a numberField, so the box an operator sets a setpoint in looks like every other box on the screen rather than a foreign widget. """ h = 74 if big else 56 self.rect(x, y, w, h, C["SF2"], C["BD2"], 1.0) self.rect(x, y, 5, h, C["BD"], C["BD"], 1.0) # left rule self.text(x + 14, y + 15, tag, size=12, colour=C["T2"], font=VAL) if editable: self.number_field(x + 12, y + 24, w - 70, 26, item) else: self.number(x + 14, y + (46 if big else 37), item, size=30 if big else 21, decimals=decimals) if units: self.text(x + w - 12, y + (52 if big else 41), units, size=12, colour=C["T2"], font=VAL, anchor="right") return h def state_box(self, x, y, w, label, colour, bg): self.rect(x, y, w, 34, bg, colour, 1.5) self.rect(x + 9, y + 12, 10, 10, colour, colour, 1.0) self.text(x + 26, y + 17, label, size=14, colour=colour, bold=True, font=VAL) def instrument(self, cx, cy, letters, loop, r=30): self.ellipse(cx, cy, r, C["SF"], C["PRC"], 1.8) self.text(cx, cy - 9, letters, size=13, colour=C["T1"], bold=True, font=VAL, anchor="centre") self.text(cx, cy + 10, loop, size=13, colour=C["T1"], bold=True, font=VAL, anchor="centre") # --- output --------------------------------------------------------- def xml(self): when = datetime.datetime.now().strftime("%Y.%m.%d %H:%M:%S.000 AEST") refs = "".join('%s' % (i, x) for i, x in self.refs) vis = "".join( '%d%s' "%strue" "%s" % (i + 2, n, d, z) for i, (n, d, z) in enumerate([ ("Overview", "Always shown", "10.0"), ("Rough", "Shown when viewing a large area", "25.0"), ("Standard", "Shown when using the default view setting", "100.0"), ("Detail", "Shown only when viewing a small area", "400.0"), ("Intricacies", "Shown only when viewing a very small area", "1000.0"), ])) head = ( '\n' '\n' " %s\n" ' \n' ' \n' " %d%d\n" " 3\n" ' \n' ' \n' ' \n' ' \n' ' ' '\n' ' ' '\n' " 700true\n" " false\n" " %s\n" ' 1' "Layer1\n" " %d\n" ) % (refs, self.w, self.h, C["BG"][0], C["BG"][1], C["BG"][2], C["T1"][0], C["T1"][1], C["T1"][2], LAB, when, vis, len(self.parts)) return head + "\n".join(" " + p for p in self.parts) + \ "\n \n\n" def esc(s): return (s.replace("&", "&").replace("<", "<").replace(">", ">")) # ===================================================================== # WRPS_Overview - the single operator display # ===================================================================== PUMP_STATES = [(0, "OFF"), (1, "IDLE"), (2, "STARTING"), (3, "RUNNING"), (4, "STOPPING"), (5, "TRIPPED"), (6, "LOCKED OUT")] STN_STATES = [(0, "OFF", "T2"), (1, "IDLE", "T2"), (2, "PUMPING", "RUN"), (3, "HIGH LEVEL", "ALM"), (4, "EMERGENCY - LSHH", "ALM"), (5, "DRY RUN LOCKOUT", "ALM"), (6, "FAULT", "ALM")] # ladder marks: % of the spill weir, label, is it an alarm level, and the # setpoint behind it - None where the level is PHYSICAL and cannot be typed. # The spill weir is concrete and the two float switches are bolted to the # wall at a height; no operator moves those, so those three marks stay put # and stay a plain string. The other five are numbers someone can change, # and a mark drawn at a height that no longer matches its setpoint is a # screen telling a lie, so those five follow their item. LADDER = [(100, "SPILL WEIR", 1, None), (91.7, "LSHH-102", 1, None), (86.7, "HIGH ALARM", 1, ".SP.HIGH_ALARM"), (83.3, "START P3", 0, ".SP.START_P3"), (75, "START P2", 0, ".SP.START_P2"), (66.7, "START DUTY", 0, ".SP.START_DUTY"), (16.7, "STOP ALL", 0, ".SP.STOP_ALL"), (10, "LSLL-103", 0, None)] def station_header(d): """The title bar, shared by every full-size display. Lifted out of overview() when WRPS_Tank was added. It is the same bar on both screens by construction rather than by someone remembering to copy a change across, which is the failure this file keeps having. DUTY stays on the tank screen even though no pump is drawn there: it says which pump the levels are about to start, which is exactly what someone editing those levels wants to know. """ d.rect(0, 0, W, 76, C["SF"], C["BD"], 2.0) d.text(28, 40, "WATERLOO ROAD PUMP STATION", size=26, bold=True) d.text(520, 44, "WW-101", size=14, colour=C["T2"], font=VAL) d.text(1180, 26, "STATION", size=11, colour=C["T2"], font=VAL) for value, label, tone in STN_STATES: p = d.equals(NS + ".STN.STATE", value, "STATE_%d" % value) d.rect(1176, 38, 230, 30, C["SF2"], C["BD2"], 1.0, show_param=p) d.text(1186, 53, label, size=15, bold=True, colour=C[tone], font=VAL, show_param=p) d.text(1430, 26, "MODE", size=11, colour=C["T2"], font=VAL) for value, label, tone in [(1, "AUTO", "RUN"), (2, "OFF", "T2")]: p = d.equals(NS + ".SP.MODE", value, "MODE_%d" % value) d.rect(1426, 38, 120, 30, C["SF2"], C["BD2"], 1.0, show_param=p) d.text(1436, 53, label, size=15, bold=True, colour=C[tone], font=VAL, show_param=p) d.text(1570, 26, "DUTY", size=11, colour=C["T2"], font=VAL) d.rect(1566, 38, 130, 30, C["SF2"], C["BD2"], 1.0) for n in (1, 2, 3): p = d.equals(NS + ".STN.DUTY_PUMP", n, "DUTY_%d" % n) d.text(1576, 53, "PU-30%d" % n, size=15, bold=True, font=VAL, show_param=p) d.text(1576, 53, "NONE", size=15, colour=C["T2"], font=VAL, show_param=d.equals(NS + ".STN.DUTY_PUMP", 0, "DUTY_0")) # exceptions only: an operator should see these, not hunt for a 0 for item, label in [(".STN.HIGH_LEVEL", "HIGH LEVEL"), (".STN.SPILL_ACTIVE", "SPILLING")]: d.rect(1726, 38, 170, 30, C["ALMBG"], C["ALM"], 1.5, show_item=NS + item) d.text(1736, 53, label, size=13, bold=True, colour=C["ALM"], font=VAL, show_item=NS + item) def draw_well(d, WX, WY, WW, WH, opens=None): """The wet well and its level ladder - the one drawing both screens share. WRPS_Tank is the overview's well with the plant either side of it taken away, so the well is drawn from here on both. Duplicating it would mean the marks agreeing today and drifting the first time a percentage moves; the ladder is the CONTRACT between what the PLC does and what the screen claims, and it can only be in one place. `opens` makes the water itself the click target for another display. """ PER = WH / 100.0 PER = WH / 100.0 d.rect(WX, WY, WW, WH, C["SF"], C["PRC"], 3.0) # the well is the way in to the setpoints: there is no SETPOINTS button, # and clicking the thing the levels are ABOUT is a better target than one d.data_bar(WX + 3, WY + 3, WW - 6, WH - 6, NS + ".STN.LEVEL", 0.0, 100.0, opens=opens) d.text(WX, WY - 14, "WET WELL WW-101 . 120 m2", size=13, bold=True) # 0% is the well floor at WY+WH, 100% the spill weir at WY - the two ends # the marks are interpolated between, and what a moving mark rides on Y0, Y100 = WY + WH, WY LBLY = 13 # label sits this far under its own line for pct, label, alarm, item in LADDER: y = WY + WH - pct * PER tone = C["ALM"] if alarm else C["T2"] if item is None: # physical: concrete and float switches # LEFT wall, while everything adjustable is on the RIGHT. An # adjustable mark can be driven to any height, so sooner or later # it lands on a fixed one; the two kinds therefore never share a # column. Outside the well was the other option and is not # available - the FIT-201 trend runs straight across the 100% line. d.line(WX, y, WX + WW, y, C["BD"], 1.0) d.text(WX + 8, y + LBLY, "%g%% %s" % (pct, label), size=11, colour=tone, font=VAL) continue # adjustable: line, live number and label all ride the same setpoint. # The label keeps the right edge it has always had; the number sits # immediately left of it, so the reading grows leftwards and the # column stays flush against the wall of the well. it = NS + item tail = "%% %s" % label # "% START DUTY" - the number supplies 86.7 numw = text_width("999.9", 11, VAL) numx = WX + WW - 8 - text_width(tail, 11, VAL) - numw d.line(WX, y, WX + WW, y, C["BD"], 1.0, move=d.moves_y("rectangle", it, Y0, Y100)) d.number(numx, y + LBLY, it, size=11, colour=tone, bold=False, width=numw, move=d.moves_y("number", it, Y0 + LBLY, Y100 + LBLY)) d.text(WX + WW - 8, y + LBLY, tail, size=11, colour=tone, font=VAL, anchor="right", move=d.moves_y("text", it, Y0 + LBLY, Y100 + LBLY)) # 34 below the floor, not 22: a mark driven to 0% puts its label at the # floor + 13, and at 22 the two interleave. Found by the sweep in # check_layout.py, which is the only thing that can see it. d.text(WX, WY + WH + 34, "0 - 6.00 m . 100% = SPILL WEIR", size=11, colour=C["T2"], font=VAL) def overview(): """The operator display. Laid out against a rendered preview rather than in a diff, one region at a time. The rules that came out of that: - a reading belongs beside the instrument that produces it, not in a rail on the far side of the screen - nothing sits on a pipe, and the inlet main runs INTO the well rather than stopping at the wall - the ladder is computed from the percentages, so 10% is at a tenth of the height, and every label sits the same distance below its own line - state is shown as a word and a colour, never as an enum number """ d = Display("WRPS Overview") station_header(d) # ---- process -------------------------------------------------------- d.panel(24, 96, 1872, 812, "PROCESS") # inlet, with FIT-201 and its reading clear of everything else d.instrument(88, 272, "FIT", "201", r=27) d.line(88, 299, 88, 390, C["T2"], 1.6, True) d.value_box(140, 240, 250, "FIT-201 INLET FLOW", NS + ".STN.INFLOW", "m3/h", 1) d.micro_trend(140, 306, 250, 56, NS + ".STN.INFLOW", 1100) d.line(60, 390, 470, 390, C["PRC"], 3.0) # runs INTO the well d.text(140, 418, "DN600 INLET GRAVITY MAIN", size=12, colour=C["T2"]) WX, WY, WW, WH = 440, 330, 300, 500 draw_well(d, WX, WY, WW, WH, opens="WRPS_Tank") # LIT-101 beside its own readings; leader drops and runs in underneath d.instrument(88, 488, "LIT", "101", r=27) d.line(88, 515, 88, 668, C["T2"], 1.6, True) d.line(88, 668, WX, 668, C["T2"], 1.6, True) d.value_box(140, 448, 280, "LIT-101 WET WELL LEVEL", NS + ".STN.LEVEL", "%", 1, big=True) d.micro_trend(140, 528, 280, 56, NS + ".STN.LEVEL", 100) # 598, not 564: the level trend above it needs the 34 px d.value_box(140, 598, 280, "LEVEL SETPOINT", NS + ".SP.LEVEL_SP", "%", 1, editable=True) # overflow: plain text, not a bordered box - a box reads as a button d.line(WX + WW, 344, WX + WW + 70, 344, C["PRC"], 3.0) d.text(818, 349, "OVERFLOW TO ERS", size=11, colour=C["T2"], font=VAL) # pumps, left of the header so no reading sits on a pipe CXP, R, BX, BW, HDR = 900, 34, 980, 210, 1300 for i, cy in enumerate((450, 590, 730)): n = i + 1 run = "%s.PU30%d.RUNNING" % (NS, n) trip = "%s.PU30%d.TRIPPED" % (NS, n) d.line(WX + WW, cy, CXP - R, cy, C["PRC"], 3.0) d.line(CXP + R, cy, HDR, cy, C["PRC"], 3.0) d.ellipse(CXP, cy, R, C["SF2"], C["PRC"], 2.4) d.ellipse(CXP, cy, R, C["RUNBG"], C["RUN"], 3.0, show_item=run) d.ellipse(CXP, cy, R, C["ALMBG"], C["ALM"], 3.0, show_item=trip) d.line(CXP - R, cy, CXP + R, cy, C["PRC"], 1.2) d.line(CXP, cy - R, CXP, cy + R, C["PRC"], 1.2) d.text(CXP - 32, cy + 58, "PU-30%d" % n, size=14, bold=True, font=VAL) d.text(CXP - 32, cy + 76, "DUTY", size=11, bold=True, colour=C["AC"], font=VAL, show_param=d.equals(NS + ".STN.DUTY_PUMP", n, "DUTYP_%d" % n)) d.rect(BX, cy - 64, BW, 52, C["SF2"], C["BD2"], 1.0) d.rect(BX, cy - 64, 5, 52, C["RUN"], C["RUN"], 1.0) d.text(BX + 14, cy - 46, "PU-30%d" % n, size=11, colour=C["T2"], font=VAL) for value, label in PUMP_STATES: # one word, never an enum p = d.equals("%s.PU30%d.STATE" % (NS, n), value, "P%dSTATE_%d" % (n, value)) tone = C["RUN"] if value == 3 else (C["ALM"] if value in (5, 6) else C["T1"]) d.text(BX + 14, cy - 24, label, size=14, bold=True, colour=tone, font=VAL, show_param=p) d.number(BX + BW - 70, cy - 30, "%s.PU30%d.RUN_HOURS" % (NS, n), size=12, width=44) d.text(BX + BW - 10, cy - 24, "h", size=11, colour=C["T2"], font=VAL, anchor="right") d.hit_area(CXP - R - 2, cy - R - 2, 2 * R + 4, 2 * R + 4, "WRPS_FP_PU30%d" % n) # speed and discharge trends sit ABOVE their boxes: below the speed box is # the dashed leader down to the pumps, and below the discharge box is the # rising main at y=590. Nothing else has to move to make room. d.micro_trend(BX, 170, BW, 56, NS + ".STN.SPEED", 100) d.value_box(BX, 240, BW, "VSD-301/2/3 SPEED", NS + ".STN.SPEED", "%", 1) d.line(1085, 304, 1085, 386, C["T2"], 1.6, True) # discharge header and FIT-301 d.line(HDR, 450, HDR, 730, C["PRC"], 3.0) d.line(HDR, 590, 1820, 590, C["PRC"], 3.0) d.text(1560, 614, "RISING MAIN . DN400", size=12, colour=C["T2"]) d.instrument(1500, 512, "FIT", "301", r=27) d.line(1500, 539, 1500, 590, C["T2"], 1.6, True) d.micro_trend(1542, 418, 240, 56, NS + ".STN.DISCHARGE", 1400) d.value_box(1542, 480, 240, "FIT-301 DISCHARGE", NS + ".STN.DISCHARGE", "m3/h", 1) # ---- simulation band ------------------------------------------------ SY = 928 d.rect(0, SY, W, H - SY, C["SF"], C["BD"], 1.0) d.rect(0, SY, W, 4, C["AC"], C["AC"], 1.0) d.text(28, SY + 30, "SIMULATION", size=14, bold=True, colour=C["AC"], font=VAL) d.text(28, SY + 52, "NOT PLANT CONTROL", size=11, colour=C["AC"], font=VAL) d.text(28, SY + 78, "ACTIVE:", size=11, colour=C["T2"], font=VAL) SCEN = [(0, "MANUAL", 206, ["inflow set by the", "box beside it"]), (1, "DIURNAL", 612, ["dry weather day,", "144-396 m3/h over 24 h"]), (2, "WET WEATHER", 770, ["storm: ramps to", "1080 m3/h, then decays"]), (3, "DEMO REF", 928, ["held at 594 m3/h,", "starts at 4.00 m"])] d.text(206, SY + 24, "SCENARIO - inflow the wet well sees", size=11, colour=C["T2"], font=VAL) for value, label, x, desc in SCEN: p = d.equals(NS + ".SIM.SCENARIO", value, "SCEN_%d" % value) d.rect(x - 4, SY + 34, 154, 54, C["ACBG"], C["AC"], 2.5, show_param=p) d.button(x, SY + 38, 146, 46, label, NS + ".SIM.SCENARIO", value, size=13) for k, line in enumerate(desc): d.text(x, SY + 104 + k * 13, line, size=10, colour=C["T2"], font=VAL) d.text(88, SY + 78, label, size=11, bold=True, colour=C["AC"], font=VAL, show_param=p) # the manual inflow lives next to MANUAL: nothing else uses it d.value_box(366, SY + 32, 230, "SIM INFLOW", NS + ".SIM.INFLOW", "m3/h", 1, editable=True) d.text(366, SY + 104, "type an inflow, used", size=10, colour=C["T2"], font=VAL) d.text(366, SY + 117, "by MANUAL only", size=10, colour=C["T2"], font=VAL) d.text(1124, SY + 24, "TIME SCALE - simulated clock", size=11, colour=C["T2"], font=VAL) for i, (s, desc) in enumerate([(1, "real time"), (10, "10x clock"), (30, "30x clock"), (60, "60x clock")]): x = 1124 + i * 96 p = d.equals(NS + ".SIM.TIME_SCALE", s, "TS_%d" % s) d.rect(x - 4, SY + 34, 92, 54, C["ACBG"], C["AC"], 2.5, show_param=p) d.button(x, SY + 38, 84, 46, "%dx" % s, NS + ".SIM.TIME_SCALE", s, size=13) d.text(x, SY + 104, desc, size=10, colour=C["T2"], font=VAL) d.button(1528, SY + 38, 180, 46, "RESET SCENARIO", NS + ".SIM.RESET", 1, size=13) d.text(1528, SY + 104, "restart the sim clock,", size=10, colour=C["T2"], font=VAL) d.text(1528, SY + 117, "level back to 3.50 m", size=10, colour=C["T2"], font=VAL) return d def tank(): """WRPS_Tank - the well at full size, with the numbers that shape it. The overview screen with the plant either side of the well taken away: no inlet, no rising main, no pumps. What replaces them is the setpoint faceplate's own idea, at the size it deserves - the levels are marks on the well and the entries sit beside it, so the number being typed and the height it means are one glance apart. The faceplate draws a small well of its own to carry those marks. Here that would be the same ladder twice at two scales, so the panel is rows only and the marks are the real well's. """ d = Display("WRPS Tank") station_header(d) # ---- the well ------------------------------------------------------- d.panel(24, 96, 780, 812, "WET WELL") # no `opens`: this IS the screen the overview's well opens WX, WY, WW, WH = 440, 330, 300, 500 draw_well(d, WX, WY, WW, WH) # LIT-101 beside its own readings; leader drops and runs in underneath d.instrument(88, 488, "LIT", "101", r=27) d.line(88, 515, 88, 668, C["T2"], 1.6, True) d.line(88, 668, WX, 668, C["T2"], 1.6, True) d.value_box(140, 448, 280, "LIT-101 WET WELL LEVEL", NS + ".STN.LEVEL", "%", 1, big=True) d.micro_trend(140, 528, 280, 56, NS + ".STN.LEVEL", 100) # the overview's LEVEL SETPOINT box is deliberately NOT repeated here: # it is a row in the panel opposite, and two entry fields writing one # item on one screen is an invitation to wonder which one is real d.button(48, 836, 200, 48, "OVERVIEW", display="WRPS_Overview") # ---- the setpoints -------------------------------------------------- PX = 828 d.panel(PX, 96, 1068, 812, "STATION SETPOINTS") d.text(PX + 40, 152, "levels in % of the 6.00 m spill weir . type a value to change it", size=12, colour=C["T2"], font=VAL) LX, FX, UX = PX + 40, PX + 652, PX + 808 LEVELS = [(".SP.HIGH_ALARM", "HIGH LEVEL ALARM", "ALM", "annunciates only - it starts no pump"), (".SP.START_P3", "START PUMP 3", "T1", "the third pump joins duty and lag at this level"), (".SP.START_P2", "START PUMP 2", "T1", "the second pump joins the duty pump at this level"), (".SP.LEVEL_SP", "LEVEL CONTROL SETPOINT", "AC", "the level the drive holds while pumping"), (".SP.START_DUTY", "START DUTY", "T1", "the duty pump starts at this level"), (".SP.STOP_ALL", "STOP ALL", "T1", "every pump stops at this level")] for i, (item, label, tone, why) in enumerate(LEVELS): ry = 186 + i * 68 d.text(LX, ry, label, size=15, bold=True, colour=C[tone]) d.text(LX, ry + 20, why, size=11, colour=C["T2"]) d.number_field(FX, ry - 16, 140, 32, NS + item) d.text(UX, ry, "%", size=12, colour=C["T2"], font=VAL) d.line(LX, 582, PX + 1028, 582, C["BD2"], 1.0) d.text(LX, 608, "NOT A LEVEL", size=11, colour=C["T2"], font=VAL) for i, (item, label, unit, why) in enumerate( [(".SP.MIN_SPEED", "MINIMUM DRIVE SPEED", "%", "38 Hz floor - below it the 22 m static lift gives no delivery"), (".SP.SERVICE_HRS", "SERVICE INTERVAL", "h", "run hours between services, counted per pump")]): ry = 648 + i * 68 d.text(LX, ry, label, size=15, bold=True) d.text(LX, ry + 20, why, size=11, colour=C["T2"]) d.number_field(FX, ry - 16, 140, 32, NS + item) d.text(UX, ry, unit, size=12, colour=C["T2"], font=VAL) d.text(LX, 792, "a start level above 100% - the spill weir - is rejected by the PLC", size=11, colour=C["T2"], font=VAL) d.text(LX, 814, "writes go straight to the PLC, which clamps and rejects bad values", size=11, colour=C["T2"], font=VAL) return d def fp_pump(n): """One faceplate per pump. Generated three times rather than once, because a single display cannot name the pump that opened it without binding an item through a display parameter - a mechanism the AOG library uses but which I have not verified. Three files cost nothing and cannot show PU-301's data under PU-303's heading, which is what the old single faceplate did. """ tag = "PU-30%d" % n P = "%s.PU30%d" % (NS, n) d = Display("WRPS %s Faceplate" % tag, 760, 700) d.rect(0, 0, 760, 700, C["SF"], C["BD"], 1.0) d.rect(0, 70, 760, 4, C["AC"], C["AC"], 1.0) d.text(24, 40, tag, size=26, bold=True) d.text(150, 44, "TRANSFER PUMP %d . 75 kW . 120 L/s @ 32 m" % n, size=12, colour=C["T2"], font=VAL) d.rect(560, 20, 80, 30, C["SF2"], C["BD2"], 1.0, show_param=d.equals(NS + ".STN.DUTY_PUMP", n, "ISDUTY_%d" % n)) d.text(570, 35, "DUTY", size=12, bold=True, colour=C["AC"], font=VAL, show_param="ISDUTY_%d" % n) d.rect(652, 20, 88, 30, C["RUNBG"], C["RUN"], 1.5, show_item=P + ".AVAILABLE") d.text(662, 35, "AVAILABLE", size=12, bold=True, colour=C["RUN"], font=VAL, show_item=P + ".AVAILABLE") # state: the reason the faceplate was opened d.rect(24, 96, 712, 78, C["SF2"], C["BD2"], 1.5) d.rect(24, 96, 712, 78, C["RUNBG"], C["RUN"], 2.0, show_item=P + ".RUNNING") d.rect(24, 96, 712, 78, C["ALMBG"], C["ALM"], 2.0, show_item=P + ".TRIPPED") for value, label in PUMP_STATES: p = d.equals(P + ".STATE", value, "FP%dSTATE_%d" % (n, value)) tone = C["RUN"] if value == 3 else (C["ALM"] if value in (5, 6) else C["T1"]) d.text(56, 130, label, size=26, bold=True, colour=tone, font=VAL, show_param=p) d.text(56, 156, "pump state from the PLC", size=12, colour=C["T2"], font=VAL) d.text(716, 122, "RUN HOURS", size=11, colour=C["T2"], font=VAL, anchor="right") d.number(600, 152, P + ".RUN_HOURS", size=24, width=116) d.text(24, 200, "THIS PUMP", size=11, colour=C["T2"], font=VAL) d.rect(24, 210, 228, 58, C["SF2"], C["BD2"], 1.0) d.text(38, 229, "RUN COMMAND", size=11, colour=C["T2"], font=VAL) d.text(38, 253, "ON", size=16, bold=True, colour=C["RUN"], font=VAL, show_item=P + ".RUN_CMD") d.rect(266, 210, 228, 58, C["SF2"], C["BD2"], 1.0) d.text(280, 229, "TRIPPED", size=11, colour=C["T2"], font=VAL) d.text(280, 253, "TRIPPED", size=16, bold=True, colour=C["ALM"], font=VAL, show_item=P + ".TRIPPED") d.value_box(508, 210, 228, "DRIVE SPEED . common", NS + ".STN.SPEED", "%", 1) d.text(24, 298, "STATION", size=11, colour=C["T2"], font=VAL) d.rect(24, 308, 228, 52, C["SF2"], C["BD2"], 1.0) d.text(38, 327, "STATION STATE", size=11, colour=C["T2"], font=VAL) for value, label, tone in STN_STATES: d.text(38, 350, label, size=14, bold=True, colour=C[tone], font=VAL, show_param="STATE_%d" % value if False else d.equals(NS + ".STN.STATE", value, "FP%dSTN_%d" % (n, value))) d.rect(266, 308, 228, 52, C["SF2"], C["BD2"], 1.0) d.text(280, 327, "DUTY PUMP", size=11, colour=C["T2"], font=VAL) for k in (1, 2, 3): d.text(280, 350, "PU-30%d" % k, size=14, bold=True, font=VAL, show_param=d.equals(NS + ".STN.DUTY_PUMP", k, "FP%dDUTY_%d" % (n, k))) d.value_box(508, 308, 228, "PUMPS RUNNING", NS + ".STN.PUMPS_RUNNING", "of 3", 0) # manual control only exists when the station is OFF, so say so d.text(24, 400, "MANUAL CONTROL", size=11, colour=C["T2"], font=VAL) auto = d.equals(NS + ".SP.MODE", 1, "FP%dAUTO" % n) off = d.equals(NS + ".SP.MODE", 2, "FP%dOFF" % n) d.rect(24, 410, 712, 72, C["WRNBG"], C["WRN"], 1.5, show_param=auto) d.text(44, 438, "STATION IS IN AUTO - manual commands do nothing", size=14, bold=True, colour=C["WRN"], font=VAL, show_param=auto) d.text(44, 462, "Put the station in OFF to start or stop a pump by hand.", size=12, colour=C["WRN"], font=VAL, show_param=auto) d.rect(24, 410, 348, 72, C["SF"], C["BD2"], 1.0, show_param=off) d.rect(388, 410, 348, 72, C["SF"], C["BD2"], 1.0, show_param=off) d.button(36, 422, 324, 48, "START", NS + ".SP.CMD_WORD", 2) d.button(400, 422, 324, 48, "STOP", NS + ".SP.CMD_WORD", 3) d.text(24, 514, "MAINTENANCE", size=11, colour=C["T2"], font=VAL) d.button(24, 524, 228, 50, "LOCK OUT", NS + ".SP.CMD_WORD", 4) d.button(266, 524, 228, 50, "RESET TRIP", NS + ".SP.CMD_WORD", 1) d.button(508, 524, 228, 50, "RESET RUN HOURS", NS + ".SP.CMD_WORD", 5) d.text(24, 592, "lock out and trip reset apply to %s . reset hours clears all three" % tag, size=11, colour=C["T2"], font=VAL) d.text(24, 630, "LAST COMMAND ACKNOWLEDGED", size=11, colour=C["T2"], font=VAL) d.number(24, 660, NS + ".STN.CMD_ACK", size=16, width=60) d.button(596, 632, 140, 46, "CLOSE", close=True) return d def fp_setpoints(): """Setpoints drawn as levels, because that is what they are. Eight identical boxes hid the one thing that matters about these numbers: they are a stack of thresholds that has to stay in order. Each one is now a mark on the well at its own height, against the live level, with its entry field on the same line. """ d = Display("WRPS Setpoints", 900, 700) d.rect(0, 0, 900, 700, C["SF"], C["BD"], 1.0) d.rect(0, 70, 900, 4, C["AC"], C["AC"], 1.0) d.text(24, 40, "STATION SETPOINTS", size=22, bold=True) d.text(290, 44, "levels in % of the 6.00 m spill weir . type a value to change it", size=12, colour=C["T2"], font=VAL) LX, LY, LW, LH = 60, 120, 120, 440 PER = LH / 100.0 d.text(24, 104, "WET WELL", size=11, colour=C["T2"], font=VAL) d.rect(LX, LY, LW, LH, C["SF"], C["PRC"], 2.5) d.data_bar(LX + 2, LY + 2, LW - 4, LH - 4, NS + ".STN.LEVEL", 0.0, 100.0) # the spill weir is the one level on this screen nobody can type, so it is # the one tick that stays put - and it moves to the RIGHT of the well, out # of the left column the six adjustable ticks slide up and down d.line(LX - 14, LY, LX + LW + 14, LY, C["ALM"], 1.5) d.text(LX + LW + 20, LY + 4, "100 SPILL WEIR", size=10, colour=C["ALM"], font=VAL) # item, label, its live level for the mark, row y, tone ROWS = [(".SP.HIGH_ALARM", "HIGH LEVEL ALARM", 86.7, 160, "ALM"), (".SP.START_P3", "START PUMP 3", 83.3, 206, "T1"), (".SP.START_P2", "START PUMP 2", 75.0, 252, "T1"), (".SP.LEVEL_SP", "LEVEL CONTROL SETPOINT", 70.0, 298, "AC"), (".SP.START_DUTY", "START DUTY", 66.7, 344, "T1"), (".SP.STOP_ALL", "STOP ALL", 16.7, 390, "T1")] # every mark here is adjustable - the fixed 100 above is the spill weir, # and it is the only level on this screen nobody can type Y0, Y100 = LY + LH, LY TICKY = 4 # the tick number's offset from its line numw = text_width("999.9", 10, VAL) for item, label, pct, ry, tone in ROWS: my = LY + LH - pct * PER it = NS + item colour = C[tone] if tone != "T1" else C["T2"] # solid, not dashed: a dashed line here is 30 small rectangles and # the dash convention belongs to instrument signals, not to a leader d.line(LX - 14, my, 420, my, colour, 1.0, move=d.moves_y("rectangle", it, Y0, Y100)) # the tick was a hard-coded "86.7"; it is the setpoint, so it reads it d.number(LX - 20 - numw, my + TICKY, it, size=10, colour=colour, bold=False, width=numw, move=d.moves_y("number", it, Y0 + TICKY, Y100 + TICKY)) # the ROW does not move - the entry fields are a list to work down, # and a list that reorders itself as you type in it is unusable d.text(440, ry + 16, label, size=13, bold=True, colour=C[tone]) d.number_field(700, ry, 110, 28, it) d.text(826, ry + 20, "%", size=11, colour=C["T2"], font=VAL) d.text(440, 440, "a start level above 100% - the spill weir - is rejected by the PLC", size=11, colour=C["T2"], font=VAL) d.line(440, 464, 860, 464, C["BD2"], 1.0) d.text(440, 492, "NOT A LEVEL", size=11, colour=C["T2"], font=VAL) d.text(440, 524, "MINIMUM DRIVE SPEED", size=13, bold=True) d.number_field(700, 508, 110, 28, NS + ".SP.MIN_SPEED") d.text(826, 528, "%", size=11, colour=C["T2"], font=VAL) d.text(440, 552, "38 Hz floor - below it the 22 m static lift gives no delivery", size=11, colour=C["T2"], font=VAL) d.text(440, 588, "SERVICE INTERVAL", size=13, bold=True) d.number_field(700, 572, 110, 28, NS + ".SP.SERVICE_HRS") d.text(826, 592, "h", size=11, colour=C["T2"], font=VAL) d.text(24, 661, "writes go straight to the PLC, which clamps and rejects bad values", size=11, colour=C["T2"], font=VAL) d.button(700, 632, 140, 46, "CLOSE", close=True) return d ITEM_FILE = Path(__file__).resolve().parents[1] / "modbus_points" / "wrps_item_df.qli" def known_items(): """Every item the CI Server import actually creates.""" import re if not ITEM_FILE.is_file(): return None return set(re.findall(r'"(AID\.WRPS\.[A-Z0-9_.]+)"', ITEM_FILE.read_text(encoding="utf-8"))) def report_unplaced(displays): """Say plainly what is NOT on any screen. Kept honest by computing it from the built XML against the item export, not from a list someone maintains by hand - a hand list is exactly the thing that goes stale and lets a point vanish quietly. """ import re bound = set() for _, d in displays: bound |= set(re.findall(r"(AID\.WRPS\.[A-Z0-9_.]+)", d.xml())) everything = set(item_nsids()) missing = sorted(everything - bound) print("") print("items on a display : %d of %d" % (len(bound & everything), len(everything))) if missing: print("NOT DISPLAYED ANYWHERE (%d):" % len(missing)) for m in missing: print(" %-22s %s" % (m.replace("AID.WRPS.", ""), UNPLACED_WHY.get(m, ""))) if UNPLACED_BUTTONS: print("BUTTONS NOT PLACED (%d):" % len(UNPLACED_BUTTONS)) for b in UNPLACED_BUTTONS: print(" %s" % b) # Why each one is still off the screen, so the list is a decision record # rather than a shrug. Emptied as they are placed. UNPLACED_WHY = { "AID.WRPS.STN.NET_ACCUM": "inflow minus discharge - no instrument to sit beside", "AID.WRPS.STN.VOL_TO_SPILL": "derived from level - awaiting a home", "AID.WRPS.STN.TIME_TO_SPILL": "derived from level - awaiting a home", "AID.WRPS.STN.TIME_TO_LSHH": "derived from level - awaiting a home", "AID.WRPS.STN.ALARM_WORD": "bitmask; needs decoding into named alarms, not a number", "AID.WRPS.STN.IN_AUTO": "duplicated by SP.MODE, which the header already shows", "AID.WRPS.SP.CMD_PARAM": "command plumbing - written with the command word, not read", } UNPLACED_BUTTONS = [ "AUTO / OFF - station mode, no device to sit beside", "RESET TRIPS - station wide", "RESET RUN HOURS - station wide (also on each pump faceplate)", "SETPOINTS... - no button; the WELL opens WRPS_Tank, which carries them", ] def check_ids(displays): """Every item a display binds must have a MEASURED id. All at once. Reported as one list rather than failing on the first, because the remedy is a single harvest that fixes all of them - being told about them one build at a time would be its own small hell. """ import re import sys if not _ids: load_ids() used = set() for _, d in displays: used |= set(re.findall(r"([^<]+)", d.xml())) missing = sorted(set(u for u in used if u not in _ids) | _missing) if missing: msg = ["BUILD FAILED: %d of %d bound items have no measured id." % (len(missing), len(used)), " These would render as 0 on a screen that looks healthy:"] msg += [" " + m.replace("AID.WRPS.", "") for m in missing] msg.append(HARVEST_ALL) sys.exit(chr(10).join(msg)) print("ids %d/%d bound items measured" % (len(used), len(used))) def check_items(displays): """Fail loudly if a display binds to an item that will not exist. Cheap to check, and the failure mode otherwise is a display full of silently dead values that looks fine in the editor. """ import re have = known_items() if have is None: print("WARN %s not found - item names not checked" % ITEM_FILE.name) return bad = {} for name, d in displays: used = set(re.findall(r"([^<]+)", d.xml())) missing = sorted(used - have) if missing: bad[name] = missing if bad: import sys msg = ["BUILD FAILED: displays bind to items that do not exist."] for name, items in bad.items(): msg.append(" %s:" % name) msg += [" " + i for i in items] msg.append(" Known items come from %s" % ITEM_FILE.name) sys.exit(chr(10).join(msg)) def harvest(path): """Read itemName/itemId pairs out of a display CI View has saved. Ids from an earlier generation are DROPPED, not merged. A re-import renumbers every item, so a file holding 24 fresh ids and 25 stale ones is worse than one holding 24 - the stale ones bind silently to nothing, and the derived-base check cannot tell which generation is current. Anything the fresh harvest disagrees with goes. """ import csv import re import sys txt = pathlib.Path(path).read_text(encoding="utf-8", errors="replace") found = dict(re.findall(r"([^<]+)\s*([^<]+)", txt)) if not found: sys.exit("no itemId pairs in %s - has CI View saved it?" % path) have = {} if IDS_FILE.is_file(): have = {r["item"]: r["itemId"] for r in csv.DictReader(IDS_FILE.open(encoding="utf-8"))} changed = {k: v for k, v in found.items() if have.get(k) != v} have.update(found) with IDS_FILE.open("w", newline="", encoding="utf-8") as fh: w = csv.writer(fh); w.writerow(["item", "itemId"]) for k in sorted(have): w.writerow([k, have[k]]) import datetime import json IDS_META.write_text(json.dumps({ "harvested_from": str(path), "harvested_at": datetime.datetime.now().isoformat(timespec="seconds"), "item_file": ITEM_FILE.name, "item_file_sha": item_file_sha(), "measured": len(have), "note": "Every .qli item import renumbers every item. Validate the " "displays in CI Server's Editor Module, then re-harvest - " "otherwise the screens go dead.", }, indent=2) + "\n", encoding="utf-8") print("harvested %d ids from %s (%d new or changed) -> %s" % (len(found), path, len(changed), IDS_FILE.name)) print(" provenance -> %s" % IDS_META.name) def verify(path): """Check a CI View-saved display against the ids this build would emit. This is the check that makes a dead screen impossible to miss. CI View resolves every connection by NAME when it saves, so the ids in a saved file are CI Server's own truth. If they match what we generate, the screens are live; if they do not, they are dead, and this says which items and by how much - without anyone having to look at a display and judge whether a 0 is a real 0. """ import re import sys txt = pathlib.Path(path).read_text(encoding="utf-8", errors="replace") found = dict(re.findall(r"([^<]+)\s*([^<]+)", txt)) if not found: sys.exit("no itemId pairs in %s - has CI View saved it?" % path) load_ids() bad = {n: (v, _ids.get(n)) for n, v in found.items() if _ids.get(n) != v} print("verified %d bindings in %s" % (len(found), pathlib.Path(path).name)) if not bad: print(" all match - the screens are bound to the live items") return 0 print(" %d MISMATCH(ES) - these values would read 0:" % len(bad)) for n, (theirs, ours) in sorted(bad.items()): print(" %-30s CI Server %s generated %s" % (n.replace("AID.WRPS.", ""), theirs, ours)) print(REHARVEST) return 1 def main(): import sys if len(sys.argv) > 2 and sys.argv[1] == "--harvest": harvest(sys.argv[2]) return if len(sys.argv) > 2 and sys.argv[1] == "--verify": sys.exit(verify(sys.argv[2])) OUT.mkdir(exist_ok=True) displays = [("WRPS_Overview", overview()), ("WRPS_Tank", tank()), ("WRPS_FP_PU301", fp_pump(1)), ("WRPS_FP_PU302", fp_pump(2)), ("WRPS_FP_PU303", fp_pump(3)), ("WRPS_FP_Setpoints", fp_setpoints())] check_items(displays) check_ids(displays) report_unplaced(displays) for name, disp in displays: path = OUT / (name + ".xml") path.write_text(disp.xml(), encoding="utf-8") print("OK %-22s %5d components, %3d shared objects, %6d bytes" % (path.name, len(disp.parts), len(disp.refs), path.stat().st_size)) if __name__ == "__main__": main()