分析アプリの構築
対応エディションCloudEnterprise
Tier0 は Python による高度なデータ分析に Marimo Notebook を使用します。このガイドでは Bowtie アプリケーションを例として使用します。
flowchart LR
collect["Source Flow"] -->|"収集データ"| uns[("UNS<br/>モデル")]
uns -->|"生データ"| notebook["Notebook"]
notebook -->|"分析結果"| uns
uns -->|"分析結果"| builder["App Builder"]
classDef t0accent fill:#EAF8C8,stroke:#A6CF38,stroke-width:1px,color:#171717
classDef t0soft fill:#F7FAF2,stroke:#D8E6B8,stroke-width:1px,color:#2A2A2A
classDef t0agent fill:#EEF4FF,stroke:#B7C7E8,stroke-width:1px,color:#1F2937
class uns t0accent
class collect,notebook t0soft
class builder t0agent
サンプル背景
Section titled “サンプル背景”製油所の腐食は、複数の運転要因に影響される複雑なプロセスです。この例では、プロセスデータを組み合わせて腐食発生確率を推定し、予防保全を支援するリアルタイム腐食リスク評価を示します。
生データを取得する
Section titled “生データを取得する”-
Tier0 で UNS に移動し、データモデルをインポートします。
UNS モデル JSON を表示
Terminal window {"notes": "type:PATH|TOPIC,topicType:STATE|ACTION|METRIC,schema.type:INTEGER|STRING|FLOAT|DOUBLE|BOOLEAN|LONG|DATETIME","namespace": [{"name": "Refinery","type": "PATH","children": [{"name": "CDU_Plant","type": "PATH","children": [{"name": "State","type": "PATH","topicType": "STATE","children": [{"name": "Corrosion_Risk","type": "TOPIC","topicType": "STATE","description": "Current corrosion risk state inferred by Bayesian Network","enableHistory": "TRUE","schema": [{"name": "risk_state","type": "STRING"},{"name": "previous_state","type": "STRING"},{"name": "model_confidence","type": "FLOAT"},{"name": "timestamp","type": "DATETIME"}]}]},{"name": "Metric","type": "PATH","topicType": "METRIC","children": [{"name": "Corrosion_Monitoring","type": "TOPIC","topicType": "METRIC","description": "Real-time process measurements for CDU atmospheric overhead corrosion monitoring","enableHistory": "TRUE","mockData": "FALSE","schema": [{"name": "d103_chloride","type": "FLOAT"},{"name": "d103_ph","type": "FLOAT"},{"name": "wash_water_flow","type": "FLOAT"},{"name": "desalter_salt_ptb","type": "FLOAT"},{"name": "desalter_bsw","type": "FLOAT"},{"name": "wash_water_rate","type": "FLOAT"},{"name": "rrd_ph","type": "FLOAT"},{"name": "rrd_chloride","type": "FLOAT"},{"name": "rrd_total_iron","type": "FLOAT"},{"name": "sap_evidence","type": "STRING"},{"name": "injection_quill_fail","type": "BOOLEAN"},{"name": "timestamp","type": "DATETIME"}]},{"name": "Corrosion_Analysis","type": "TOPIC","topicType": "METRIC","description": "Posterior probabilities generated by the Bayesian Network","enableHistory": "TRUE","mockData": "FALSE","schema": [{"name": "p_normal","type": "DOUBLE"},{"name": "p_developing","type": "DOUBLE"},{"name": "p_confirmed","type": "DOUBLE"},{"name": "total_corrosion_risk","type": "DOUBLE"},{"name": "lopc_probability","type": "DOUBLE"},{"name": "shutdown_probability","type": "DOUBLE"},{"name": "escalation_probability","type": "DOUBLE"},{"name": "timestamp","type": "DATETIME"}]}]}]}]}]} -
Flows に移動し、Source Flow を作成して生データを接続し、UNS に publish します。
Source Flow JSON を表示
Terminal window [{"id": "6173469ed5064ccf","type": "tab","label": "corrosion","disabled": false,"info": ""},{"id": "ad20b016baf7247f","type": "OpcUa-Client","z": "6173469ed5064ccf","endpoint": "66bd118c33eb12f8","action": "subscribe","deadbandtype": "a","deadbandvalue": 1,"time": 10,"timeUnit": "s","certificate": "n","localfile": "","localkeyfile": "","securitymode": "None","securitypolicy": "None","useTransport": false,"maxChunkCount": 1,"maxMessageSize": 8192,"receiveBufferSize": 8192,"sendBufferSize": 8192,"setstatusandtime": false,"keepsessionalive": false,"name": "","x": 280,"y": 440,"wires": [["ac9195bb4863dbdd","5a3d61b2ae40dcb6"],[],[]]},{"id": "b6746ed066ba0e1a","type": "inject","z": "6173469ed5064ccf","name": "","props": [{"p": "payload"},{"p": "topic","vt": "str"}],"repeat": "","crontab": "","once": false,"onceDelay": 0.1,"topic": "multiple","payload": "[ { \"name\": \"D103_CHLORIDE\", \"nodeId\": \"ns=2;i=3\", \"datatype\": \"Float\" }, { \"name\": \"D103_PH\", \"nodeId\": \"ns=2;i=4\", \"datatype\": \"Float\" }, { \"name\": \"WASH_WATER_FLOW\", \"nodeId\": \"ns=2;i=5\", \"datatype\": \"Float\" }, { \"name\": \"DESALTER_SALT_PTB\", \"nodeId\": \"ns=2;i=6\", \"datatype\": \"Float\" }, { \"name\": \"DESALTER_BSW\", \"nodeId\": \"ns=2;i=7\", \"datatype\": \"Float\" }, { \"name\": \"WASH_WATER_RATE\", \"nodeId\": \"ns=2;i=8\", \"datatype\": \"Float\" }, { \"name\": \"RRD_PH\", \"nodeId\": \"ns=2;i=9\", \"datatype\": \"Float\" }, { \"name\": \"RRD_CHLORIDE\", \"nodeId\": \"ns=2;i=10\", \"datatype\": \"Float\" }, { \"name\": \"RRD_TOTAL_IRON\", \"nodeId\": \"ns=2;i=11\", \"datatype\": \"Float\" }, { \"name\": \"SAP_EVIDENCE\", \"nodeId\": \"ns=2;i=12\", \"datatype\": \"UInt16\" }, { \"name\": \"INJECTION_QUILL_FAIL\", \"nodeId\": \"ns=2;i=13\", \"datatype\": \"Boolean\" } ]","payloadType": "json","x": 90,"y": 440,"wires": [["ad20b016baf7247f"]]},{"id": "ac9195bb4863dbdd","type": "debug","z": "6173469ed5064ccf","name": "debug 2","active": true,"tosidebar": true,"console": false,"tostatus": false,"complete": "false","statusVal": "","statusType": "auto","x": 490,"y": 480,"wires": []},{"id": "5a3d61b2ae40dcb6","type": "function","z": "6173469ed5064ccf","name": "function 1","func": "const TOPIC = \"Refinery/CDU_Plant/Metric/Corrosion_Monitoring\";\n\nconst nodeMap = {\n \"ns=2;i=3\": { field: \"d103_chloride\", type: \"float\" },\n \"ns=2;i=4\": { field: \"d103_ph\", type: \"float\" },\n \"ns=2;i=5\": { field: \"wash_water_flow\", type: \"float\" },\n \"ns=2;i=6\": { field: \"desalter_salt_ptb\", type: \"float\" },\n \"ns=2;i=7\": { field: \"desalter_bsw\", type: \"float\" },\n \"ns=2;i=8\": { field: \"wash_water_rate\", type: \"float\" },\n \"ns=2;i=9\": { field: \"rrd_ph\", type: \"float\" },\n \"ns=2;i=10\": { field: \"rrd_chloride\", type: \"float\" },\n \"ns=2;i=11\": { field: \"rrd_total_iron\", type: \"float\" },\n \"ns=2;i=12\": { field: \"sap_evidence\", type: \"sap\" },\n \"ns=2;i=13\": { field: \"injection_quill_fail\", type: \"boolean\" }\n};\n\nconst sapMap = {\n 0: \"none\",\n 1: \"work_order\",\n 2: \"inspection_finding\"\n};\n\nfunction normalizeNodeId(v) {\n if (!v) return \"\";\n if (typeof v === \"string\") return v;\n if (v.toString) return v.toString();\n return \"\";\n}\n\nfunction extractValue(payload) {\n // OPC UA DataValue: payload.value.value\n if (payload && payload.value && payload.value.value !== undefined) {\n return payload.value.value;\n }\n\n // Some OPC UA nodes output Variant directly.\n if (payload && payload.value !== undefined) {\n return payload.value;\n }\n\n return payload;\n}\n\nfunction extractNodeId(msg) {\n return normalizeNodeId(\n msg.nodeId ||\n msg.topic ||\n msg.payload?.nodeId ||\n msg.payload?.nodeid\n );\n}\n\nconst nodeId = extractNodeId(msg);\nconst spec = nodeMap[nodeId];\n\nif (!spec) {\n node.warn(`Unknown OPC UA nodeId: ${nodeId}`);\n return null;\n}\n\nlet value = extractValue(msg.payload);\n\nif (spec.type === \"float\") {\n value = Number(value);\n} else if (spec.type === \"boolean\") {\n value = Boolean(value);\n} else if (spec.type === \"sap\") {\n value = sapMap[Number(value)] || \"unknown\";\n}\n\nlet latest = context.get(\"latest\") || {};\nlatest[spec.field] = value;\nlatest.timestamp = new Date().toISOString();\ncontext.set(\"latest\", latest);\n\n// Publish only after all 11 fields have been received at least once.\nconst requiredFields = Object.values(nodeMap).map(x => x.field);\nconst ready = requiredFields.every(field => latest[field] !== undefined);\n\nif (!ready) {\n node.status({\n fill: \"yellow\",\n shape: \"ring\",\n text: `waiting ${Object.keys(latest).length - 1}/11`\n });\n return null;\n}\n\nmsg.topic = TOPIC;\nmsg.payload = {\n d103_chloride: latest.d103_chloride,\n d103_ph: latest.d103_ph,\n wash_water_flow: latest.wash_water_flow,\n desalter_salt_ptb: latest.desalter_salt_ptb,\n desalter_bsw: latest.desalter_bsw,\n wash_water_rate: latest.wash_water_rate,\n rrd_ph: latest.rrd_ph,\n rrd_chloride: latest.rrd_chloride,\n rrd_total_iron: latest.rrd_total_iron,\n sap_evidence: latest.sap_evidence,\n injection_quill_fail: latest.injection_quill_fail,\n timestamp: latest.timestamp\n};\n\nnode.status({\n fill: \"green\",\n shape: \"dot\",\n text: \"published corrosion monitoring\"\n});\n\nreturn msg;","outputs": 1,"timeout": 0,"noerr": 0,"initialize": "","finalize": "","libs": [],"x": 370,"y": 580,"wires": [["aa15be33a84fb812","4fdd68dda3ee5c9e"]]},{"id": "aa15be33a84fb812","type": "mqtt out","z": "6173469ed5064ccf","name": "","topic": "","qos": "","retain": "","respTopic": "","contentType": "","userProps": "","correl": "","expiry": "","broker": "broker-djx75cwgiywq","x": 550,"y": 580,"wires": []},{"id": "4fdd68dda3ee5c9e","type": "debug","z": "6173469ed5064ccf","name": "debug 3","active": true,"tosidebar": true,"console": false,"tostatus": false,"complete": "false","statusVal": "","statusType": "auto","x": 450,"y": 680,"wires": []},{"id": "66bd118c33eb12f8","type": "OpcUa-Endpoint","endpoint": "opc.tcp://172.31.151.237:4850","secpol": "None","secmode": "None","none": true,"login": false,"usercert": false,"usercertificate": "","userprivatekey": ""},{"id": "broker-djx75cwgiywq","type": "mqtt-broker","z": "6173469ed5064ccf","name": "corrosion","broker": "emqx","port": "1883","clientid": "357356352615696","usetls": false,"protocolVersion": "4","keepalive": "60","cleansession": true,"birthTopic": "","birthQos": "0","birthPayload": "","closeTopic": "","closeQos": "0","closePayload": "","willTopic": "","willQos": "0","willPayload": ""},{"id": "c5201f1ad482b182","type": "global-config","env": [],"modules": {"node-red-contrib-opcua": "0.2.339"}}]
Notebook で分析アプリを構築する
Section titled “Notebook で分析アプリを構築する”-
Tier0 で Notebook に移動し、新しい notebook を作成します。
-
notebook を開き、UNS のデータを分析するために次の cell を追加します。
-
セル 1 - 必要な依存関係をインポートし、MQTT broker、topics、実行時設定を構成します。
コードを表示
Terminal window @app.celldef _():"""Imports and runtime configuration."""import jsonimport mathimport osimport threadingimport timefrom datetime import datetime, timezoneimport marimo as moimport pandas as pdimport paho.mqtt.client as mqttMQTT_BROKER_HOST = os.getenv("TIER0_MQTT_HOST", "enterprise-new-dev.tier0.dev")MQTT_BROKER_PORT = int(os.getenv("TIER0_MQTT_PORT", "1883"))MQTT_USERNAME = os.getenv("TIER0_MQTT_USERNAME", "357890055321856")MQTT_PASSWORD = os.getenv("TIER0_MQTT_PASSWORD", "caa6b34242002a3dff4948dd73ce8a")MQTT_CLIENT_ID = os.getenv("TIER0_MQTT_CLIENT_ID","357890055321856_bn",)SOURCE_TOPIC = "Refinery/CDU_Plant/Metric/Corrosion_Monitoring"ANALYSIS_TOPIC = "Refinery/CDU_Plant/Metric/Corrosion_Analysis"RISK_TOPIC = "Refinery/CDU_Plant/State/Corrosion_Risk"SOURCE_QOS = int(os.getenv("TIER0_SOURCE_QOS", "1"))OUTPUT_QOS = int(os.getenv("TIER0_OUTPUT_QOS", "1"))AUTOSTART = os.getenv("TIER0_BN_AUTOSTART", "true").lower() in {"1", "true", "yes", "on"}ROOT_CONFIDENCE = float(os.getenv("TIER0_BN_ROOT_CONFIDENCE", "0.85"))def utc_now_iso() -> str:return (datetime.now(timezone.utc).isoformat(timespec="milliseconds").replace("+00:00", "Z"))mo.md("""# CDU Corrosion Bayesian AnalysisThis notebook subscribes to a complete corrosion-monitoring snapshot,performs one Bayesian inference for each new source message, and publishesthe analysis results back to the UNS.""")return (ANALYSIS_TOPIC,AUTOSTART,MQTT_BROKER_HOST,MQTT_BROKER_PORT,MQTT_CLIENT_ID,MQTT_PASSWORD,MQTT_USERNAME,OUTPUT_QOS,RISK_TOPIC,ROOT_CONFIDENCE,SOURCE_QOS,SOURCE_TOPIC,datetime,json,math,mo,mqtt,os,pd,threading,time,timezone,utc_now_iso,) -
セル 2 - Bayesian Network の topology、状態空間、CPT generator、確率ルールを定義します。
コードを表示
Terminal window @app.celldef _(math, pd):"""Static Bayesian-network definition and CPT generators."""import itertoolsfrom pgmpy.factors.discrete import TabularCPDfrom pgmpy.inference import VariableEliminationfrom pgmpy.models import DiscreteBayesianNetworkSTATE_SPACES = {"STATE_3": ["LOW", "NORMAL", "HIGH"],"STATE_YN": ["NO", "YES"],"PB2_3STATE": ["FAIL", "IL_OK", "SL_OK"],"TOP_EVENT_STATE": ["NORMAL", "DEVELOPING", "CONFIRMED"],}NODE_STATE_SPACE = {"T1_HighChlorideLoad": "STATE_3","T2_ChemProtectionFail": "STATE_3","T3_WashWaterLow": "STATE_3","PB1_Desalter_OK": "STATE_YN","PB2_WashQty_State": "PB2_3STATE","PB3_Neutralization_OK": "STATE_YN","MB1_ESD_OK": "STATE_YN","MB2_TSV_OK": "STATE_YN","TE_CorrosionState": "TOP_EVENT_STATE","EV_TOTIRON_High": "STATE_YN","EV_SAP_Positive": "STATE_YN","EV_QuillFailure": "STATE_YN","C1_LOPC": "STATE_YN","C2_UnplannedShutdown": "STATE_YN","C3_Escalation": "STATE_YN",}EDGES = [("T1_HighChlorideLoad", "TE_CorrosionState"),("T2_ChemProtectionFail", "TE_CorrosionState"),("T3_WashWaterLow", "TE_CorrosionState"),("PB1_Desalter_OK", "TE_CorrosionState"),("PB2_WashQty_State", "TE_CorrosionState"),("PB3_Neutralization_OK", "TE_CorrosionState"),("TE_CorrosionState", "EV_TOTIRON_High"),("TE_CorrosionState", "EV_SAP_Positive"),("TE_CorrosionState", "EV_QuillFailure"),("TE_CorrosionState", "C1_LOPC"),("MB1_ESD_OK", "C1_LOPC"),("MB2_TSV_OK", "C1_LOPC"),("TE_CorrosionState", "C2_UnplannedShutdown"),("MB1_ESD_OK", "C2_UnplannedShutdown"),("C1_LOPC", "C3_Escalation"),("MB1_ESD_OK", "C3_Escalation"),("MB2_TSV_OK", "C3_Escalation"),]def states_for(node_id: str) -> list[str]:return STATE_SPACES[NODE_STATE_SPACE[node_id]]def softmax(values: list[float]) -> list[float]:maximum = max(values)exps = [math.exp(value - maximum) for value in values]total = sum(exps)return [value / total for value in exps]THREAT_SCORE = {"LOW": 0.0, "NORMAL": 1.0, "HIGH": 2.0}PB1_OK_BONUS = {"NO": 0.8, "YES": -0.6}PB3_OK_BONUS = {"NO": 1.0, "YES": -0.7}PB2_STATE_BONUS = {"FAIL": 1.2, "IL_OK": 0.3, "SL_OK": -0.6}def top_event_distribution(t1, t2, t3, pb1, pb2, pb3) -> dict[str, float]:score = (THREAT_SCORE[t1]+ THREAT_SCORE[t2]+ THREAT_SCORE[t3]+ PB1_OK_BONUS[pb1]+ PB2_STATE_BONUS[pb2]+ PB3_OK_BONUS[pb3])probabilities = softmax([2.2 - score,-0.2 + 0.6 * score,-1.2 + 0.8 * score,])return dict(zip(STATE_SPACES["TOP_EVENT_STATE"], probabilities))TE_PARENTS = ["T1_HighChlorideLoad","T2_ChemProtectionFail","T3_WashWaterLow","PB1_Desalter_OK","PB2_WashQty_State","PB3_Neutralization_OK",]TE_PARENT_SPACES = [states_for(parent) for parent in TE_PARENTS]te_rows = []for combination in itertools.product(*TE_PARENT_SPACES):distribution = top_event_distribution(*combination)te_rows.append([*combination,distribution["NORMAL"],distribution["DEVELOPING"],distribution["CONFIRMED"],])TE_CPT_DF = pd.DataFrame(te_rows,columns=TE_PARENTS+ ["P_NORMAL", "P_DEVELOPING", "P_CONFIRMED"],)def evidence_yes_probability(te_state: str, kind: str) -> float:mappings = {"TOTIRON": {"NORMAL": 0.15, "DEVELOPING": 0.65, "CONFIRMED": 0.85},"SAP": {"NORMAL": 0.08, "DEVELOPING": 0.35, "CONFIRMED": 0.75},"QUILL": {"NORMAL": 0.03, "DEVELOPING": 0.15, "CONFIRMED": 0.55},}return mappings[kind][te_state]def lopc_yes_probability(te: str, mb1: str, mb2: str) -> float:base = {"NORMAL": 0.02, "DEVELOPING": 0.08, "CONFIRMED": 0.25}[te]return min(0.98, base + (0.10 if mb1 == "NO" else 0) + (0.18 if mb2 == "NO" else 0))def shutdown_yes_probability(te: str, mb1: str) -> float:base = {"NORMAL": 0.05, "DEVELOPING": 0.35, "CONFIRMED": 0.60}[te]return min(0.98, base + (0.10 if mb1 == "NO" else 0))def escalation_yes_probability(lopc: str, mb1: str, mb2: str) -> float:if lopc == "NO":return 0.01return min(0.98, 0.10 + (0.18 if mb1 == "NO" else 0) + (0.35 if mb2 == "NO" else 0))def peaked_prior(node_id: str, center_state: str, confidence: float) -> list[float]:states = states_for(node_id)confidence = min(max(float(confidence), 0.0), 1.0)if center_state not in states:return [1.0 / len(states)] * len(states)if len(states) == 1:return [1.0]remainder = (1.0 - confidence) / (len(states) - 1)probabilities = [remainder] * len(states)probabilities[states.index(center_state)] = confidencereturn probabilitiesdef binary_cpd(child: str, parent: str, yes_probability_fn):parent_states = states_for(parent)yes_values = [float(yes_probability_fn(state)) for state in parent_states]return TabularCPD(variable=child,variable_card=2,values=[[1.0 - value for value in yes_values], yes_values],evidence=[parent],evidence_card=[len(parent_states)],state_names={child: states_for(child), parent: parent_states},)def binary_multi_cpd(child: str, parents: list[str], yes_probability_fn):parent_spaces = [states_for(parent) for parent in parents]combinations = list(itertools.product(*parent_spaces))yes_values = [float(yes_probability_fn(*combination)) for combination in combinations]return TabularCPD(variable=child,variable_card=2,values=[[1.0 - value for value in yes_values], yes_values],evidence=parents,evidence_card=[len(space) for space in parent_spaces],state_names={child: states_for(child),**{parent: states_for(parent) for parent in parents},},)return (DiscreteBayesianNetwork,EDGES,NODE_STATE_SPACE,STATE_SPACES,TE_CPT_DF,TE_PARENTS,TabularCPD,VariableElimination,binary_cpd,binary_multi_cpd,escalation_yes_probability,evidence_yes_probability,itertools,lopc_yes_probability,peaked_prior,shutdown_yes_probability,states_for,) -
セル 3 - 受信した MQTT payload を検証し、連続的な process value を Bayesian Network の状態に変換します。
コードを表示
Terminal window @app.celldef _(ROOT_CONFIDENCE, utc_now_iso):"""Source validation, discretization, and one-snapshot analysis."""import hashlibREQUIRED_SOURCE_FIELDS = ["d103_chloride","d103_ph","wash_water_flow","desalter_salt_ptb","desalter_bsw","wash_water_rate","rrd_ph","rrd_chloride","rrd_total_iron","sap_evidence","injection_quill_fail","timestamp",]def parse_boolean(value) -> bool:if isinstance(value, bool):return valueif isinstance(value, (int, float)):return bool(value)if isinstance(value, str):normalized = value.strip().lower()if normalized in {"true", "1", "yes", "y", "on"}:return Trueif normalized in {"false", "0", "no", "n", "off"}:return Falseraise ValueError(f"invalid boolean value: {value!r}")def normalize_sap(value) -> str:normalized = str(value).strip().lower()if normalized in {"0", "none", "normal", "", "nan"}:return "None"if normalized in {"1", "work_order", "work order", "minor", "warning"}:return "Minor corrosion observed"if normalized in {"2","inspection_finding","inspection finding","confirmed","abnormal",}:return "Confirmed corrosion / thickness loss"raise ValueError(f"unsupported sap_evidence value: {value!r}")def normalize_source_payload(payload: dict) -> dict:if not isinstance(payload, dict):raise TypeError("source MQTT payload must be a JSON object")lower = {str(key).lower(): value for key, value in payload.items()}missing = [field for field in REQUIRED_SOURCE_FIELDS if field not in lower]if missing:raise ValueError(f"source payload is missing fields: {missing}")normalized = {"d103_chloride": float(lower["d103_chloride"]),"d103_ph": float(lower["d103_ph"]),"wash_water_flow": float(lower["wash_water_flow"]),"desalter_salt_ptb": float(lower["desalter_salt_ptb"]),"desalter_bsw": float(lower["desalter_bsw"]),"wash_water_rate": float(lower["wash_water_rate"]),"rrd_ph": float(lower["rrd_ph"]),"rrd_chloride": float(lower["rrd_chloride"]),"rrd_total_iron": float(lower["rrd_total_iron"]),"sap_evidence": normalize_sap(lower["sap_evidence"]),"injection_quill_fail": parse_boolean(lower["injection_quill_fail"]),"timestamp": str(lower["timestamp"]).strip(),}numeric_ranges = {"d103_chloride": (0.0, 1000.0),"d103_ph": (0.0, 14.0),"wash_water_flow": (0.0, 10000.0),"desalter_salt_ptb": (0.0, 1000.0),"desalter_bsw": (0.0, 100.0),"wash_water_rate": (0.0, 100.0),"rrd_ph": (0.0, 14.0),"rrd_chloride": (0.0, 10000.0),"rrd_total_iron": (0.0, 10000.0),}for field, (minimum, maximum) in numeric_ranges.items():value = normalized[field]if not minimum <= value <= maximum:raise ValueError(f"{field} out of accepted range: {value}")if not normalized["timestamp"]:raise ValueError("timestamp cannot be empty")return normalizeddef source_message_id(payload: dict) -> str:# Timestamp is the primary event key. Hash protects against two different# snapshots accidentally sharing the same timestamp.ordered = "|".join(str(payload[field]) for field in REQUIRED_SOURCE_FIELDS)return hashlib.sha256(ordered.encode("utf-8")).hexdigest()def discretize_source(payload: dict) -> tuple[dict, dict]:t1 = ("LOW"if payload["d103_chloride"] < 5.0else "NORMAL"if payload["d103_chloride"] <= 20.0else "HIGH")chemical_failure_score = int(payload["d103_ph"] < 5.5) + int(payload["d103_chloride"] > 10.0)t2 = ("HIGH"if chemical_failure_score == 2else "NORMAL"if chemical_failure_score == 1else "LOW")t3 = ("HIGH"if payload["wash_water_flow"] < 6.0else "NORMAL"if payload["wash_water_flow"] <= 10.0else "LOW")pb1 = ("YES"if payload["desalter_salt_ptb"] <= 1.0and payload["desalter_bsw"] <= 0.2else "NO")pb2 = ("FAIL"if payload["wash_water_rate"] < 4.0else "IL_OK"if payload["wash_water_rate"] < 6.0else "SL_OK")pb3 = ("YES"if payload["rrd_ph"] >= 5.5 and payload["rrd_chloride"] <= 10.0else "NO")root_centers = {"T1_HighChlorideLoad": t1,"T2_ChemProtectionFail": t2,"T3_WashWaterLow": t3,"PB1_Desalter_OK": pb1,"PB2_WashQty_State": pb2,"PB3_Neutralization_OK": pb3,# The source model currently has no MB fields. Keep both mitigations# available by default, matching the previous notebook UI defaults."MB1_ESD_OK": "YES","MB2_TSV_OK": "YES",}evidence = {"EV_TOTIRON_High": "YES" if payload["rrd_total_iron"] >= 10.0 else "NO","EV_SAP_Positive": "YES" if payload["sap_evidence"] != "None" else "NO","EV_QuillFailure": "YES" if payload["injection_quill_fail"] else "NO",}return root_centers, evidencereturn (REQUIRED_SOURCE_FIELDS,discretize_source,normalize_source_payload,parse_boolean,source_message_id,) -
セル 4 - 現在の snapshot に対して Bayesian Network を構築し、推論を実行して output payload を生成します。
コードを表示
Terminal window @app.celldef _(DiscreteBayesianNetwork,EDGES,ROOT_CONFIDENCE,STATE_SPACES,TE_CPT_DF,TE_PARENTS,TabularCPD,VariableElimination,binary_cpd,binary_multi_cpd,discretize_source,escalation_yes_probability,evidence_yes_probability,itertools,lopc_yes_probability,peaked_prior,shutdown_yes_probability,states_for,utc_now_iso,):"""Build a fresh BN for the current snapshot and execute inference."""def factor_probability(factor, variable: str, state: str) -> float:states = factor.state_names[variable]values = list(map(float, factor.values))return float(values[states.index(state)])def build_model(root_centers: dict):model = DiscreteBayesianNetwork(EDGES)root_cpds = []for node_id, center_state in root_centers.items():states = states_for(node_id)probabilities = peaked_prior(node_id, center_state, ROOT_CONFIDENCE)root_cpds.append(TabularCPD(variable=node_id,variable_card=len(states),values=[[probability] for probability in probabilities],state_names={node_id: states},))te_values = [TE_CPT_DF["P_NORMAL"].astype(float).tolist(),TE_CPT_DF["P_DEVELOPING"].astype(float).tolist(),TE_CPT_DF["P_CONFIRMED"].astype(float).tolist(),]te_cpd = TabularCPD(variable="TE_CorrosionState",variable_card=3,values=te_values,evidence=TE_PARENTS,evidence_card=[len(states_for(parent)) for parent in TE_PARENTS],state_names={"TE_CorrosionState": STATE_SPACES["TOP_EVENT_STATE"],**{parent: states_for(parent) for parent in TE_PARENTS},},)evidence_cpds = [binary_cpd("EV_TOTIRON_High","TE_CorrosionState",lambda state: evidence_yes_probability(state, "TOTIRON"),),binary_cpd("EV_SAP_Positive","TE_CorrosionState",lambda state: evidence_yes_probability(state, "SAP"),),binary_cpd("EV_QuillFailure","TE_CorrosionState",lambda state: evidence_yes_probability(state, "QUILL"),),]consequence_cpds = [binary_multi_cpd("C1_LOPC",["TE_CorrosionState", "MB1_ESD_OK", "MB2_TSV_OK"],lopc_yes_probability,),binary_multi_cpd("C2_UnplannedShutdown",["TE_CorrosionState", "MB1_ESD_OK"],shutdown_yes_probability,),binary_multi_cpd("C3_Escalation",["C1_LOPC", "MB1_ESD_OK", "MB2_TSV_OK"],escalation_yes_probability,),]model.add_cpds(*root_cpds, te_cpd, *evidence_cpds, *consequence_cpds)if not model.check_model():raise RuntimeError("Bayesian network model validation failed")return modeldef analyze_snapshot(payload: dict, previous_risk_state: str | None = None) -> dict:root_centers, evidence = discretize_source(payload)model = build_model(root_centers)inference = VariableElimination(model)q_te = inference.query(["TE_CorrosionState"], evidence=evidence, show_progress=False)q_c1 = inference.query(["C1_LOPC"], evidence=evidence, show_progress=False)q_c2 = inference.query(["C2_UnplannedShutdown"], evidence=evidence, show_progress=False)q_c3 = inference.query(["C3_Escalation"], evidence=evidence, show_progress=False)p_normal = factor_probability(q_te, "TE_CorrosionState", "NORMAL")p_developing = factor_probability(q_te, "TE_CorrosionState", "DEVELOPING")p_confirmed = factor_probability(q_te, "TE_CorrosionState", "CONFIRMED")probabilities = {"NORMAL": p_normal,"DEVELOPING": p_developing,"CONFIRMED": p_confirmed,}risk_state = max(probabilities, key=probabilities.get)analysis_timestamp = utc_now_iso()analysis_payload = {"p_normal": p_normal,"p_developing": p_developing,"p_confirmed": p_confirmed,"total_corrosion_risk": p_developing + p_confirmed,"lopc_probability": factor_probability(q_c1, "C1_LOPC", "YES"),"shutdown_probability": factor_probability(q_c2, "C2_UnplannedShutdown", "YES"),"escalation_probability": factor_probability(q_c3, "C3_Escalation", "YES"),"timestamp": analysis_timestamp,}risk_payload = {"risk_state": risk_state,"previous_state": previous_risk_state or risk_state,"model_confidence": max(probabilities.values()),"timestamp": analysis_timestamp,}return {"source_timestamp": payload["timestamp"],"root_states": root_centers,"evidence": evidence,"analysis_payload": analysis_payload,"risk_payload": risk_payload,"risk_state": risk_state,}return analyze_snapshot, build_model, factor_probability -
セル 5 - source topic を購読し、受信した各 snapshot を処理して Bayesian 分析を実行し、結果を publish します。
コードを表示
Terminal window @app.celldef _(ANALYSIS_TOPIC,AUTOSTART,MQTT_BROKER_HOST,MQTT_BROKER_PORT,MQTT_CLIENT_ID,MQTT_PASSWORD,MQTT_USERNAME,OUTPUT_QOS,RISK_TOPIC,SOURCE_QOS,SOURCE_TOPIC,analyze_snapshot,json,mqtt,normalize_source_payload,source_message_id,threading,utc_now_iso,):"""Long-running MQTT subscriber and publisher service."""class CorrosionBNService:def __init__(self):self.lock = threading.RLock()self.client = Noneself.running = Falseself.connected = Falseself.last_message_id = Noneself.last_source_timestamp = Noneself.last_analysis_timestamp = Noneself.previous_risk_state = Noneself.processed_count = 0self.duplicate_count = 0self.error_count = 0self.last_error = Noneself.last_result = Noneself.connection_rc = Nonedef _create_client(self):# Use MQTT 3.1.1 and callback API v1 for broad compatibility with# EMQX and different paho-mqtt releases bundled in Notebook images.kwargs = {"client_id": MQTT_CLIENT_ID,"clean_session": True,"protocol": mqtt.MQTTv311,"transport": "tcp",}try:client = mqtt.Client(callback_api_version=mqtt.CallbackAPIVersion.VERSION1,**kwargs,)except (AttributeError, TypeError):client = mqtt.Client(**kwargs)client.username_pw_set(MQTT_USERNAME, MQTT_PASSWORD)client.on_connect = self._on_connectclient.on_disconnect = self._on_disconnectclient.on_message = self._on_messageclient.reconnect_delay_set(min_delay=1, max_delay=30)client.enable_logger()return clientdef _on_connect(self, client, userdata, flags, rc, properties=None):code = int(rc)with self.lock:self.connected = code == 0self.connection_rc = codeself.last_error = None if code == 0 else f"MQTT connection rejected, rc={code}"if code == 0:sub_rc, mid = client.subscribe(SOURCE_TOPIC, qos=SOURCE_QOS)if sub_rc != mqtt.MQTT_ERR_SUCCESS:with self.lock:self.last_error = f"MQTT subscribe failed, rc={sub_rc}"returnprint(f"[BN SERVICE] connected and subscribed to {SOURCE_TOPIC}")else:print(f"[BN SERVICE ERROR] connection rejected rc={code}")def _on_disconnect(self, client, userdata, rc, properties=None):with self.lock:self.connected = Falseself.connection_rc = int(rc) if rc is not None else Noneif self.running and rc:self.last_error = f"MQTT disconnected unexpectedly, rc={rc}"print(f"[BN SERVICE] disconnected: {rc}")def _publish_json(self, topic: str, payload: dict):info = self.client.publish(topic,payload=json.dumps(payload, separators=(",", ":"), ensure_ascii=False),qos=OUTPUT_QOS,retain=False,)info.wait_for_publish(timeout=10)if info.rc != mqtt.MQTT_ERR_SUCCESS:raise RuntimeError(f"MQTT publish failed for {topic}: rc={info.rc}")def _on_message(self, client, userdata, msg):try:raw_payload = json.loads(msg.payload.decode("utf-8"))payload = normalize_source_payload(raw_payload)message_id = source_message_id(payload)with self.lock:if message_id == self.last_message_id:self.duplicate_count += 1returnprevious_risk_state = self.previous_risk_stateresult = analyze_snapshot(payload, previous_risk_state)# Publish only after both payloads have been generated successfully.self._publish_json(ANALYSIS_TOPIC, result["analysis_payload"])self._publish_json(RISK_TOPIC, result["risk_payload"])with self.lock:self.last_message_id = message_idself.last_source_timestamp = payload["timestamp"]self.last_analysis_timestamp = result["analysis_payload"]["timestamp"]self.previous_risk_state = result["risk_state"]self.processed_count += 1self.last_result = resultself.last_error = Noneprint("[BN SERVICE] processed",payload["timestamp"],"->",result["risk_state"],)except Exception as exc:with self.lock:self.error_count += 1self.last_error = f"{type(exc).__name__}: {exc}"print("[BN SERVICE ERROR]", self.last_error)def start(self):with self.lock:if self.running:return Falseif not MQTT_PASSWORD:raise RuntimeError("TIER0_MQTT_PASSWORD is empty. Set it before starting the service.")self.client = self._create_client()self.running = Truetry:# Synchronous connect surfaces DNS, socket, and authentication# failures immediately instead of leaving running=True forever.rc = self.client.connect(MQTT_BROKER_HOST,MQTT_BROKER_PORT,keepalive=30,)if rc != mqtt.MQTT_ERR_SUCCESS:raise RuntimeError(f"MQTT connect() failed, rc={rc}")self.client.loop_start()return Trueexcept Exception as exc:with self.lock:self.running = Falseself.connected = Falseself.last_error = f"{type(exc).__name__}: {exc}"self.client = Noneraisedef stop(self):with self.lock:if not self.running:return Falseclient = self.clientself.running = Falseself.connected = Falseself.client = Nonetry:client.disconnect()finally:client.loop_stop()return Truedef restart(self):self.stop()return self.start()def status(self) -> dict:with self.lock:return {"running": self.running,"connected": self.connected,"connection_rc": self.connection_rc,"client_id": MQTT_CLIENT_ID,"broker": f"{MQTT_BROKER_HOST}:{MQTT_BROKER_PORT}","source_topic": SOURCE_TOPIC,"analysis_topic": ANALYSIS_TOPIC,"risk_topic": RISK_TOPIC,"processed_count": self.processed_count,"duplicate_count": self.duplicate_count,"error_count": self.error_count,"last_source_timestamp": self.last_source_timestamp,"last_analysis_timestamp": self.last_analysis_timestamp,"previous_risk_state": self.previous_risk_state,"last_error": self.last_error,}# Reuse a previous service instance when the cell is re-executed, preventing# duplicate subscribers inside the same Python process.existing = globals().get("_CORROSION_BN_SERVICE")if existing is not None:try:existing.stop()except Exception:passCORROSION_BN_SERVICE = CorrosionBNService()globals()["_CORROSION_BN_SERVICE"] = CORROSION_BN_SERVICEAUTOSTART_ERROR = Noneif AUTOSTART:try:CORROSION_BN_SERVICE.start()except Exception as exc:AUTOSTART_ERROR = f"{type(exc).__name__}: {exc}"return AUTOSTART_ERROR, CORROSION_BN_SERVICE, CorrosionBNService -
セル 6 - MQTT 接続ステータスとサービス統計を表示します。
コードを表示
Terminal window @app.celldef _(AUTOSTART_ERROR, CORROSION_BN_SERVICE, mo, pd):"""Service status and usage instructions."""status = CORROSION_BN_SERVICE.status()blocks = [mo.md("## MQTT analysis service"),pd.DataFrame([status]),]if status.get("last_error"):blocks.append(mo.md(f"### Connection / processing error\n`{status['last_error']}`"))elif status.get("running") and not status.get("connected"):blocks.append(mo.md("### Connecting\nThe MQTT client has started but has not completed the broker connection yet."))if AUTOSTART_ERROR:blocks.append(mo.md(f"""### Service not started`{AUTOSTART_ERROR}`Set the MQTT password in the Notebook environment and run this cell again:~~~pythonimport os# MQTT credentials are already configured in this notebook.~~~"""))else:blocks.append(mo.md("""The service starts automatically and processes one complete sourcesnapshot for every new MQTT message. No manual analysis trigger or SQLpolling is required.Available controls in Python:~~~pythonCORROSION_BN_SERVICE.status()CORROSION_BN_SERVICE.stop()CORROSION_BN_SERVICE.start()CORROSION_BN_SERVICE.restart()~~~"""))mo.vstack(blocks)return status, -
セル 7 - 最新の Bayesian 分析結果を表示します。
コードを表示
Terminal window @app.celldef _(CORROSION_BN_SERVICE, mo):"""Display the most recent analysis result when one is available."""latest = CORROSION_BN_SERVICE.last_resultif latest is None:mo.md("### Latest result\nNo source message has been processed yet.")else:mo.vstack([mo.md("### Latest source snapshot result"),mo.md(f"- Source timestamp: `{latest['source_timestamp']}`"),mo.md(f"- Risk state: `{latest['risk_state']}`"),mo.md("#### Root states"),latest["root_states"],mo.md("#### Evidence"),latest["evidence"],mo.md("#### Corrosion analysis payload"),latest["analysis_payload"],mo.md("#### Corrosion risk payload"),latest["risk_payload"],])returnif __name__ == "__main__":app.run()
-
-
すべての cell を実行し、UNS に移動して結果を確認します。
Bow-tie App を構築する
Section titled “Bow-tie App を構築する”-
Tier0 で Builder に移動します。
-
ダイアログにアプリケーション要件を入力し、構築を開始します。
Builder prompt を表示
Terminal window CDU 常圧塔頂腐食向けの、モダンな産業用 Bow-Tie Analysis アプリケーションを構築してください。次の UNS topic のみをデータソースとして使用してください。- Refinery/CDU_Plant/Metric/Corrosion_Monitoring- Refinery/CDU_Plant/Metric/Corrosion_Analysis- Refinery/CDU_Plant/State/Corrosion_Risk新しい UNS メッセージが届いたら、アプリケーションは自動的に更新される必要があります。このアプリケーションは、オペレーターが現在の腐食リスク、関連する脅威、バリア状態、起こり得る結果を理解できるようにしてください。次を含む単一ページの dashboard を作成してください。1. メインセクションとして Bow-Tie 図を配置する：- 左側に脅威を表示- 脅威とトップイベントの間に予防バリアを表示- 腐食を中央のトップイベントとして表示- トップイベントの後に緩和バリアを表示- 右側に結果を表示2. 目立つリスクサマリーを表示する：- 現在のリスク状態- モデル信頼度- normal、developing、confirmed の確率- 総腐食リスク3. Corrosion_Monitoring から主要値を表示するコンパクトな live process data panel を作成する。chloride、pH、wash water flow、desalter performance、wash water rate、total iron、SAP evidence、injection quill status を含めてください。4. 結果指標：- 一次封じ込めの喪失- 計画外停止- エスカレーションリスク5. 視覚ステータスルール：- 緑は normal または healthy condition- 琥珀色/黄色は developing risk または warning condition- 赤は confirmed risk、failed barrier、または severe consequence- グレーは data unavailable- 青は interaction と informational highlight のみに使用6. インタラクション：- 脅威、バリア、結果を選択したら、現在の状態、関連する process value、短い説明を表示してください。- 新しい UNS データが到着したら、インターフェースを自動更新してください。- 最新データの timestamp と、組み込み SDK からの connection/subscription status を表示してください。デザイン要件：アプリケーション全体で、既存の visual design language と styling を維持してください。モダンな industrial operations dashboard スタイルを使用してください。- pure black ではなく dark blue-gray color palette を使用する。- subtle gradient を持つ layered panels を使用する。- 16-20px 程度の大きめの角丸を持つ card-based layout を使用する。- heavy shadow ではなく thin borders で分離する。- soft depth と控えめな glow effects を使用する。- 十分な padding、一貫した alignment、きれいな spacing を保つ。- body text には IBM Plex Sans を使用する。- headings には Space Grotesk を使用する。- 明確な visual hierarchy を持つ、コンパクトで情報密度の高い card を使用する。- letter spacing を広げた uppercase section labels を使用する。- business analytics ではなく、professional refinery / SCADA interface にする。- skeuomorphic elements ではなく、flat illustrations と SVG graphics を使用する。- smooth で subtle な transitions と hover effects を使用する。全体の見た目は clean、modern、technical、operational にしてください。避けるもの：- pure black backgrounds- glassmorphism- cyberpunk styling- neon lighting- overly decorative effects- business analytics dashboard stylingclear hierarchy、compact cards、readable labels、responsive layout を備えた、clean で professional な refinery operations style を使用してください。対応する UNS model で定義された payload fields を正確に使用してください。追加フィールド、計算値、mock data を作らないでください。 -
Once the application is complete after certain rounds of refining, deploy it.
-
Go to Launchpad, open the application and check.