#!/usr/bin/env python3 """ SIMONE Chatbot v1.6 AI-powered CLI for managing SIMONE gas network models. No external dependencies — uses Python stdlib only. Launched from within SIMONE environment (same as forecaster). Environment variables (optional overrides): ANTHROPIC_API_KEY - defaults to built-in key SIMONE_ROOT - auto-detected from sys.executable SIMONE_CHATBOT_MODEL - Claude model (default: claude-sonnet-4-6) """ import os import sys import json import fnmatch import traceback import urllib.request import urllib.error import ssl # ━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━ # Claude API (stdlib — no pip install needed) # ━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━ ANTHROPIC_API_URL = "https://api.anthropic.com/v1/messages" ANTHROPIC_API_VERSION = "2023-06-01" # Default key (override with ANTHROPIC_API_KEY env var) _DEFAULT_API_KEY = ( "sk-ant-api03-UhnPNpeSHZhZagCq9CFJT5J6pxUnm9cnFHDmDV0e" "SUJW89RbCXyzNnsVXLORHz0bM7wxtoq-JIESrFTlXyUi5w-Mc5CfAAA" ) def claude_api_call(api_key, model, system, tools, messages, max_tokens=4096): """Call Claude Messages API using urllib (no external dependencies).""" payload = { "model": model, "max_tokens": max_tokens, "system": system, "tools": tools, "messages": messages, } data = json.dumps(payload, ensure_ascii=False).encode("utf-8") req = urllib.request.Request( ANTHROPIC_API_URL, data=data, headers={ "Content-Type": "application/json", "x-api-key": api_key, "anthropic-version": ANTHROPIC_API_VERSION, }, method="POST", ) # Allow HTTPS (handle missing certs on some Windows installs) ctx = ssl.create_default_context() try: with urllib.request.urlopen(req, context=ctx, timeout=120) as resp: return json.loads(resp.read().decode("utf-8")) except urllib.error.HTTPError as e: body = e.read().decode("utf-8", errors="replace") raise RuntimeError(f"Claude API HTTP {e.code}: {body}") class ThinkingIndicator: """Shows a red 'Thinking...' message that disappears when done.""" # ANSI: red text, then reset RED = "\033[91m" RESET = "\033[0m" def __init__(self, message="Thinking..."): self._message = message def __enter__(self): sys.stdout.write(f"\n{self.RED}SIMONE> {self._message}{self.RESET}") sys.stdout.flush() return self def __exit__(self, *exc): # Move cursor back to start of line and clear it sys.stdout.write(f"\r\033[2K\033[A\033[2K\r") sys.stdout.flush() # ━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━ # SIMONE SDK Initialization # ━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━ def find_simone_root(): """Locate SIMONE installation directory (same logic as forecaster config.py).""" # 1. Env var override root = os.environ.get("SIMONE_ROOT") or os.environ.get("SIMONE_ROOT_OVERRIDE") if root and os.path.isdir(root): return root # 2. Auto-detect from sys.executable (SIMONE Python is at /Simone-API-SDK/Python3/python.exe) exe = os.path.abspath(sys.executable) candidate = os.path.dirname(os.path.dirname(os.path.dirname(exe))) if os.path.isdir(os.path.join(candidate, "exe")): return candidate # 3. Auto-detect from script location here = os.path.dirname(os.path.abspath(__file__)) for _ in range(4): if os.path.isdir(os.path.join(here, "exe")): return here here = os.path.dirname(here) # 4. Common Windows paths for p in [r"C:\Simone\Simone-V6_37e", r"C:\Simone\Simone-V6_37"]: if os.path.isdir(p): return p return None def init_simone_sdk(root): """Add SIMONE SDK to path and import SimoneApi.""" sdk = os.path.join(root, "Simone-API-SDK", "Python3") sdk_lib = os.path.join(sdk, "Lib") exe_dir = os.path.join(root, "exe") sys.path.insert(0, sdk) sys.path.insert(0, sdk_lib) if hasattr(os, "add_dll_directory") and os.path.isdir(exe_dir): os.add_dll_directory(exe_dir) import SimoneApi return SimoneApi # ━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━ # SIMONE Session Wrapper # ━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━ class SimoneSession: """High-level wrapper around SIMONE API for chatbot use.""" def __init__(self, api, simone_root): self.api = api self.root = simone_root self.F = getattr(api, "SIMONE_NO_FLAG", 0) self.OK = api.simone_status_ok self.h_nw = None self.network_name = None self.net_dir = None self._edit_started = False self._dirty = False self._element_type_map = {} self._param_map = {} self._scenario_open = False self._scenario_name = None # ── Connection ────────────────────────────────────────────── def connect(self): """Initialize SIMONE and detect current network.""" ss = self.api.simone_init("") if ss != self.OK: raise RuntimeError(f"simone_init failed (status={ss})") ss, network = self.api.simone_get_info( self.api.SIMONE_CONFIGURED_NETWORK, self.F ) if ss == self.OK and network and network.strip() != "VOID": self.network_name = network.strip() if self.network_name: self.api.simone_select(self.network_name) self._detect_net_dir() self._discover_constants() return self.network_name def _release_network(self): """Release the current network handle and deselect — frees the lock.""" # Close any open scenario first if self._scenario_open: try: self.api.simone_close() except Exception: pass self._scenario_open = False self._scenario_name = None # Release network topology handle if self.h_nw is not None: fn = getattr(self.api, "simone_nw_release", None) if fn: try: fn(self.h_nw, self.F) except Exception: pass self.h_nw = None # Deselect (frees network lock) try: fn = getattr(self.api, "simone_deselect", None) if fn: fn(self.F) except Exception: pass self._edit_started = False def _detect_net_dir(self): """Detect the directory for the currently selected network.""" self.net_dir = None if not self.network_name: return try: ss, nets_dir = self.api.simone_get_info( self.api.SIMONE_CURRENT_NETWORK_DIR, self.F ) if ss == self.OK and nets_dir: cand = os.path.join(str(nets_dir).strip(), self.network_name) if os.path.isdir(cand): self.net_dir = cand except Exception: pass if not self.net_dir: cand = os.path.join(self.root, "nets", self.network_name) if os.path.isdir(cand): self.net_dir = cand def list_networks(self): """List all available networks.""" networks = [] fn_start = getattr(self.api, "simone_network_list_start", None) fn_next = getattr(self.api, "simone_network_list_next", None) if fn_start and fn_next: try: ss = fn_start(self.F) if ss == self.OK: while True: r = fn_next(self.F) if not isinstance(r, (list, tuple)) or r[0] != self.OK: break name = str(r[1]).strip() if len(r) > 1 else None if name: networks.append(name) except Exception: pass # Filesystem fallback if not networks: nets_path = os.path.join(self.root, "nets") if os.path.isdir(nets_path): for d in sorted(os.listdir(nets_path)): if os.path.isdir(os.path.join(nets_path, d)): networks.append(d) return { "networks": networks, "count": len(networks), "current": self.network_name, } def select_network(self, name): """Switch to a different network. Releases the current network lock first.""" if self._dirty: return { "error": "You have unsaved network changes. " "Use save_network first, or proceed with caution.", "unsaved_changes": True, } # Release current network completely self._release_network() # Select new network self.network_name = name try: self.api.simone_select(str(name)) except Exception as e: self.network_name = None return {"error": f"simone_select('{name}') failed: {e}"} self._detect_net_dir() self._dirty = False return { "success": True, "network": self.network_name, "directory": self.net_dir, "message": f"Switched to network '{name}'.", } def _discover_constants(self): """Build lookup maps for element types and parameters.""" # Element type constants (SIMONE_API_ELEMENT_TYPE_* or SIMONE_OBJTYPE_*) type_prefixes = ["SIMONE_API_ELEMENT_TYPE_", "SIMONE_OBJTYPE_"] for name in dir(self.api): for pfx in type_prefixes: if name.startswith(pfx): short = name[len(pfx):].lower() self._element_type_map[short] = getattr(self.api, name) # Friendly aliases aliases = { "pipe": ["pipe"], "valve": ["valve", "va"], "compressor": ["compressor_station", "cs", "compressor"], "control_valve": ["control_valve", "cv"], "regulator": ["regulator", "re"], "short_element": ["short_element", "se", "short"], "measuring_station": ["measuring_station", "ms"], "non_return_valve": ["non_return_valve", "nrv"], "mixer": ["mixer", "mix"], "resistor": ["resistor", "recp"], } for friendly, keys in aliases.items(): for k in keys: if k in self._element_type_map and friendly not in self._element_type_map: self._element_type_map[friendly] = self._element_type_map[k] # Element parameter constants pfx = "SIMONE_API_ELEMENT_PARAMETER_" for name in dir(self.api): if name.startswith(pfx): short = name[len(pfx):].lower() self._param_map[short] = getattr(self.api, name) # ── Scenario type (runtype) mapping ───────────────────────── # SIMONE_OBJTYPE_* element type constants → human-readable names # From topo API section 7.4: simone_nw_element_create ELEMENT_TYPE_NAMES = { "PIPE": "pipe", "VA": "valve", "CS": "compressor station", "CV": "control valve", "NRV": "non return valve", "RE": "resistor", "RECP": "storage", "MS": "metering station", "SE": "short cut / joint", } # Human-friendly names for SIMONE_RUNTYPE_* constants RUNTYPE_NAMES = { "DYN": "Dynamic simulation", "STA": "Static (steady-state) simulation", "REC": "Reconstruction (state estimation)", "FIL": "Filter (data reconciliation)", "S_O": "Set-point optimization", "C_O": "Configuration optimization", "PER": "Periodic steady-state", "CPO": "Compressor optimization", "CPS": "Compressor scheduling", "LTO": "Long-term optimization", "VOP": "Virtual online prediction", } def _resolve_runtype(self, runtype_int): """Convert a numeric runtype to its short code and description.""" for name in dir(self.api): if name.startswith("SIMONE_RUNTYPE_"): if getattr(self.api, name) == runtype_int: code = name[len("SIMONE_RUNTYPE_"):] desc = self.RUNTYPE_NAMES.get(code, code) return code, desc return str(runtype_int), f"Unknown type ({runtype_int})" def _resolve_runtype_const(self, type_str): """Convert a type string like 'DYN' to the SIMONE_RUNTYPE_* constant value. Returns (constant_value, code, description) or raises ValueError.""" code = type_str.upper().strip() attr = f"SIMONE_RUNTYPE_{code}" val = getattr(self.api, attr, None) if val is not None: desc = self.RUNTYPE_NAMES.get(code, code) return val, code, desc # Try fuzzy match on descriptions for k, v in self.RUNTYPE_NAMES.items(): if code in k or code in v.upper(): attr2 = f"SIMONE_RUNTYPE_{k}" val2 = getattr(self.api, attr2, None) if val2 is not None: return val2, k, v available = [] for name in dir(self.api): if name.startswith("SIMONE_RUNTYPE_"): c = name[len("SIMONE_RUNTYPE_"):] available.append(c) raise ValueError( f"Unknown scenario type '{type_str}'. " f"Available: {', '.join(available) if available else ', '.join(self.RUNTYPE_NAMES.keys())}" ) def _format_simone_time(self, t): """Convert SIMONE time_t (seconds since epoch) to ISO string.""" import datetime as _dt if not t or t <= 0: return None try: return (_dt.datetime(1970, 1, 1) + _dt.timedelta(seconds=int(t))).strftime("%Y-%m-%d %H:%M") except Exception: return str(t) # ── Unit handling helpers ────────────────────────────────────── def _get_unit_info(self, obj_id, ext_id): """Get the unit abbreviation for a variable identified by obj_id/ext_id. Returns (unit_abbr, unit_descriptor) or (None, None) on failure.""" try: fn_info = getattr(self.api, "simone_varid_info", None) fn_get = getattr(self.api, "simone_get_api_default_unit", None) fn_des = getattr(self.api, "simone_des2unit", None) if not (fn_info and fn_get and fn_des): return None, None # Get unit_type for this variable res = fn_info(obj_id, ext_id) if not isinstance(res, (list, tuple)) or res[0] != self.OK: return None, None unit_type = res[-1] # last OUT param is unit_type if unit_type is None or unit_type == getattr(self.api, "SIMONE_UNKNOWN", -1): return None, None # Get current default unit descriptor for this unit_type res2 = fn_get(unit_type) if isinstance(res2, (list, tuple)): unit_desc = int(res2[-1]) else: unit_desc = int(res2) if unit_desc == 0: return None, None # Get abbreviation string from descriptor flag = getattr(self.api, "SIMONE_FLAG_INPUT_ABBR", self.F) res3 = fn_des(unit_desc, 64, flag) if isinstance(res3, (list, tuple)) and res3[0] == self.OK: abbr = str(res3[1]).strip() return abbr, unit_desc # Try without flag res3 = fn_des(unit_desc, 64, self.F) if isinstance(res3, (list, tuple)) and res3[0] == self.OK: abbr = str(res3[1]).strip() return abbr, unit_desc except Exception: pass return None, None def _resolve_unit_descriptor(self, obj_id, ext_id, unit_abbr): """Build a unit descriptor from an abbreviation string for a given variable. Returns the unit descriptor int, or SIMONE_UNIT_DEFAULT on failure.""" default = getattr(self.api, "SIMONE_UNIT_DEFAULT", 0) if not unit_abbr: return default try: fn_info = getattr(self.api, "simone_varid_info", None) fn_u2d = getattr(self.api, "simone_unit2des", None) if not (fn_info and fn_u2d): return default # Get unit_type for this variable res = fn_info(obj_id, ext_id) if not isinstance(res, (list, tuple)) or res[0] != self.OK: return default unit_type = res[-1] # Build descriptor: abbreviation must be in brackets, e.g. "[bar]" abbr_str = unit_abbr.strip() if not abbr_str.startswith("["): abbr_str = f"[{abbr_str}]" res2 = fn_u2d(unit_type, abbr_str, self.F) if isinstance(res2, (list, tuple)) and res2[0] == self.OK: return int(res2[1]) # Might be (status, unit) or just unit if isinstance(res2, int) and res2 != 0: return res2 except Exception: pass return default def _udes_to_abbr(self, udes): """Convert a unit descriptor int to a human-readable abbreviation string.""" if not udes: return None fn_des = getattr(self.api, "simone_des2unit", None) if not fn_des: return None try: flag = getattr(self.api, "SIMONE_FLAG_INPUT_ABBR", self.F) res = fn_des(int(udes), 64, flag) if isinstance(res, (list, tuple)) and res[0] == self.OK: return str(res[1]).strip() res = fn_des(int(udes), 64, self.F) if isinstance(res, (list, tuple)) and res[0] == self.OK: return str(res[1]).strip() except Exception: pass return None def _param_abbr_to_udes(self, param_const, unit_abbr): """Convert a unit abbreviation to a unit descriptor for a topology parameter. Uses simone_nw_element_get_parameter's returned udes to find the unit type, or falls back to known parameter-to-unit-type mappings.""" if not unit_abbr: return getattr(self.api, "SIMONE_UNIT_DEFAULT", 0) # Known parameter → unit type mappings param_unit_types = {} for suffix, attr in [ ("DIAMETER", "SIMONE_UNIT_TYPE_D"), ("LENGTH", "SIMONE_UNIT_TYPE_L"), ("ROUGHNESS", "SIMONE_UNIT_TYPE_RR"), ("RESISTANCE", "SIMONE_UNIT_TYPE_P"), # pressure drop for resistance ]: pkey = f"SIMONE_API_ELEMENT_PARAMETER_{suffix}" pc = getattr(self.api, pkey, None) ut = getattr(self.api, attr, None) if pc is not None and ut is not None: param_unit_types[pc] = ut unit_type = param_unit_types.get(param_const) if unit_type is None: return getattr(self.api, "SIMONE_UNIT_DEFAULT", 0) fn_u2d = getattr(self.api, "simone_unit2des", None) if not fn_u2d: return getattr(self.api, "SIMONE_UNIT_DEFAULT", 0) try: abbr_str = unit_abbr.strip() if not abbr_str.startswith("["): abbr_str = f"[{abbr_str}]" res = fn_u2d(unit_type, abbr_str, self.F) if isinstance(res, (list, tuple)) and res[0] == self.OK: return int(res[1]) if isinstance(res, int) and res != 0: return res except Exception: pass return getattr(self.api, "SIMONE_UNIT_DEFAULT", 0) # ── Editing session management ────────────────────────────── def _ensure_edit(self): """Start network editing session and load topology if needed.""" if not self._edit_started: fn = getattr(self.api, "simone_nw_start_edit", None) if fn is None: raise RuntimeError( "Network Edit Extensions not available. " "Check your SIMONE license includes the API option." ) ss = fn(self.F) if ss != self.OK: raise RuntimeError(f"simone_nw_start_edit failed (status={ss})") self._edit_started = True if self.h_nw is None: fn = getattr(self.api, "simone_nw_load_active", None) if fn is None: raise RuntimeError("simone_nw_load_active not available") result = fn(self.F) if isinstance(result, (list, tuple)): if result[0] != self.OK: raise RuntimeError(f"simone_nw_load_active failed (status={result[0]})") self.h_nw = result[1] else: raise RuntimeError(f"Unexpected return from simone_nw_load_active: {result}") def _element_type_to_name(self, type_val): """Convert a numeric element type constant to a human-readable name.""" # Try reverse lookup via discovered constants for k, v in self._element_type_map.items(): if v == type_val: # Map short key (e.g. 'va') to the ELEMENT_TYPE_NAMES description for code, desc in self.ELEMENT_TYPE_NAMES.items(): if k == code.lower(): return desc return k # Fallback — scan SIMONE_OBJTYPE_* attributes directly for name in dir(self.api): if name.startswith("SIMONE_OBJTYPE_") and getattr(self.api, name) == type_val: code = name[len("SIMONE_OBJTYPE_"):] return self.ELEMENT_TYPE_NAMES.get(code, code.lower()) return f"unknown({type_val})" def _resolve_element_type(self, type_name): """Resolve a friendly element type name to an API constant.""" key = type_name.lower().replace(" ", "_").replace("-", "_") val = self._element_type_map.get(key) if val is not None: return val # Try matching against ELEMENT_TYPE_NAMES codes (PIPE, VA, CS, etc.) for code in self.ELEMENT_TYPE_NAMES: if key == code.lower() or key in self.ELEMENT_TYPE_NAMES[code]: attr = f"SIMONE_OBJTYPE_{code}" val = getattr(self.api, attr, None) if val is not None: return val # Try partial match on discovered map for k, v in self._element_type_map.items(): if key in k or k in key: return v raise ValueError( f"Unknown element type '{type_name}'. " f"Available: {', '.join(sorted(set(self._element_type_map.keys())))}" ) def _resolve_param(self, param_name): """Resolve a friendly parameter name to an API constant.""" key = param_name.lower().replace(" ", "_").replace("-", "_") val = self._param_map.get(key) if val is not None: return val for k, v in self._param_map.items(): if key in k or k in key: return v raise ValueError( f"Unknown parameter '{param_name}'. " f"Available: {', '.join(sorted(self._param_map.keys()))}" ) # ── Query: Network info ───────────────────────────────────── def get_network_info(self): """Get current network information and statistics.""" self._ensure_edit() info = {"network": self.network_name, "directory": self.net_dir} try: ss, count = self.api.simone_nw_node_get_count(self.h_nw, self.F) if ss == self.OK: info["node_count"] = count except Exception: pass try: ss, count = self.api.simone_nw_element_get_count(self.h_nw, self.F) if ss == self.OK: info["element_count"] = count except Exception: pass try: ss, count = self.api.simone_nw_subsystem_get_count(self.h_nw, self.F) if ss == self.OK: info["subsystem_count"] = count except Exception: pass # Owner info for fn_name, key in [ ("simone_nw_statistics_get_owner", "owner"), ("simone_nw_statistics_get_network_name", "model_name"), ]: try: fn = getattr(self.api, fn_name, None) if fn: result = fn(self.h_nw, 256, self.F) if isinstance(result, (list, tuple)) and result[0] == self.OK: info[key] = str(result[1]).strip() except Exception: pass return info # ── Query: List nodes ─────────────────────────────────────── def list_nodes(self, pattern=None, limit=200): """List nodes, optionally filtered by name pattern (supports * and ?).""" self._ensure_edit() nodes = [] try: result = self.api.simone_nw_node_get_first(self.h_nw, self.F) except Exception as e: return {"error": str(e)} if not isinstance(result, (list, tuple)) or result[0] != self.OK: return {"nodes": [], "total": 0} h_node = result[1] while True: name = self._get_node_name(h_node) if name: if pattern is None or fnmatch.fnmatch(name.upper(), pattern.upper()): nodes.append(name) self._release(h_node) if len(nodes) >= limit: break # h_node is OUT param — try 2-arg call first (SWIG returns it) try: result = self.api.simone_nw_node_get_next(self.h_nw, self.F) except TypeError: try: result = self.api.simone_nw_node_get_next(self.h_nw, h_node, self.F) except Exception: break except Exception: break if not isinstance(result, (list, tuple)) or result[0] != self.OK: break h_node = result[1] return {"nodes": nodes, "count": len(nodes), "limited": len(nodes) >= limit} def _get_node_name(self, h_node): """Get name string from a node handle.""" for args in [(h_node, 256, self.F), (h_node, self.F)]: try: result = self.api.simone_nw_node_get_name(*args) if isinstance(result, (list, tuple)) and result[0] == self.OK: return str(result[1]).strip() except TypeError: continue except Exception: break return None # ── Query: Node details ───────────────────────────────────── def get_node_details(self, name): """Get all properties of a node by name.""" self._ensure_edit() h_node = self._find_node(name) if h_node is None: return {"error": f"Node '{name}' not found"} info = {"name": name} # Coordinates try: result = self.api.simone_nw_node_get_xy(h_node, self.F) if isinstance(result, (list, tuple)) and result[0] == self.OK: info["x"] = result[1] info["y"] = result[2] except Exception: pass # Height try: result = self.api.simone_nw_node_get_height(h_node, self.F) if isinstance(result, (list, tuple)) and result[0] == self.OK: info["height"] = result[1] except Exception: pass # Supply flag try: result = self.api.simone_nw_node_is_supply(h_node, self.F) if isinstance(result, (list, tuple)) and result[0] == self.OK: info["is_supply"] = bool(result[1]) except Exception: pass # Alias try: for args in [(h_node, 256, self.F), (h_node, self.F)]: try: result = self.api.simone_nw_node_get_alias(*args) if isinstance(result, (list, tuple)) and result[0] == self.OK: alias = str(result[1]).strip() if alias: info["alias"] = alias break except TypeError: continue except Exception: pass # ID try: result = self.api.simone_nw_node_get_id(h_node, self.F) if isinstance(result, (list, tuple)) and result[0] == self.OK: info["obj_id"] = result[1] except Exception: pass # Connected elements try: elements = [] result = self.api.simone_nw_node_get_first_connected_element( h_node, self.F ) while isinstance(result, (list, tuple)) and result[0] == self.OK: h_el = result[1] el_name = self._get_element_name(h_el) if el_name: elements.append(el_name) self._release(h_el) try: result = self.api.simone_nw_node_get_next_connected_element( h_node, self.F ) except TypeError: result = self.api.simone_nw_node_get_next_connected_element( h_node, h_el, self.F ) if elements: info["connected_elements"] = elements except Exception: pass self._release(h_node) return info # ── Query: List elements ──────────────────────────────────── def list_elements(self, pattern=None, element_type=None, limit=200): """List elements, optionally filtered by name pattern and/or type.""" self._ensure_edit() elements = [] type_filter = None if element_type: try: type_filter = self._resolve_element_type(element_type) except ValueError as e: return {"error": str(e)} try: result = self.api.simone_nw_element_get_first(self.h_nw, self.F) except Exception as e: return {"error": str(e)} if not isinstance(result, (list, tuple)) or result[0] != self.OK: return {"elements": [], "count": 0} h_el = result[1] while True: include = True name = self._get_element_name(h_el) # Get element type el_type_val = None el_type_name = None try: r = self.api.simone_nw_element_get_type(h_el, self.F) if isinstance(r, (list, tuple)) and r[0] == self.OK: el_type_val = r[1] el_type_name = self._element_type_to_name(el_type_val) except Exception: pass if name and pattern: if not fnmatch.fnmatch(name.upper(), pattern.upper()): include = False if include and type_filter is not None: if el_type_val is None or el_type_val != type_filter: include = False if include and name: entry = {"name": name} if el_type_name: entry["type"] = el_type_name elements.append(entry) self._release(h_el) if len(elements) >= limit: break # h_element is OUT param — SWIG returns it, so call with 2 args like get_first try: result = self.api.simone_nw_element_get_next(self.h_nw, self.F) except TypeError: # Some SWIG versions may need the old handle as IN/OUT try: result = self.api.simone_nw_element_get_next(self.h_nw, h_el, self.F) except Exception: break except Exception: break if not isinstance(result, (list, tuple)) or result[0] != self.OK: break h_el = result[1] return {"elements": elements, "count": len(elements), "limited": len(elements) >= limit} def _get_element_name(self, h_el): """Get name string from an element handle.""" for args in [(h_el, 256, self.F), (h_el, self.F)]: try: result = self.api.simone_nw_element_get_name(*args) if isinstance(result, (list, tuple)) and result[0] == self.OK: return str(result[1]).strip() except TypeError: continue except Exception: break return None # ── Query: Element details ────────────────────────────────── def get_element_details(self, name): """Get all properties of an element by name.""" self._ensure_edit() h_el = self._find_element(name) if h_el is None: return {"error": f"Element '{name}' not found"} info = {"name": name} # Type try: r = self.api.simone_nw_element_get_type(h_el, self.F) if isinstance(r, (list, tuple)) and r[0] == self.OK: type_val = r[1] info["type_id"] = type_val info["type"] = self._element_type_to_name(type_val) except Exception: pass # Start and end nodes for fn_name, key in [ ("simone_nw_element_get_start_node", "start_node"), ("simone_nw_element_get_end_node", "end_node"), ]: try: fn = getattr(self.api, fn_name) r = fn(h_el, self.F) if isinstance(r, (list, tuple)) and r[0] == self.OK: node_name = self._get_node_name(r[1]) if node_name: info[key] = node_name self._release(r[1]) except Exception: pass # Subsystem try: r = self.api.simone_nw_element_get_subsystem(h_el, self.F) if isinstance(r, (list, tuple)) and r[0] == self.OK: h_sub = r[1] sub_name = self._get_subsystem_name(h_sub) if sub_name: info["subsystem"] = sub_name self._release(h_sub) except Exception: pass # Parameters (diameter, length, roughness, etc.) — topology API for param_name, param_const in self._param_map.items(): try: r = self.api.simone_nw_element_get_parameter( h_el, param_const, self.F ) if isinstance(r, (list, tuple)) and r[0] == self.OK: info[f"param_{param_name}"] = r[1] # r[2] is unit descriptor — resolve to abbreviation if len(r) > 2 and r[2]: udes = int(r[2]) abbr = self._udes_to_abbr(udes) if abbr: info[f"param_{param_name}_unit"] = abbr else: info[f"param_{param_name}_udes"] = udes except Exception: pass # Alias try: for args in [(h_el, 256, self.F), (h_el, self.F)]: try: r = self.api.simone_nw_element_get_alias(*args) if isinstance(r, (list, tuple)) and r[0] == self.OK: alias = str(r[1]).strip() if alias: info["alias"] = alias break except TypeError: continue except Exception: pass self._release(h_el) return info # ── Query: List subsystems ────────────────────────────────── def list_subsystems(self): """List all subsystems in the network.""" self._ensure_edit() subs = [] try: result = self.api.simone_nw_subsystem_get_first(self.h_nw, self.F) except Exception as e: return {"error": str(e)} if not isinstance(result, (list, tuple)) or result[0] != self.OK: return {"subsystems": [], "count": 0} h_sub = result[1] while True: name = self._get_subsystem_name(h_sub) if name: subs.append(name) self._release(h_sub) try: result = self.api.simone_nw_subsystem_get_next(self.h_nw, self.F) except TypeError: try: result = self.api.simone_nw_subsystem_get_next(self.h_nw, h_sub, self.F) except Exception: break except Exception: break if not isinstance(result, (list, tuple)) or result[0] != self.OK: break h_sub = result[1] return {"subsystems": subs, "count": len(subs)} def _get_subsystem_name(self, h_sub): for args in [(h_sub, 256, self.F), (h_sub, self.F)]: try: r = self.api.simone_nw_subsystem_get_name(*args) if isinstance(r, (list, tuple)) and r[0] == self.OK: return str(r[1]).strip() except TypeError: continue except Exception: break return None # ── Query: List scenarios ─────────────────────────────────── def list_scenarios(self): """List scenarios available in the network directory, with type info.""" scenarios = [] # Method 1: API enumeration with detailed info try: fn_start = getattr(self.api, "simone_scenario_list_start", None) fn_next = getattr(self.api, "simone_scenario_list_next", None) fn_info = getattr(self.api, "simone_scenario_list_info", None) if fn_start and fn_next: ss = fn_start(self.F) if ss == self.OK: while True: r = fn_next(self.F) if not isinstance(r, (list, tuple)) or r[0] != self.OK: break scen_name = str(r[1]).strip() if len(r) > 1 else None if not scen_name: continue entry = {"name": scen_name} # Get detailed info right after list_next if fn_info: info = self._call_scenario_list_info(fn_info) if info: entry.update(info) scenarios.append(entry) except Exception: pass # Method 2: Filesystem fallback (no type info available) if not scenarios and self.net_dir: seen = set() for ext in [".RDF", ".rdf", ".SDF", ".sdf"]: for f in os.listdir(self.net_dir): if f.endswith(ext): name = f[: -len(ext)] if name not in seen: seen.add(name) scenarios.append({"name": name}) scenarios.sort(key=lambda s: s.get("name", "")) return {"scenarios": scenarios, "count": len(scenarios)} def _call_scenario_list_info(self, fn_info): """Call simone_scenario_list_info with SWIG argument variants.""" # C: (runtype OUT, inic OUT, inic_len IN, initime OUT, termtime OUT, # owner OUT, owner_len IN, comment OUT, comment_len IN, flags IN) # SWIG may or may not absorb buffer-size IN params for args in [(256, 256, 256, self.F), (self.F,)]: try: r = fn_info(*args) if isinstance(r, (list, tuple)) and r[0] == self.OK and len(r) >= 6: runtype_int = r[1] code, desc = self._resolve_runtype(runtype_int) info = {"type": code, "type_description": desc} # r[2] = initial_condition, r[3] = initime, r[4] = termtime inic = str(r[2]).strip() if len(r) > 2 and r[2] else "" if inic: info["initial_condition"] = inic initime = self._format_simone_time(r[3]) if len(r) > 3 else None termtime = self._format_simone_time(r[4]) if len(r) > 4 else None if initime: info["start_time"] = initime if termtime: info["end_time"] = termtime # r[5] = owner, r[6] = comment owner = str(r[5]).strip() if len(r) > 5 and r[5] else "" comment = str(r[6]).strip() if len(r) > 6 and r[6] else "" if owner: info["owner"] = owner if comment: info["comment"] = comment return info except TypeError: continue except Exception: break return None # ── Query: Scenario info ───────────────────────────────────── def get_scenario_info(self, name): """Get detailed info about a scenario (type, times, owner, comment) without opening it.""" # Method 1: simone_scenario_info (5.24) — works without opening fn = getattr(self.api, "simone_scenario_info", None) if fn: # C: (scenario IN, runtype OUT, inic_file OUT, inic_len IN, # initime OUT, termtime OUT, owner OUT, owner_len IN, # comment OUT, comment_len IN, flags IN) for args in [(str(name), 256, 256, 256, self.F), (str(name), self.F)]: try: r = fn(*args) if isinstance(r, (list, tuple)) and r[0] == self.OK and len(r) >= 5: runtype_int = r[1] code, desc = self._resolve_runtype(runtype_int) info = { "scenario": name, "type": code, "type_description": desc, } inic = str(r[2]).strip() if len(r) > 2 and r[2] else "" if inic: info["initial_condition"] = inic initime = self._format_simone_time(r[3]) if len(r) > 3 else None termtime = self._format_simone_time(r[4]) if len(r) > 4 else None if initime: info["start_time"] = initime if termtime: info["end_time"] = termtime owner = str(r[5]).strip() if len(r) > 5 and r[5] else "" comment = str(r[6]).strip() if len(r) > 6 and r[6] else "" if owner: info["owner"] = owner if comment: info["comment"] = comment # Also get calculation status try: calc = self.get_calculation_status(name) if "status" in calc: info["calculation_status"] = calc["status"] info["calculated"] = calc.get("calculated", False) except Exception: pass return info except TypeError: continue except Exception as e: return {"scenario": name, "error": f"simone_scenario_info failed: {e}"} # Method 2: Iterate scenario list to find the one we want try: fn_start = getattr(self.api, "simone_scenario_list_start", None) fn_next = getattr(self.api, "simone_scenario_list_next", None) fn_linfo = getattr(self.api, "simone_scenario_list_info", None) if fn_start and fn_next and fn_linfo: ss = fn_start(self.F) if ss == self.OK: while True: r = fn_next(self.F) if not isinstance(r, (list, tuple)) or r[0] != self.OK: break scen_name = str(r[1]).strip() if len(r) > 1 else "" if scen_name.upper() == name.upper(): linfo = self._call_scenario_list_info(fn_linfo) if linfo: linfo["scenario"] = name return linfo return {"scenario": name, "message": "Found but could not read details."} except Exception: pass # Method 3: open, read properties, close fn_props = getattr(self.api, "simone_get_properties", None) if fn_props: try: open_r = self.open_scenario_results(name) if open_r.get("success"): for args in [(256, 256), ()]: try: r = fn_props(*args) if isinstance(r, (list, tuple)) and r[0] == self.OK: runtype_int = r[1] code, desc = self._resolve_runtype(runtype_int) info = {"scenario": name, "type": code, "type_description": desc} owner = str(r[2]).strip() if len(r) > 2 and r[2] else "" inic = str(r[3]).strip() if len(r) > 3 and r[3] else "" if owner: info["owner"] = owner if inic: info["initial_condition"] = inic self.close_scenario_results() return info except TypeError: continue self.close_scenario_results() except Exception: pass return {"scenario": name, "error": "Could not retrieve scenario info. Function not available in this API version."} # ── Query: Find objects by pattern ────────────────────────── def find_objects(self, pattern): """Search for nodes and elements matching a name pattern.""" nodes_result = self.list_nodes(pattern=pattern, limit=50) elements_result = self.list_elements(pattern=pattern, limit=50) return { "matching_nodes": nodes_result.get("nodes", []), "matching_elements": elements_result.get("elements", []), } # ── Modification: Create node ─────────────────────────────── def create_node(self, name, x=0.0, y=0.0): """Create a new node at the given coordinates.""" self._ensure_edit() try: result = self.api.simone_nw_node_create( self.h_nw, float(x), float(y), str(name), self.F ) if isinstance(result, (list, tuple)): if result[0] != self.OK: return {"error": f"Failed to create node (status={result[0]})"} self._release(result[1]) elif result != self.OK: return {"error": f"Failed to create node (status={result})"} except Exception as e: return {"error": str(e)} self._dirty = True return {"success": True, "message": f"Node '{name}' created at ({x}, {y})"} # ── Modification: Remove node ─────────────────────────────── def remove_node(self, name): """Remove a node by name.""" self._ensure_edit() h_node = self._find_node(name) if h_node is None: return {"error": f"Node '{name}' not found"} try: ss = self.api.simone_nw_node_remove(h_node, self.F) if isinstance(ss, (list, tuple)): ss = ss[0] if ss != self.OK: self._release(h_node) return {"error": f"Failed to remove node (status={ss})"} except Exception as e: self._release(h_node) return {"error": str(e)} self._dirty = True return {"success": True, "message": f"Node '{name}' removed"} # ── Modification: Modify node ─────────────────────────────── def modify_node(self, name, new_name=None, x=None, y=None, height=None, is_supply=None, alias=None): """Modify properties of an existing node.""" self._ensure_edit() h_node = self._find_node(name) if h_node is None: return {"error": f"Node '{name}' not found"} changes = [] errors = [] if new_name is not None: try: ss = self.api.simone_nw_node_set_name(h_node, str(new_name), self.F) if isinstance(ss, (list, tuple)): ss = ss[0] if ss == self.OK: changes.append(f"name -> '{new_name}'") else: errors.append(f"rename failed (status={ss})") except Exception as e: errors.append(f"rename: {e}") if x is not None or y is not None: try: # Get current coords first if only one is changing cur_x, cur_y = 0.0, 0.0 r = self.api.simone_nw_node_get_xy(h_node, self.F) if isinstance(r, (list, tuple)) and r[0] == self.OK: cur_x, cur_y = r[1], r[2] new_x = float(x) if x is not None else cur_x new_y = float(y) if y is not None else cur_y ss = self.api.simone_nw_node_set_xy(h_node, new_x, new_y, self.F) if isinstance(ss, (list, tuple)): ss = ss[0] if ss == self.OK: changes.append(f"coordinates -> ({new_x}, {new_y})") else: errors.append(f"set_xy failed (status={ss})") except Exception as e: errors.append(f"set_xy: {e}") if height is not None: try: ss = self.api.simone_nw_node_set_height(h_node, float(height), self.F) if isinstance(ss, (list, tuple)): ss = ss[0] if ss == self.OK: changes.append(f"height -> {height}") else: errors.append(f"set_height failed (status={ss})") except Exception as e: errors.append(f"set_height: {e}") if is_supply is not None: try: ss = self.api.simone_nw_node_set_supply( h_node, 1 if is_supply else 0, self.F ) if isinstance(ss, (list, tuple)): ss = ss[0] if ss == self.OK: changes.append(f"is_supply -> {is_supply}") else: errors.append(f"set_supply failed (status={ss})") except Exception as e: errors.append(f"set_supply: {e}") if alias is not None: try: ss = self.api.simone_nw_node_set_alias(h_node, str(alias), self.F) if isinstance(ss, (list, tuple)): ss = ss[0] if ss == self.OK: changes.append(f"alias -> '{alias}'") else: errors.append(f"set_alias failed (status={ss})") except Exception as e: errors.append(f"set_alias: {e}") self._release(h_node) if changes: self._dirty = True result = {"changes": changes} if errors: result["errors"] = errors return result # ── Modification: Create element ──────────────────────────── def create_element(self, name, element_type, start_node, end_node, subsystem=None): """Create a new element connecting two nodes.""" self._ensure_edit() type_const = self._resolve_element_type(element_type) h_start = self._find_node(start_node) if h_start is None: return {"error": f"Start node '{start_node}' not found"} h_end = self._find_node(end_node) if h_end is None: self._release(h_start) return {"error": f"End node '{end_node}' not found"} h_sub = 0 if subsystem: h_sub_found = self._find_subsystem(subsystem) if h_sub_found is None: self._release(h_start) self._release(h_end) return {"error": f"Subsystem '{subsystem}' not found"} h_sub = h_sub_found try: result = self.api.simone_nw_element_create( self.h_nw, type_const, h_start, h_end, h_sub, str(name), self.F ) if isinstance(result, (list, tuple)): if result[0] != self.OK: return {"error": f"Failed to create element (status={result[0]})"} self._release(result[1]) # release new element handle elif result != self.OK: return {"error": f"Failed to create element (status={result})"} except Exception as e: return {"error": str(e)} finally: self._release(h_start) self._release(h_end) if h_sub: self._release(h_sub) self._dirty = True return { "success": True, "message": f"Element '{name}' ({element_type}) created: {start_node} -> {end_node}", } # ── Modification: Remove element ──────────────────────────── def remove_element(self, name): """Remove an element by name.""" self._ensure_edit() h_el = self._find_element(name) if h_el is None: return {"error": f"Element '{name}' not found"} try: ss = self.api.simone_nw_element_remove(h_el, self.F) if isinstance(ss, (list, tuple)): ss = ss[0] if ss != self.OK: self._release(h_el) return {"error": f"Failed to remove element (status={ss})"} except Exception as e: self._release(h_el) return {"error": str(e)} self._dirty = True return {"success": True, "message": f"Element '{name}' removed"} # ── Modification: Modify element ──────────────────────────── def modify_element(self, name, new_name=None, diameter=None, length=None, roughness=None, resistance=None, alias=None, subsystem=None, diameter_unit=None, length_unit=None, roughness_unit=None, resistance_unit=None): """Modify properties/parameters of an existing element (topology API). Args: name: element name new_name: rename element diameter, length, roughness, resistance: parameter values diameter_unit, length_unit, roughness_unit, resistance_unit: unit abbreviations (e.g. 'mm', 'km', 'um'). Uses unit manager defaults if omitted. alias: alias string subsystem: assign to subsystem """ self._ensure_edit() h_el = self._find_element(name) if h_el is None: return {"error": f"Element '{name}' not found"} changes = [] errors = [] if new_name is not None: try: ss = self.api.simone_nw_element_set_name(h_el, str(new_name), self.F) if isinstance(ss, (list, tuple)): ss = ss[0] if ss == self.OK: changes.append(f"name -> '{new_name}'") else: errors.append(f"rename failed (status={ss})") except Exception as e: errors.append(f"rename: {e}") # Set numeric parameters via topo API (simone_nw_element_set_parameter) param_args = { "diameter": (diameter, diameter_unit), "length": (length, length_unit), "roughness": (roughness, roughness_unit), "resistance": (resistance, resistance_unit), } for pname, (pval, punit) in param_args.items(): if pval is not None: try: pc = self._resolve_param(pname) udes = self._param_abbr_to_udes(pc, punit) ss = self.api.simone_nw_element_set_parameter( h_el, pc, float(pval), udes, self.F ) if isinstance(ss, (list, tuple)): ss = ss[0] if ss == self.OK: unit_str = f" {punit}" if punit else "" changes.append(f"{pname} -> {pval}{unit_str}") else: errors.append(f"set {pname} failed (status={ss})") except Exception as e: errors.append(f"set {pname}: {e}") if alias is not None: try: ss = self.api.simone_nw_element_set_alias(h_el, str(alias), self.F) if isinstance(ss, (list, tuple)): ss = ss[0] if ss == self.OK: changes.append(f"alias -> '{alias}'") else: errors.append(f"set_alias failed (status={ss})") except Exception as e: errors.append(f"set_alias: {e}") if subsystem is not None: h_sub = self._find_subsystem(subsystem) if h_sub is None: errors.append(f"Subsystem '{subsystem}' not found") else: try: ss = self.api.simone_nw_element_set_subsystem(h_el, h_sub, self.F) if isinstance(ss, (list, tuple)): ss = ss[0] if ss == self.OK: changes.append(f"subsystem -> '{subsystem}'") else: errors.append(f"set_subsystem failed (status={ss})") except Exception as e: errors.append(f"set_subsystem: {e}") self._release(h_sub) self._release(h_el) if changes: self._dirty = True result = {"changes": changes} if errors: result["errors"] = errors return result # ── Modification: Create subsystem ────────────────────────── def create_subsystem(self, name): """Create a new subsystem.""" self._ensure_edit() try: result = self.api.simone_nw_subsystem_create( self.h_nw, str(name), self.F ) if isinstance(result, (list, tuple)): if result[0] != self.OK: return {"error": f"Failed to create subsystem (status={result[0]})"} self._release(result[1]) elif result != self.OK: return {"error": f"Failed to create subsystem (status={result})"} except Exception as e: return {"error": str(e)} self._dirty = True return {"success": True, "message": f"Subsystem '{name}' created"} # ── Session: Save network ─────────────────────────────────── def save_network(self): """Store the current network model to edited network file.""" if self.h_nw is None: return {"error": "No network loaded for editing"} try: ss = self.api.simone_nw_store(self.h_nw, self.F) if isinstance(ss, (list, tuple)): ss = ss[0] if ss != self.OK: return {"error": f"simone_nw_store failed (status={ss})"} except Exception as e: return {"error": str(e)} self._dirty = False return { "success": True, "message": "Network saved to edited network file. " "Use activate_network to make it the active version.", } def activate_network(self): """Activate the edited network — make it the active version. Must release the network handle first (unlock), then re-acquire after.""" fn = getattr(self.api, "simone_activate_network", None) if fn is None: return {"error": "simone_activate_network not available in this SIMONE version"} # Release network handle to unlock the topology file if self.h_nw is not None: fn_rel = getattr(self.api, "simone_nw_release", None) if fn_rel: try: fn_rel(self.h_nw, self.F) except Exception: pass self.h_nw = None try: ss = fn(self.F) if isinstance(ss, (list, tuple)): ss = ss[0] if ss != self.OK: return {"error": f"simone_activate_network failed (status={ss})"} except Exception as e: return {"error": str(e)} finally: # Re-load the (now active) network for further editing self._edit_started = False try: self._ensure_edit() except Exception: pass return { "success": True, "message": "Edited network activated as the active version.", } # ── Session: Load scenario ────────────────────────────────── def load_scenario(self, name): """Open/load a scenario by name (for writing — allows save_as, execute, etc.).""" # Close any currently open scenario if self._scenario_open: try: self.api.simone_close() except Exception: pass self._scenario_open = False # Re-select network if self.network_name: try: self.api.simone_select(self.network_name) except Exception: pass # Open scenario for writing (allows save_as, execute, modifications) mode_write = getattr(self.api, "SIMONE_MODE_WRITE", None) mode_read = getattr(self.api, "SIMONE_MODE_READ", None) opened = False open_mode = None for mode in [mode_write, mode_read]: if mode is None: continue try: ss = self.api.simone_open(str(name), mode) if isinstance(ss, (list, tuple)): ss = ss[0] if ss == self.OK: opened = True open_mode = "write" if mode == mode_write else "read" break except Exception: continue if not opened: return {"error": f"Failed to open scenario '{name}'. Does it exist?"} self._scenario_open = True self._scenario_name = name return {"success": True, "message": f"Scenario '{name}' loaded ({open_mode} mode)"} # ── Session: Save scenario ────────────────────────────────── def save_scenario_as(self, name, comment="", overwrite=False, scenario_type=None, duration_hours=None, start_time=None, end_time=None): """Save current scenario with a new name, optionally changing type and time range. Args: name: new scenario name comment: optional comment overwrite: remove existing scenario with same name first scenario_type: type code like "DYN", "STA", "REC" etc. (None = keep same) duration_hours: for DYN scenarios — simulation duration in hours start_time: ISO start time (e.g. "2025-01-15 06:00"). Overrides source times. end_time: ISO end time. Alternative to duration_hours. """ import datetime as _dt # Resolve scenario type runtype_val = 0 # 0 = keep same type type_code = None if scenario_type: try: runtype_val, type_code, type_desc = self._resolve_runtype_const(scenario_type) except ValueError as e: return {"error": str(e)} # Set times on current scenario BEFORE save_as (they get copied to the new one) times_set = False if start_time or end_time or duration_hours: try: initime = None termtime = None if start_time: dt = _dt.datetime.fromisoformat(str(start_time)) initime = int(dt.timestamp()) if end_time: dt = _dt.datetime.fromisoformat(str(end_time)) termtime = int(dt.timestamp()) elif duration_hours and initime: termtime = initime + int(float(duration_hours) * 3600) elif duration_hours: # No start_time — get current configured times fn_gt = getattr(self.api, "simone_get_configured_times", None) if fn_gt is None: fn_gt = getattr(self.api, "simone_get_times", None) if fn_gt: r = fn_gt() if isinstance(r, (list, tuple)) and r[0] == self.OK: initime = int(r[1]) termtime = initime + int(float(duration_hours) * 3600) if initime and termtime: fn_st = getattr(self.api, "simone_set_times", None) if fn_st: ss_t = fn_st(initime, termtime) if isinstance(ss_t, (list, tuple)): ss_t = ss_t[0] times_set = (ss_t == self.OK) except Exception: pass # Times are optional — don't fail the save try: fn = getattr(self.api, "simone_scenario_save_as", None) if fn is None: return {"error": "simone_scenario_save_as not available"} # If overwrite requested, remove existing scenario first if overwrite: rm_fn = getattr(self.api, "simone_remove", None) rm_flag = getattr(self.api, "SIMONE_FLAG_REMOVE_ALL", self.F) if rm_fn: try: rm_fn(str(name), rm_flag) except Exception: pass # simone_scenario_save_as(new_scenario, scenario_type, inic_file, comment, flags) visible = getattr(self.api, "SIMONE_FLAG_VISIBLE", self.F) ss = fn(str(name), runtype_val, "", str(comment), visible) if isinstance(ss, (list, tuple)): ss = ss[0] if ss != self.OK: if not overwrite: return {"error": f"Failed to save scenario as '{name}' (status={ss}). Scenario may already exist — use overwrite=true."} return {"error": f"Failed to save scenario as '{name}' (status={ss})"} except Exception as e: return {"error": str(e)} self._scenario_name = name msg = f"Scenario saved as '{name}'" if type_code: msg += f" (type: {type_code})" if times_set: msg += " with updated time range" return {"success": True, "message": msg} def write_scenario_values(self, entries, rtime=0): """Write scenario parameter values. Scenario must be open in WRITE/CREATE mode. Args: entries: list of {"name": "OUT1.Q", "value": 500.0, "unit": "bar"} dicts name is . (e.g. OUT1.Q, PIPE1.BP) unit is optional — any valid SIMONE unit abbreviation for the variable's unit type; the unit manager handles conversion. rtime: time for the parameter (0 = valid from start / static) """ if not self._scenario_open: return {"error": "No scenario open. Use load_scenario first."} unit_default = getattr(self.api, "SIMONE_UNIT_DEFAULT", 0) results = [] for entry in entries: var_name = entry.get("name", "") value = entry.get("value", 0.0) unit_str = entry.get("unit", None) if not var_name: results.append({"name": var_name, "error": "empty name"}) continue # Resolve variable name to obj_id and ext_id try: # SWIG: simone_varid(varnam) → (status, obj_id, ext_id) res = self.api.simone_varid(str(var_name)) if isinstance(res, (list, tuple)) and res[0] == self.OK: obj_id = res[1] ext_id = res[2] else: results.append({"name": var_name, "error": f"variable not found (status={res[0] if isinstance(res, (list,tuple)) else res})"}) continue except Exception as e: results.append({"name": var_name, "error": f"varid failed: {e}"}) continue # Resolve unit descriptor — use explicit unit if provided, else default write_unit = self._resolve_unit_descriptor(obj_id, ext_id, unit_str) if unit_str else unit_default # Write the value try: # simone_write(rtime, obj_id, ext_id, value, unit) — all IN ss = self.api.simone_write(int(rtime), obj_id, ext_id, float(value), write_unit) if isinstance(ss, (list, tuple)): ss = ss[0] if ss == self.OK: results.append({"name": var_name, "value": value, "success": True}) else: results.append({"name": var_name, "value": value, "error": f"write failed (status={ss})"}) except Exception as e: results.append({"name": var_name, "value": value, "error": f"write error: {e}"}) n_ok = sum(1 for r in results if r.get("success")) return {"results": results, "written": n_ok, "total": len(entries)} # ── Helpers ───────────────────────────────────────────────── def _find_node(self, name): """Find a node by name and return its handle (caller must release).""" try: r = self.api.simone_nw_node_find_by_name(self.h_nw, str(name), self.F) if isinstance(r, (list, tuple)) and r[0] == self.OK: return r[1] except Exception: pass return None def _find_element(self, name): """Find an element by name and return its handle.""" try: r = self.api.simone_nw_element_find_by_name(self.h_nw, str(name), self.F) if isinstance(r, (list, tuple)) and r[0] == self.OK: return r[1] except Exception: pass return None def _find_subsystem(self, name): """Find a subsystem by name and return its handle.""" try: r = self.api.simone_nw_subsystem_find_by_name( self.h_nw, str(name), self.F ) if isinstance(r, (list, tuple)) and r[0] == self.OK: return r[1] except Exception: pass return None def _release(self, handle): """Safely release a SIMONE handle.""" if handle is None: return try: self.api.simone_nw_handle_release(handle, self.F) except Exception: pass # ── Generic API caller ───────────────────────────────────── # Registry of SIMONE API functions with their SWIG Python call signatures. # Format: "function_name": (n_args_min, n_args_max, description) # SWIG absorbs buffer-size IN params and OUT params — so Python call signatures # are shorter than the C signatures. Only IN params remain as Python arguments. # OUT params come back as return tuple elements. # Where SWIG behavior is uncertain, we list min/max arg counts to try both. API_FUNCTIONS = { # ── Ch5: Init & Scenario ── "simone_api_version": (0, 0, "Get API version number"), "simone_init": (1, 1, "Initialize API. Args: init_file(str)"), "simone_init_ex": (2, 2, "Extended init. Args: init_file(str), flags(int)"), "simone_init_setopt": (2, 2, "Set init option. Args: initopt(int), optval(int)"), "simone_get_license_status": (0, 0, "Get license status → (status, server, mode, lifetime)"), "simone_change_network_dir": (2, 2, "Change network dir. Args: path(str), flags(int)"), "simone_network_list_start": (1, 1, "Start network list iteration. Args: flags(int)"), "simone_network_list_next": (1, 1, "Get next network name. Args: flags(int) → (status, name)"), "simone_select": (1, 1, "Select network. Args: network_name(str)"), "simone_deselect": (1, 1, "Deselect network. Args: flags(int)"), "simone_open": (2, 2, "Open scenario. Args: scenario(str), mode(int: SIMONE_MODE_READ/WRITE/CREATE)"), "simone_set_properties": (3, 3, "Set scenario properties. Args: runtype(int), owner(str), inic_file(str)"), "simone_get_properties": (0, 0, "Get scenario properties → (status, runtype, owner, inic_file)"), "simone_set_inic": (3, 3, "Set initial conditions. Args: inic_file(str), inic_time(int), flags(int)"), "simone_set_scenario_comment": (1, 1, "Set comment. Args: comment(str)"), "simone_get_scenario_comment": (0, 0, "Get comment → (status, comment)"), "simone_set_simulation_defaults": (0, 0, "Set simulation defaults"), "simone_close": (0, 0, "Close current scenario"), "simone_remove": (2, 2, "Remove scenario. Args: scenario(str), flags(int)"), "simone_end": (0, 0, "End API session"), "simone_scenario_list_start": (1, 1, "Start scenario list. Args: flags(int)"), "simone_scenario_list_next": (1, 1, "Next scenario. Args: flags(int) → (status, name)"), "simone_scenario_list_info": (1, 4, "Scenario list info. Args: flags(int) → (status, runtype, inic, initime, termtime, owner, comment)"), "simone_scenario_info": (2, 5, "Scenario info. Args: scenario(str), flags(int) → (status, runtype, inic, initime, termtime, owner, comment)"), "simone_get_info": (2, 2, "Get session info. Args: mode(int), flags(int) → (status, info_str)"), "simone_scenario_save_as": (5, 5, "Save as. Args: name(str), type(int), inic(str), comment(str), flags(int)"), "simone_activate_network": (1, 1, "Activate edited network. Args: flags(int)"), # ── Ch6: Time ── "simone_set_times": (2, 2, "Set scenario times. Args: initime(int), termtime(int)"), "simone_set_times_with_flag": (3, 3, "Set times with flags. Args: initime(int), termtime(int), flags(int)"), "simone_get_configured_times": (0, 0, "Get configured times → (status, initime, termtime)"), "simone_get_times": (0, 0, "Get actual times → (status, initime, termtime)"), # ── Ch7: Objects/Names/IDs ── "simone_varid": (1, 1, "Resolve variable name. Args: varnam(str) → (status, obj_id, ext_id)"), "simone_varid_info": (2, 2, "Variable info. Args: obj_id(int), ext_id(int) → (status, obj_type, data_type, unit_type)"), "simone_varid_ex": (4, 4, "Extended varid. Args: varnam(str), obj_id(int), ext_id(int), flags(int) → (status, obj_id, ext_id)"), "simone_var2name": (2, 2, "IDs to name. Args: obj_id(int), ext_id(int) → (status, name)"), "simone_id2name": (1, 1, "ID to name. Args: id(int) → (status, name, obj_type)"), "simone_get_first_object": (2, 2, "First object. Args: req_type(int), subsys(str) → (status, obj_id, name, type, subsys)"), "simone_get_next_object": (0, 0, "Next object → (status, obj_id, name, type, subsys)"), "simone_get_object_info": (1, 1, "Object info. Args: obj_id(int) → (status, type, subsys_id, in_id, out_id, parent_id)"), "simone_define_source_name": (1, 1, "Define source. Args: name(str) → (status, src_id)"), "simone_get_next_id": (2, 2, "Next ID. Args: req_type(int), id(int) → (status, id, result_type)"), "simone_extid2name": (1, 1, "Extension ID to name. Args: ext_id(int) → (status, name)"), "simone_extname2id": (2, 2, "Extension name to ID. Args: ext_name(str), flags(int) → (status, ext_id)"), # ── Ch8: Units ── "simone_unit2des": (3, 3, "Unit type+abbr→descriptor. Args: unit_type(int), abbr(str), flags(int) → (status, descriptor)"), "simone_des2unit": (3, 3, "Descriptor→abbr. Args: descriptor(int), abbr_len(int), flags(int) → (status, abbr)"), "simone_set_api_default_unit": (2, 2, "Set default unit. Args: unit_type(int), unit(int)"), "simone_set_api_default_unit_ex": (3, 3, "Set default unit extended. Args: unit_type(int), unit(int), flags(int)"), "simone_get_api_default_unit": (1, 1, "Get default unit. Args: unit_type(int) → (status, unit)"), # ── Ch9: Reading ── "simone_set_rtime": (1, 1, "Set read time. Args: rtime(int)"), "simone_get_rtime": (0, 0, "Get read time → (status, rtime)"), "simone_next_rtime": (0, 0, "Next time step → (status, rtime)"), "simone_set_next_rtime": (1, 1, "Set next read time. Args: rtime(int)"), "simone_read": (3, 3, "Read value. Args: obj_id(int), ext_id(int), unit(int) → (status, value)"), "simone_read_str": (5, 5, "Read as string. Args: obj_id(int), ext_id(int), unit(int), width(int), precision(int) → (status, str)"), "simone_get_entry_set_filter": (2, 2, "Set entry filter. Args: flag(int), value(int)"), "simone_get_entry_reset_filter": (0, 0, "Reset entry filters"), "simone_get_entry": (0, 0, "Get scenario entry → (status, rtime, obj_id, ext_id, cond_flags, cond_id, value, value_str, unit, func_id, value_flags, src_id, comment)"), "simone_position_info": (3, 3, "Pipe position info. Args: obj_id(int), distance(float), unit(int) → (status, obj_id1, obj_id2, height, total_len)"), # ── Ch10: Writing ── "simone_write": (5, 5, "Write entry. Args: rtime(int), obj_id(int), ext_id(int), value(float), unit(int)"), "simone_write_with_flag": (6, 6, "Write with flag. Args: rtime(int), obj_id(int), ext_id(int), value(float), unit(int), flag(int)"), "simone_write_configuration": (4, 4, "Write config. Args: rtime(int), obj_id(int), ext_id(int), config_str(str)"), "simone_write_ex": (12, 12, "Write extended. Args: rtime(int), obj_id(int), ext_id(int), cond_flags(int), cond_id(int), value(float), value_str(str), unit(int), func_id(int), value_flags(int), src_id(int), comment(str)"), # ── Ch11: Merge/Erase ── "simone_include_scenario": (6, 6, "Include scenario. Args: src_network(str), src_scenario(str), start(int), end(int), src_id(int), flags(int)"), "simone_erase_entries": (4, 4, "Erase entries. Args: start(int), end(int), filter_id(int), filter_flag(int)"), "simone_erase_entry": (5, 5, "Erase single entry. Args: rtime(int), obj_id(int), ext_id(int), cond_id(int), flags(int)"), # ── Ch12: User attributes ── "simone_create_attr_def": (9, 9, "Create attr def. Args: name(str), ext_name(str), flags(int), mask(int), type(int), default(float), lower(float), upper(float), flags2(int) → (status, ext_id)"), "simone_modify_attr_def": (8, 8, "Modify attr def. Args: ext_id(int), name(str), flags(int), mask(int), default(float), lower(float), upper(float), flags2(int)"), "simone_delete_attr_def": (2, 2, "Delete attr def. Args: ext_id(int), flags(int)"), "simone_get_first_attr_def": (1, 1, "First attr def. Args: flags(int) → (status, ext_id)"), "simone_get_next_attr_def": (1, 1, "Next attr def. Args: flags(int) → (status, ext_id)"), "simone_attr_def_info": (2, 2, "Attr def info. Args: ext_id(int), flags(int) → (status, name, ext_name, flags, mask, type, default, lower, upper)"), "simone_write_attr": (3, 3, "Write attr value. Args: obj_id(int), ext_id(int), value_str(str)"), "simone_delete_attr_val": (3, 3, "Delete attr val. Args: obj_id(int), ext_id(int), flags(int)"), "simone_start_write_attr": (1, 1, "Begin attr batch. Args: flags(int)"), "simone_end_write_attr": (1, 1, "End attr batch. Args: flags(int)"), # ── Ch13: Profiles/Functions ── "simone_begin_profile": (3, 3, "Begin profile. Args: name(str), interpolation(int), prf_id(int) → (status, prf_id)"), "simone_write_profile": (3, 3, "Write profile value. Args: prf_id(int), atime(int), value(float)"), "simone_end_profile": (1, 1, "End profile. Args: prf_id(int)"), "simone_begin_read_profile": (2, 2, "Begin read profile. Args: prf_id(int), flags(int) → (status, properties)"), "simone_read_profile": (1, 1, "Read profile value. Args: prf_id(int) → (status, atime, value)"), "simone_profile_in_use": (2, 2, "Check profile usage. Args: prf_id(int), flags(int)"), "simone_delete_profile": (2, 2, "Delete profile. Args: prf_id(int), flags(int)"), "simone_rename_profile": (3, 3, "Rename profile. Args: prf_id(int), new_name(str), flags(int)"), "simone_update_profile": (5, 5, "Update profile. Args: mode(int), prf_id(int), atime(int), value(float), flags(int)"), "simone_define_function": (4, 4, "Define function. Args: mode(int), name(str), definition(str), func_id(int) → (status, func_id, func_type)"), "simone_define_function_ex": (8, 8, "Define function ext. Args: mode(int), name(str), definition(str), unit(int), comment(str), category(str), func_id(int), flags(int) → (status, func_id, func_type)"), "simone_get_function": (1, 2, "Get function. Args: func_id_or_name → (status, func_id, name, definition, func_type)"), "simone_get_function_ex": (2, 2, "Get function ext. Args: func_id(int), flags(int) → (status, func_id, name, definition, unit, comment, category, func_type)"), "simone_remove_function": (2, 2, "Remove function. Args: func_id(int), flags(int)"), "simone_rename_function": (3, 3, "Rename function. Args: func_id(int), name(str), flags(int)"), # ── Ch14: Object Sets ── "simone_create_object_set": (2, 2, "Create object set. Args: name(str), flags(int) → (status, set_id)"), "simone_delete_object_set": (2, 2, "Delete object set. Args: set_id(int), flags(int)"), "simone_add_to_object_set": (3, 3, "Add to set. Args: set_id(int), obj_id(int), flags(int)"), "simone_make_path_from_object_set": (3, 3, "Make path from set. Args: set_id(int), slip_factor(float), flags(int)"), "simone_flood_area": (4, 4, "Flood area. Args: type(int), set_id(int), obj_id(int), flags(int)"), "simone_get_next_id_from_set": (3, 3, "Next ID from set. Args: set_id(int), obj_id(int), flags(int) → (status, obj_id)"), # ── Ch15: Calculation/Messages ── "simone_execute": (0, 0, "Execute scenario → (status, status_txt)"), "simone_execute_ex": (1, 1, "Execute ext. Args: flags(int) → (status, status_txt)"), "simone_calculation_status": (0, 0, "Calc status (open scenario) → (status, status_txt)"), "simone_calculation_status_ex": (2, 2, "Calc status ext. Args: scenario(str), flags(int) → (status, status_txt)"), "simone_set_message_filter": (2, 2, "Set msg filter. Args: obj_name(str), msg_name(str)"), "simone_get_first_message": (0, 0, "First message → (status, msg, time, severity, obj_name, msg_name)"), "simone_get_next_message": (0, 0, "Next message → (status, msg, time, severity, obj_name, msg_name)"), "simone_write_message": (4, 4, "Write message. Args: status_code(int), category(str), message(str), flags(int)"), # ── Ch16: Time Conversion ── "simone_time_ansi2simone": (2, 2, "ANSI→SIMONE time. Args: atime(int), initime(int) → float_time"), "simone_time_simone2ansi": (4, 4, "SIMONE→ANSI time. Args: day(int), month(int), year(int), ftime(float) → time_t"), "simone_time_string2float": (1, 1, "String→float time. Args: str_time(str) → float_time"), "simone_date_string2int": (1, 1, "Date string→int. Args: str_date(str) → (status, day, month, year)"), "simone_datetime_string2ansi": (2, 2, "DateTime string→ANSI. Args: str_date(str), str_time(str) → time_t"), "simone_time_float2string": (3, 3, "Float→string. Args: ftime(float), str_len(int), flags(int) → (status, str)"), "simone_time_ansi2int": (1, 1, "ANSI→int components. Args: atime(int) → (day, month, year, hour, minute, second)"), "simone_time_int2ansi": (6, 6, "Int→ANSI. Args: day(int), month(int), year(int), hour(int), minute(int), second(int) → time_t"), # ── Ch17: Configuration ── "simone_get_config_item": (2, 2, "Get config. Args: section(str), name(str) → (status, value)"), "simone_set_config_item": (3, 3, "Set config. Args: section(str), name(str), value(str)"), # ── Ch18: Request Lists ── "simone_start_req_list": (1, 1, "Start request list. Args: required_type(int) → (status, attached_type, import_export)"), "simone_get_req_item": (0, 0, "Get request item → (status, obj_type, obj_id, ext_id, scada_id, value, flags)"), "simone_get_req_item_ex": (0, 0, "Get request item ext → (status, obj_type, obj_id, ext_id, scada_id, value, info, flags)"), "simone_get_req_times": (1, 1, "Get request times. Args: req_type(int) → (status, initime, termtime, cycle_time)"), # ── Ch19: Errors/Logging ── "simone_last_error": (0, 0, "Get last error → (status, message)"), "simone_set_log_level": (2, 2, "Set log level. Args: log_level(int), log_file_size(int)"), "simone_reset_log_level": (0, 0, "Reset log level"), "simone_get_log_level": (0, 0, "Get log level → (status, level, file_size)"), # ── Topo Ch2: General/Network ── "simone_nw_start_edit": (1, 1, "Start edit session. Args: flags(int)"), "simone_nw_load_active": (1, 1, "Load active network. Args: flags(int) → (status, h_nw)"), "simone_nw_load_edited": (1, 1, "Load edited network. Args: flags(int) → (status, h_nw)"), "simone_nw_store": (2, 2, "Store network. Args: h_nw(int), flags(int)"), "simone_nw_remove_edited": (1, 1, "Remove edited file. Args: flags(int)"), "simone_nw_reset": (2, 2, "Reset network. Args: h_nw(int), flags(int)"), "simone_nw_release": (2, 2, "Release network. Args: h_nw(int), flags(int)"), "simone_nw_set_unit_length": (3, 3, "Set unit length. Args: h_nw(int), unit_length(int), flags(int)"), "simone_nw_get_unit_length": (2, 2, "Get unit length. Args: h_nw(int), flags(int) → (status, unit_length)"), # ── Topo Ch3: Handles ── "simone_nw_handle_release": (2, 2, "Release handle. Args: handle(int), flags(int)"), "simone_nw_handle_addref": (2, 2, "Add ref to handle. Args: handle(int), flags(int)"), "simone_nw_handle_get_type": (2, 2, "Get handle type. Args: handle(int), flags(int) → (status, type)"), # ── Topo Ch4: Statistics ── "simone_nw_statistics_get_owner": (3, 3, "Get owner. Args: h_nw(int), buf_len(int), flags(int) → (status, owner)"), "simone_nw_statistics_get_owner_at_creation": (3, 3, "Get creator. Args: h_nw(int), buf_len(int), flags(int) → (status, owner)"), "simone_nw_statistics_get_network_name": (3, 3, "Get network name. Args: h_nw(int), buf_len(int), flags(int) → (status, name)"), "simone_nw_statistics_get_network_name_at_creation": (3, 3, "Get original name. Args: h_nw(int), buf_len(int), flags(int) → (status, name)"), "simone_nw_statistics_get_created_at": (2, 2, "Get creation time. Args: h_nw(int), flags(int) → (status, timestamp)"), "simone_nw_statistics_get_last_saved_at": (2, 2, "Get last save. Args: h_nw(int), flags(int) → (status, timestamp)"), "simone_nw_statistics_get_update_count": (2, 2, "Get update count. Args: h_nw(int), flags(int) → (status, count)"), # ── Topo Ch5: Subsystems ── "simone_nw_subsystem_get_count": (2, 2, "Subsystem count. Args: h_nw(int), flags(int) → (status, count)"), "simone_nw_subsystem_get_first": (2, 2, "First subsystem. Args: h_nw(int), flags(int) → (status, h_sub)"), "simone_nw_subsystem_get_next": (2, 3, "Next subsystem. Args: h_nw(int), [h_prev(int),] flags(int) → (status, h_sub)"), "simone_nw_subsystem_create": (3, 3, "Create subsystem. Args: h_nw(int), name(str), flags(int) → (status, h_sub)"), "simone_nw_subsystem_get_id": (2, 2, "Subsystem ID. Args: h_sub(int), flags(int) → (status, obj_id)"), "simone_nw_subsystem_set_name": (3, 3, "Set subsystem name. Args: h_sub(int), name(str), flags(int)"), "simone_nw_subsystem_get_name": (2, 3, "Get subsystem name. Args: h_sub(int), [buf_len(int),] flags(int) → (status, name)"), "simone_nw_subsystem_find_by_id": (3, 3, "Find subsystem by ID. Args: h_nw(int), obj_id(int), flags(int) → (status, h_sub)"), "simone_nw_subsystem_find_by_name": (3, 3, "Find subsystem by name. Args: h_nw(int), name(str), flags(int) → (status, h_sub)"), # ── Topo Ch6: Nodes ── "simone_nw_node_get_count": (2, 2, "Node count. Args: h_nw(int), flags(int) → (status, count)"), "simone_nw_node_get_first": (2, 2, "First node. Args: h_nw(int), flags(int) → (status, h_node)"), "simone_nw_node_get_next": (2, 3, "Next node. Args: h_nw(int), [h_prev(int),] flags(int) → (status, h_node)"), "simone_nw_node_create": (5, 5, "Create node. Args: h_nw(int), x(float), y(float), name(str), flags(int) → (status, h_node)"), "simone_nw_node_remove": (2, 2, "Remove node. Args: h_node(int), flags(int)"), "simone_nw_node_find_by_id": (3, 3, "Find node by ID. Args: h_nw(int), obj_id(int), flags(int) → (status, h_node)"), "simone_nw_node_find_by_name": (3, 3, "Find node by name. Args: h_nw(int), name(str), flags(int) → (status, h_node)"), "simone_nw_node_get_id": (2, 2, "Node ID. Args: h_node(int), flags(int) → (status, obj_id)"), "simone_nw_node_set_name": (3, 3, "Set node name. Args: h_node(int), name(str), flags(int)"), "simone_nw_node_get_name": (2, 3, "Get node name. Args: h_node(int), [buf_len(int),] flags(int) → (status, name)"), "simone_nw_node_set_alias": (3, 3, "Set node alias. Args: h_node(int), alias(str), flags(int)"), "simone_nw_node_get_alias": (2, 3, "Get node alias. Args: h_node(int), [buf_len(int),] flags(int) → (status, alias)"), "simone_nw_node_is_supply": (2, 2, "Check supply. Args: h_node(int), flags(int) → (status, is_supply)"), "simone_nw_node_set_supply": (3, 3, "Set supply. Args: h_node(int), is_supply(int), flags(int)"), "simone_nw_node_set_xy": (4, 4, "Set coords. Args: h_node(int), x(float), y(float), flags(int)"), "simone_nw_node_get_xy": (2, 2, "Get coords. Args: h_node(int), flags(int) → (status, x, y)"), "simone_nw_node_set_height": (3, 3, "Set height. Args: h_node(int), height(float), flags(int)"), "simone_nw_node_get_height": (2, 2, "Get height. Args: h_node(int), flags(int) → (status, height)"), "simone_nw_node_is_knee": (2, 2, "Check knee. Args: h_node(int), flags(int) → (status, is_knee)"), "simone_nw_node_set_isknee": (3, 3, "Set knee. Args: h_node(int), is_knee(int), flags(int)"), "simone_nw_node_get_first_connected_element": (2, 2, "First connected element. Args: h_node(int), flags(int) → (status, h_element)"), "simone_nw_node_get_next_connected_element": (2, 3, "Next connected element. Args: h_node(int), [h_prev(int),] flags(int) → (status, h_element)"), # ── Topo Ch7: Elements ── "simone_nw_element_get_count": (2, 2, "Element count. Args: h_nw(int), flags(int) → (status, count)"), "simone_nw_element_get_first": (2, 2, "First element. Args: h_nw(int), flags(int) → (status, h_element)"), "simone_nw_element_get_next": (2, 3, "Next element. Args: h_nw(int), [h_prev(int),] flags(int) → (status, h_element)"), "simone_nw_element_create": (7, 7, "Create element. Args: h_nw(int), type(int), h_start(int), h_end(int), h_sub(int), name(str), flags(int) → (status, h_element)"), "simone_nw_element_remove": (2, 2, "Remove element. Args: h_element(int), flags(int)"), "simone_nw_element_get_id": (2, 2, "Element ID. Args: h_element(int), flags(int) → (status, obj_id)"), "simone_nw_element_get_type": (2, 2, "Element type. Args: h_element(int), flags(int) → (status, type)"), "simone_nw_element_add_knee": (4, 4, "Add knee. Args: h_element(int), index(int), h_knee(int), flags(int)"), "simone_nw_element_remove_knee": (3, 3, "Remove knee. Args: h_element(int), h_knee(int), flags(int) → (status, index)"), "simone_nw_element_set_symbol_position": (3, 3, "Set symbol pos. Args: h_element(int), position(float), flags(int)"), "simone_nw_element_get_symbol_position": (2, 2, "Get symbol pos. Args: h_element(int), flags(int) → (status, position)"), "simone_nw_element_get_start_node": (2, 2, "Start node. Args: h_element(int), flags(int) → (status, h_node)"), "simone_nw_element_get_end_node": (2, 2, "End node. Args: h_element(int), flags(int) → (status, h_node)"), "simone_nw_element_get_first_node": (2, 2, "First node (incl knees). Args: h_element(int), flags(int) → (status, h_node)"), "simone_nw_element_get_next_node": (2, 3, "Next node (incl knees). Args: h_element(int), [h_prev(int),] flags(int) → (status, h_node)"), "simone_nw_element_get_subsystem": (2, 2, "Get subsystem. Args: h_element(int), flags(int) → (status, h_sub)"), "simone_nw_element_set_subsystem": (3, 3, "Set subsystem. Args: h_element(int), h_sub(int), flags(int)"), "simone_nw_element_set_name": (3, 3, "Set element name. Args: h_element(int), name(str), flags(int)"), "simone_nw_element_get_name": (2, 3, "Get element name. Args: h_element(int), [buf_len(int),] flags(int) → (status, name)"), "simone_nw_element_find_by_id": (3, 3, "Find element by ID. Args: h_nw(int), obj_id(int), flags(int) → (status, h_element)"), "simone_nw_element_find_by_name": (3, 3, "Find element by name. Args: h_nw(int), name(str), flags(int) → (status, h_element)"), "simone_nw_element_set_alias": (3, 3, "Set element alias. Args: h_element(int), alias(str), flags(int)"), "simone_nw_element_get_alias": (2, 3, "Get element alias. Args: h_element(int), [buf_len(int),] flags(int) → (status, alias)"), "simone_nw_element_set_parameter": (5, 5, "Set parameter. Args: h_element(int), parameter(int), value(float), udes(int), flags(int)"), "simone_nw_element_get_parameter": (3, 4, "Get parameter. Args: h_element(int), parameter(int), [udes(int),] flags(int) → (status, value, udes)"), "simone_nw_element_set_symbol_size_factor": (3, 3, "Set symbol size. Args: h_element(int), factor(float), flags(int)"), "simone_nw_element_get_symbol_size_factor": (2, 2, "Get symbol size. Args: h_element(int), flags(int) → (status, factor)"), "simone_nw_element_set_pipe_width_factor": (3, 3, "Set pipe width. Args: h_element(int), factor(float), flags(int)"), "simone_nw_element_get_pipe_width_factor": (2, 2, "Get pipe width. Args: h_element(int), flags(int) → (status, factor)"), "simone_nw_element_set_pipe_width_mode": (3, 3, "Set pipe width mode. Args: h_element(int), mode(int), flags(int)"), "simone_nw_element_get_pipe_width_mode": (2, 2, "Get pipe width mode. Args: h_element(int), flags(int) → (status, mode)"), # ── Topo Ch8: Layers ── "simone_nw_layer_get_first": (2, 2, "First layer. Args: h_nw(int), flags(int) → (status, h_layer)"), "simone_nw_layer_get_next": (2, 3, "Next layer. Args: h_nw(int), [h_prev(int),] flags(int) → (status, h_layer)"), "simone_nw_layer_create": (3, 3, "Create layer. Args: h_nw(int), name(str), flags(int) → (status, h_layer)"), "simone_nw_layer_remove": (2, 2, "Remove layer. Args: h_layer(int), flags(int)"), "simone_nw_layer_get_slot": (2, 2, "Layer slot. Args: h_layer(int), flags(int) → (status, slot)"), "simone_nw_layer_set_name": (3, 3, "Set layer name. Args: h_layer(int), name(str), flags(int)"), "simone_nw_layer_get_name": (2, 3, "Get layer name. Args: h_layer(int), [buf_len(int),] flags(int) → (status, name)"), "simone_nw_layer_set_visibility_mode": (3, 3, "Set visibility. Args: h_layer(int), mode(int), flags(int)"), "simone_nw_layer_get_visibility_mode": (2, 2, "Get visibility. Args: h_layer(int), flags(int) → (status, mode)"), "simone_nw_layer_set_min_zoom": (3, 3, "Set min zoom. Args: h_layer(int), min_zoom(float), flags(int)"), "simone_nw_layer_get_min_zoom": (2, 2, "Get min zoom. Args: h_layer(int), flags(int) → (status, min_zoom)"), "simone_nw_layer_set_max_zoom": (3, 3, "Set max zoom. Args: h_layer(int), max_zoom(float), flags(int)"), "simone_nw_layer_get_max_zoom": (2, 2, "Get max zoom. Args: h_layer(int), flags(int) → (status, max_zoom)"), # ── Topo Ch12: Gas defaults ── "simone_nw_gas_defval_set_adiabatic_coefficient": (3, 3, "Set adiabatic coeff. Args: h_nw(int), value(float), flags(int)"), "simone_nw_gas_defval_get_adiabatic_coefficient": (2, 2, "Get adiabatic coeff. Args: h_nw(int), flags(int) → (status, value)"), "simone_nw_gas_defval_set_viscosity": (3, 3, "Set viscosity. Args: h_nw(int), value(float), flags(int)"), "simone_nw_gas_defval_get_viscosity": (2, 2, "Get viscosity. Args: h_nw(int), flags(int) → (status, value)"), "simone_nw_gas_defval_set_basic_temperature": (4, 4, "Set basic temp. Args: h_nw(int), temp(float), udes(int), flags(int)"), "simone_nw_gas_defval_get_basic_temperature": (3, 3, "Get basic temp. Args: h_nw(int), udes(int), flags(int) → (status, temp)"), "simone_nw_gas_defval_set_basic_pressure": (4, 4, "Set basic pressure. Args: h_nw(int), pressure(float), udes(int), flags(int)"), "simone_nw_gas_defval_get_basic_pressure": (3, 3, "Get basic pressure. Args: h_nw(int), udes(int), flags(int) → (status, pressure)"), # ── Topo Ch13: Quality Tracking ── "simone_nw_qt_set_tracking_style": (3, 3, "Set QT style. Args: h_nw(int), style(int), flags(int)"), "simone_nw_qt_get_tracking_style": (2, 2, "Get QT style. Args: h_nw(int), flags(int) → (status, style)"), "simone_nw_qt_set_hydrate_formation_risk": (3, 3, "Set HFR. Args: h_nw(int), hfr(int), flags(int)"), "simone_nw_qt_get_hydrate_formation_risk": (2, 2, "Get HFR. Args: h_nw(int), flags(int) → (status, hfr)"), "simone_nw_qt_set_hires_factor": (3, 3, "Set hires factor. Args: h_nw(int), factor(int), flags(int)"), "simone_nw_qt_get_hires_factor": (2, 2, "Get hires factor. Args: h_nw(int), flags(int) → (status, factor)"), # ── Topo Ch14: Gas Quality ── "simone_nw_gasqual_get_count": (2, 2, "Gas quality count. Args: h_nw(int), flags(int) → (status, count)"), "simone_nw_gasqual_get_first": (2, 2, "First gas quality. Args: h_nw(int), flags(int) → (status, h_gasqual)"), "simone_nw_gasqual_get_next": (2, 3, "Next gas quality. Args: h_nw(int), [h_prev(int),] flags(int) → (status, h_gasqual)"), "simone_nw_gasqual_create": (5, 5, "Create gas quality. Args: h_nw(int), name(str), description(str), default_value(float), flags(int) → (status, h_gasqual)"), "simone_nw_gasqual_remove": (2, 2, "Remove gas quality. Args: h_gasqual(int), flags(int)"), "simone_nw_gasqual_get_id": (2, 2, "Gas quality ID. Args: h_gasqual(int), flags(int) → (status, id)"), "simone_nw_gasqual_get_name": (2, 3, "Gas quality name. Args: h_gasqual(int), [buf_len(int),] flags(int) → (status, name)"), "simone_nw_gasqual_find_by_id": (3, 3, "Find GQ by ID. Args: h_nw(int), id(int), flags(int) → (status, h_gasqual)"), "simone_nw_gasqual_find_by_name": (3, 3, "Find GQ by name. Args: h_nw(int), name(str), flags(int) → (status, h_gasqual)"), "simone_nw_gasqual_get_type": (2, 2, "Gas quality type. Args: h_gasqual(int), flags(int) → (status, type)"), "simone_nw_gasqual_set_default_value": (4, 4, "Set GQ default. Args: h_gasqual(int), value(float), udes(int), flags(int)"), "simone_nw_gasqual_get_default_value": (3, 3, "Get GQ default. Args: h_gasqual(int), udes(int), flags(int) → (status, value)"), "simone_nw_gasqual_set_computed": (3, 3, "Set GQ computed. Args: h_gasqual(int), computed(int), flags(int)"), "simone_nw_gasqual_is_computed": (2, 2, "Check GQ computed. Args: h_gasqual(int), flags(int) → (status, computed)"), "simone_nw_gasqual_set_description": (3, 3, "Set GQ description. Args: h_gasqual(int), description(str), flags(int)"), "simone_nw_gasqual_get_description": (2, 3, "Get GQ description. Args: h_gasqual(int), [buf_len(int),] flags(int) → (status, description)"), "simone_nw_node_get_gas_quality_value": (3, 3, "Node GQ value handle. Args: h_node(int), h_gasqual(int), flags(int) → (status, h_gqval)"), "simone_nw_node_gas_quality_value_get_id": (2, 2, "Node GQ value ID. Args: h_gqval(int), flags(int) → (status, gq_id)"), "simone_nw_node_gas_quality_value_set_float": (4, 4, "Set node GQ float. Args: h_gqval(int), value(float), udes(int), flags(int)"), "simone_nw_node_gas_quality_value_get_float": (3, 3, "Get node GQ float. Args: h_gqval(int), udes(int), flags(int) → (status, value)"), "simone_nw_node_gas_quality_value_set_computed": (3, 3, "Set node GQ computed. Args: h_gqval(int), computed(int), flags(int)"), "simone_nw_node_gas_quality_value_is_computed": (2, 2, "Check node GQ computed. Args: h_gqval(int), flags(int) → (status, computed)"), "simone_nw_node_gas_quality_value_get_first": (2, 2, "First node GQ value. Args: h_node(int), flags(int) → (status, h_gqval)"), "simone_nw_node_gas_quality_value_get_next": (2, 2, "Next node GQ value. Args: h_node(int), flags(int) → (status, h_gqval)"), # ── Topo Ch15: Reference Conditions ── "simone_nw_refcond_metric_set": (3, 3, "Set metric refcond. Args: h_nw(int), ref_cond(int), flags(int)"), "simone_nw_refcond_metric_get": (2, 2, "Get metric refcond. Args: h_nw(int), flags(int) → (status, ref_cond)"), "simone_nw_refcond_imperial_set": (3, 3, "Set imperial refcond. Args: h_nw(int), ref_cond(int), flags(int)"), "simone_nw_refcond_imperial_get": (2, 2, "Get imperial refcond. Args: h_nw(int), flags(int) → (status, ref_cond)"), "simone_nw_refcond_btu_set": (3, 3, "Set BTU refcond. Args: h_nw(int), ref_cond(int), flags(int)"), "simone_nw_refcond_btu_get": (2, 2, "Get BTU refcond. Args: h_nw(int), flags(int) → (status, ref_cond)"), } def call_simone_api(self, function_name, args=None): """Call any SIMONE API function by name with given arguments. Args: function_name: exact SIMONE API function name (e.g. 'simone_read', 'simone_nw_node_get_count') args: list of arguments to pass to the function Returns: dict with either result or error """ if args is None: args = [] # Validate function name fn = getattr(self.api, function_name, None) if fn is None: # Check if it exists with different casing available = [n for n in dir(self.api) if n.lower() == function_name.lower()] if available: return {"error": f"Function '{function_name}' not found. Did you mean '{available[0]}'?"} return {"error": f"Function '{function_name}' not found in SIMONE API"} # Auto-substitute special argument values coerced_args = [] for a in args: if isinstance(a, str): # Resolve SIMONE_* constants passed as strings if a.startswith("SIMONE_"): const_val = getattr(self.api, a, None) if const_val is not None: coerced_args.append(const_val) continue # Try numeric conversion for string numbers try: if "." in a: coerced_args.append(float(a)) else: coerced_args.append(int(a)) continue except ValueError: pass coerced_args.append(a) info = self.API_FUNCTIONS.get(function_name) # Auto-supply h_nw for topo functions if first arg is missing if (self.h_nw is not None and function_name.startswith("simone_nw_") and info and len(coerced_args) < info[0]): # Many topo functions take h_nw as first arg — auto-supply it if info[2] and "h_nw" in info[2].split(".")[0]: coerced_args = [self.h_nw] + coerced_args try: result = fn(*coerced_args) except TypeError as e: # If arg count is wrong and we have min/max info, try alternatives err_msg = str(e) if info and info[0] != info[1]: # Try with different arg counts for n in range(info[0], info[1] + 1): if n == len(coerced_args): continue trial_args = coerced_args[:n] if n < len(coerced_args) else coerced_args + [self.F] * (n - len(coerced_args)) try: result = fn(*trial_args) break except TypeError: continue else: return {"error": f"TypeError calling {function_name}: {err_msg}", "hint": f"Expected {info[0]}-{info[1]} args, got {len(coerced_args)}. {info[2]}"} else: hint = info[2] if info else "Check the API_FUNCTIONS registry for correct arguments" return {"error": f"TypeError calling {function_name}: {err_msg}", "hint": hint} except Exception as e: return {"error": f"Error calling {function_name}: {e}"} # Format result if isinstance(result, (list, tuple)): formatted = [] for v in result: if isinstance(v, (int, float, str, bool, type(None))): formatted.append(v) else: formatted.append(str(v)) # Check status status_ok = (len(formatted) > 0 and formatted[0] == self.OK) return { "function": function_name, "status_ok": status_ok, "result": formatted, } elif isinstance(result, (int, float)): return {"function": function_name, "result": result} else: return {"function": function_name, "result": str(result)} def list_api_functions(self, category=None): """List available API functions, optionally filtered by category keyword.""" results = {} for fname, (nmin, nmax, desc) in self.API_FUNCTIONS.items(): if category: if category.lower() not in fname.lower() and category.lower() not in desc.lower(): continue # Check if actually available in the loaded API available = hasattr(self.api, fname) results[fname] = { "available": available, "args": f"{nmin}" if nmin == nmax else f"{nmin}-{nmax}", "description": desc, } return {"functions": results, "count": len(results)} # ── Scenario Results ─────────────────────────────────────── def execute_scenario(self, name): """Execute/calculate a scenario. Blocking — waits until simulation completes.""" if self._scenario_open: try: self.api.simone_close() except Exception: pass self._scenario_open = False # Re-select network if self.network_name: try: self.api.simone_select(self.network_name) except Exception: pass # Open scenario — try WRITE mode first (needed for execution), fall back to READ mode_write = getattr(self.api, "SIMONE_MODE_WRITE", None) mode_read = getattr(self.api, "SIMONE_MODE_READ", None) opened = False for mode in [mode_write, mode_read]: if mode is None: continue try: ss = self.api.simone_open(str(name), mode) if isinstance(ss, (list, tuple)): ss = ss[0] if ss == self.OK: opened = True break except Exception: continue if not opened: return {"error": f"Cannot open scenario '{name}' for execution. Does it exist?"} self._scenario_open = True self._scenario_name = name # Execute execute_fn = getattr(self.api, "simone_execute", None) execute_ex_fn = getattr(self.api, "simone_execute_ex", None) if execute_fn is None and execute_ex_fn is None: return {"error": "simone_execute not available in this API version"} try: if execute_ex_fn: # SWIG: simone_execute_ex(flags) → (status, status_txt) res = execute_ex_fn(self.F) else: # SWIG: simone_execute() → (status, status_txt) res = execute_fn() if isinstance(res, (list, tuple)): ss = res[0] status_txt = str(res[1]).strip() if len(res) > 1 else "" else: ss = res status_txt = "" if ss == self.OK or status_txt == "RUNOK": # After execution, scenario is in READ mode return { "success": True, "scenario": name, "status": status_txt, "message": f"Scenario '{name}' calculated successfully. Results are now available.", } else: return { "success": False, "scenario": name, "status": status_txt, "api_status": str(ss), "message": f"Execution finished with status: {status_txt}", } except Exception as e: return {"error": f"Execution failed: {e}"} def get_calculation_status(self, name): """Check the calculation status of a scenario (without opening it).""" fn = getattr(self.api, "simone_calculation_status_ex", None) if fn is None: fn2 = getattr(self.api, "simone_calculation_status", None) if fn2 is None: return {"error": "simone_calculation_status not available"} # simone_calculation_status requires scenario to be open return {"error": "Only simone_calculation_status available (requires open scenario)"} try: # SWIG: simone_calculation_status_ex(scenario, flags) → (status, status_txt) res = fn(str(name), 0) if isinstance(res, (list, tuple)): ss = res[0] status_txt = str(res[1]).strip() if len(res) > 1 else "" else: ss = res status_txt = "" calculated = (status_txt == "RUNOK") return { "scenario": name, "status": status_txt, "calculated": calculated, } except Exception as e: return {"error": str(e)} def open_scenario_results(self, name): """Open a scenario for reading simulation results.""" if self._scenario_open: try: self.api.simone_close() except Exception: pass self._scenario_open = False mode = getattr(self.api, "SIMONE_MODE_READ", None) if mode is None: return {"error": "SIMONE_MODE_READ not available — API may not support result reading"} # Re-select network before opening (required by SIMONE API) if self.network_name: try: self.api.simone_select(self.network_name) except Exception: pass try: ss = self.api.simone_open(str(name), mode) if isinstance(ss, (list, tuple)): ss = ss[0] if ss != self.OK: return {"error": f"simone_open('{name}') failed (status={ss}). Is the scenario name correct?"} except Exception as e: return {"error": str(e)} self._scenario_open = True self._scenario_name = name return {"success": True, "message": f"Scenario '{name}' opened for reading results."} def close_scenario_results(self): """Close the currently open scenario.""" if not self._scenario_open: return {"message": "No scenario is open."} try: self.api.simone_close() except Exception: pass self._scenario_open = False name = self._scenario_name self._scenario_name = None return {"success": True, "message": f"Scenario '{name}' closed."} def _init_scenario_vars(self): """Register at least one variable so simone_get_entry can iterate. Must be called after simone_open, before simone_get_entry. Returns (node_names, node_to_ids, debug_info).""" debug = {} # Find objects in the scenario node_names = [] objtype = getattr(self.api, "SIMONE_OBJTYPE_NO", None) if objtype is None: # Fallback — try ALL objtype = getattr(self.api, "SIMONE_OBJTYPE_ALL", 0) try: res = self.api.simone_get_first_object(objtype, "") while isinstance(res, (list, tuple)) and res[0] == self.OK: oname = str(res[2]).strip() if len(res) > 2 else "" if oname: node_names.append(oname) if len(node_names) >= 50: break res = self.api.simone_get_next_object() except Exception as e: debug["get_objects_error"] = str(e) debug["n_objects_found"] = len(node_names) if not node_names: return node_names, {}, debug # Resolve variable IDs (required before simone_get_entry works) var_names = [n + ".Q" for n in node_names] nvars = len(var_names) node_to_ids = {} try: ids = self.api.new_intArray(nvars) exts = self.api.new_intArray(nvars) stats_arr = self.api.new_intArray(nvars) ss = self.api.simone_varid_array_python(var_names, ids, exts, stats_arr) debug["varid_status"] = str(ss) if ss == self.OK: for i, name in enumerate(node_names): st = self.api.intArray_getitem(stats_arr, i) if st == self.OK: oid = self.api.intArray_getitem(ids, i) eid = self.api.intArray_getitem(exts, i) node_to_ids[name] = (oid, eid) debug["n_vars_resolved"] = len(node_to_ids) except Exception as e: debug["varid_error"] = str(e) return node_names, node_to_ids, debug def _collect_rtimes(self, use_filter=True, limit=100): """Iterate simone_get_entry to collect available rtimes.""" rtimes = [] debug_info = {"filter_used": False, "entries_scanned": 0, "first_entry_raw": None} filter_fn = getattr(self.api, "simone_get_entry_set_filter", None) reset_fn = getattr(self.api, "simone_get_entry_reset_filter", None) if use_filter and filter_fn: try: filter_fn( self.api.SIMONE_FLAG_FILTER_PARAM_TYPE, self.api.SIMONE_PARAM_TYPE_SUPPLY_OFFTAKE, ) debug_info["filter_used"] = True except Exception as e: debug_info["filter_error"] = str(e) try: self.api.simone_set_rtime(0) while True: res = self.api.simone_get_entry() if debug_info["first_entry_raw"] is None: debug_info["first_entry_raw"] = str(res)[:200] if not isinstance(res, (list, tuple)) or res[0] != self.OK: break debug_info["entries_scanned"] += 1 rtime = int(res[1]) if rtime > 0 and rtime not in rtimes: rtimes.append(rtime) if len(rtimes) >= limit: break finally: if debug_info["filter_used"] and reset_fn: try: reset_fn() except Exception: pass return rtimes, debug_info def get_scenario_timesteps(self, limit=100): """Get available time steps (rtimes) in the open scenario.""" if not self._scenario_open: return {"error": "No scenario open. Use open_scenario_results first."} import datetime as _dt try: # Must register variables before simone_get_entry works node_names, node_to_ids, init_debug = self._init_scenario_vars() # Now collect rtimes (with filter, then without) rtimes, debug = self._collect_rtimes(use_filter=True, limit=limit) if not rtimes: rtimes, debug2 = self._collect_rtimes(use_filter=False, limit=limit) debug["fallback_no_filter"] = debug2 except Exception as e: return {"error": str(e)} rtimes.sort() timesteps = [] for rt in rtimes: dt = _dt.datetime(1970, 1, 1) + _dt.timedelta(seconds=rt) timesteps.append({"rtime": rt, "datetime": dt.strftime("%Y-%m-%d %H:%M:%S")}) result = { "scenario": self._scenario_name, "timesteps": timesteps, "count": len(timesteps), "limited": len(timesteps) >= limit, } if not timesteps: result["debug"] = {"init": init_debug, "entry_iteration": debug} return result def list_scenario_objects(self, object_type="NO", limit=200): """List objects in the open scenario by type (NO=nodes, PIPE, CS, VA, etc.).""" if not self._scenario_open: return {"error": "No scenario open. Use open_scenario_results first."} # Resolve object type constant type_const = None otype_upper = object_type.upper().strip() for prefix in ["SIMONE_OBJTYPE_"]: cname = prefix + otype_upper type_const = getattr(self.api, cname, None) if type_const is not None: break if type_const is None: # Try ALL type_const = getattr(self.api, "SIMONE_OBJTYPE_ALL", None) if type_const is None: return {"error": f"Unknown object type '{object_type}'"} objects = [] try: res = self.api.simone_get_first_object(type_const, "") while isinstance(res, (list, tuple)) and res[0] == self.OK: obj_name = str(res[2]).strip() if len(res) > 2 else "?" obj_type = str(res[3]).strip() if len(res) > 3 else "" subsys = str(res[4]).strip() if len(res) > 4 else "" entry = {"name": obj_name} if obj_type: entry["type"] = obj_type if subsys: entry["subsystem"] = subsys objects.append(entry) if len(objects) >= limit: break res = self.api.simone_get_next_object() except Exception as e: return {"error": str(e)} return {"objects": objects, "count": len(objects), "object_type": otype_upper} def read_scenario_values(self, object_names, variable_suffix=".Q", rtime=None, last_timestep=False): """Read values for given objects at a specific time from the open scenario. Args: object_names: list of object names (e.g. ["NODE1", "NODE2"]) variable_suffix: variable suffix (e.g. ".Q" for flow, ".P" for pressure) rtime: specific rtime (seconds since epoch). If None, uses last timestep. last_timestep: if True and rtime is None, find and use the last timestep. """ if not self._scenario_open: return {"error": "No scenario open. Use open_scenario_results first."} import datetime as _dt if not isinstance(object_names, list): object_names = [object_names] # Build variable names and resolve IDs FIRST (required before simone_get_entry) var_names = [n + variable_suffix for n in object_names] nvars = len(var_names) try: ids = self.api.new_intArray(nvars) exts = self.api.new_intArray(nvars) stats_arr = self.api.new_intArray(nvars) ss = self.api.simone_varid_array_python(var_names, ids, exts, stats_arr) if ss != self.OK: return {"error": f"simone_varid_array_python failed (status={ss}). Variable suffix '{variable_suffix}' may be invalid."} except Exception as e: return {"error": f"Variable resolution failed: {e}"} # If no rtime specified, find one (now that vars are registered) if rtime is None or last_timestep: try: rtimes, _ = self._collect_rtimes(use_filter=True, limit=1000) if not rtimes: rtimes, _ = self._collect_rtimes(use_filter=False, limit=1000) if rtimes: rtimes.sort() rtime = rtimes[-1] if last_timestep else rtimes[0] except Exception as e: return {"error": f"Failed to determine timestep: {e}"} # If still no rtime, try reading at rtime=0 as last resort if not rtime: rtime = 0 # Read values at the specified time self.api.simone_set_rtime(int(rtime)) unit_default = getattr(self.api, "SIMONE_UNIT_DEFAULT", 0) results = [] for i, name in enumerate(object_names): st = self.api.intArray_getitem(stats_arr, i) if st != self.OK: results.append({"name": name, "variable": var_names[i], "error": "variable not found"}) continue oid = self.api.intArray_getitem(ids, i) eid = self.api.intArray_getitem(exts, i) # Get unit info for this specific variable unit_abbr, unit_desc = self._get_unit_info(oid, eid) read_desc = unit_desc if unit_desc else unit_default try: sr = self.api.simone_read(oid, eid, read_desc) if isinstance(sr, (list, tuple)) and sr[0] == self.OK: entry = { "name": name, "variable": var_names[i], "value": float(sr[1]), } if unit_abbr: entry["unit"] = unit_abbr results.append(entry) else: results.append({"name": name, "variable": var_names[i], "error": f"read failed (status={sr[0] if isinstance(sr, (list,tuple)) else sr})"}) except Exception as e: results.append({"name": name, "variable": var_names[i], "error": str(e)}) dt = _dt.datetime(1970, 1, 1) + _dt.timedelta(seconds=int(rtime)) return { "scenario": self._scenario_name, "rtime": rtime, "datetime": dt.strftime("%Y-%m-%d %H:%M:%S"), "variable_suffix": variable_suffix, "results": results, } def close(self): """Clean up SIMONE resources in correct order.""" self._release_network() # End the API session try: self.api.simone_end() except Exception: pass # ━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━ # Claude Tool Definitions # ━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━ TOOLS = [ { "name": "list_networks", "description": "List all available SIMONE networks. Shows which one is currently selected.", "input_schema": { "type": "object", "properties": {}, }, }, { "name": "select_network", "description": "Switch to a different network. Releases the current network lock first. Will refuse if there are unsaved changes.", "input_schema": { "type": "object", "properties": { "name": {"type": "string", "description": "Network name to switch to."}, }, "required": ["name"], }, }, { "name": "get_network_info", "description": "Get current network information: name, directory, node/element/subsystem counts, owner.", "input_schema": { "type": "object", "properties": {}, }, }, { "name": "list_nodes", "description": "List nodes in the network. Use pattern with * and ? wildcards to filter (e.g. 'KS*' for compressor stations).", "input_schema": { "type": "object", "properties": { "pattern": { "type": "string", "description": "Wildcard filter pattern (e.g. '*FUEL*', 'KS??'). Omit for all nodes.", }, "limit": { "type": "integer", "description": "Max nodes to return (default 200).", }, }, }, }, { "name": "get_node_details", "description": "Get full details of a specific node: coordinates, height, supply flag, connected elements, alias.", "input_schema": { "type": "object", "properties": { "name": {"type": "string", "description": "Exact node name."}, }, "required": ["name"], }, }, { "name": "list_elements", "description": "List elements (pipes, valves, compressors, etc.). Filter by name pattern and/or type.", "input_schema": { "type": "object", "properties": { "pattern": { "type": "string", "description": "Wildcard filter for element name.", }, "element_type": { "type": "string", "description": "Filter by type: pipe, valve, compressor, control_valve, regulator, short_element, measuring_station, non_return_valve, mixer.", }, "limit": {"type": "integer", "description": "Max results (default 200)."}, }, }, }, { "name": "get_element_details", "description": "Get full TOPOLOGY details of a specific element: type, start/end nodes, parameters (diameter, length, roughness, etc. with units), subsystem. Uses topo API. For scenario results (flow, pressure), use read_scenario_values instead.", "input_schema": { "type": "object", "properties": { "name": {"type": "string", "description": "Exact element name."}, }, "required": ["name"], }, }, { "name": "list_subsystems", "description": "List all subsystems in the network.", "input_schema": { "type": "object", "properties": {}, }, }, { "name": "list_scenarios", "description": "List available scenarios in the network with type info (DYN=dynamic, STA=static, REC=reconstruction, etc.), time range, owner, and comment.", "input_schema": { "type": "object", "properties": {}, }, }, { "name": "get_scenario_info", "description": "Get detailed info about a specific scenario: type (DYN/STA/REC/FIL/S_O/C_O/PER/CPO/CPS/LTO/VOP), time range, initial conditions, owner, comment, calculation status. Does not require opening the scenario.", "input_schema": { "type": "object", "properties": { "name": {"type": "string", "description": "Scenario name."}, }, "required": ["name"], }, }, { "name": "find_objects", "description": "Search for nodes AND elements matching a name pattern. Good for exploring the network.", "input_schema": { "type": "object", "properties": { "pattern": { "type": "string", "description": "Wildcard pattern to search (e.g. '*BRUMOV*').", }, }, "required": ["pattern"], }, }, { "name": "create_node", "description": "Create a new node at given coordinates. DESTRUCTIVE: modifies the network.", "input_schema": { "type": "object", "properties": { "name": {"type": "string", "description": "Name for the new node."}, "x": {"type": "number", "description": "X coordinate (default 0)."}, "y": {"type": "number", "description": "Y coordinate (default 0)."}, }, "required": ["name"], }, }, { "name": "remove_node", "description": "Remove a node from the network. DESTRUCTIVE: the node and its connections will be deleted.", "input_schema": { "type": "object", "properties": { "name": {"type": "string", "description": "Name of the node to remove."}, }, "required": ["name"], }, }, { "name": "modify_node", "description": "Modify properties of an existing node. Only provide the properties you want to change.", "input_schema": { "type": "object", "properties": { "name": {"type": "string", "description": "Current node name."}, "new_name": {"type": "string", "description": "New name for the node."}, "x": {"type": "number", "description": "New X coordinate."}, "y": {"type": "number", "description": "New Y coordinate."}, "height": {"type": "number", "description": "New height value."}, "is_supply": {"type": "boolean", "description": "Set supply flag."}, "alias": {"type": "string", "description": "Set alias text."}, }, "required": ["name"], }, }, { "name": "create_element", "description": "Create a new element (pipe, valve, etc.) connecting two existing nodes. DESTRUCTIVE.", "input_schema": { "type": "object", "properties": { "name": {"type": "string", "description": "Name for the new element."}, "element_type": { "type": "string", "description": "Type: pipe, valve, compressor, control_valve, regulator, short_element, measuring_station, non_return_valve, mixer.", }, "start_node": {"type": "string", "description": "Name of the start node."}, "end_node": {"type": "string", "description": "Name of the end node."}, "subsystem": {"type": "string", "description": "Optional subsystem name."}, }, "required": ["name", "element_type", "start_node", "end_node"], }, }, { "name": "remove_element", "description": "Remove an element from the network. DESTRUCTIVE.", "input_schema": { "type": "object", "properties": { "name": {"type": "string", "description": "Name of the element to remove."}, }, "required": ["name"], }, }, { "name": "modify_element", "description": "Modify an element's TOPOLOGY parameters (diameter, length, roughness, resistance) using the topo API. These are permanent network properties stored in the topology file. Always specify units. For scenario entries (flow, pressure, setpoints), use write_scenario_values instead.", "input_schema": { "type": "object", "properties": { "name": {"type": "string", "description": "Element name."}, "new_name": {"type": "string", "description": "New name."}, "diameter": {"type": "number", "description": "New diameter value."}, "diameter_unit": {"type": "string", "description": "Diameter unit (e.g. 'mm', 'in'). Uses unit manager default if omitted."}, "length": {"type": "number", "description": "New length value."}, "length_unit": {"type": "string", "description": "Length unit (e.g. 'km', 'm', 'mi'). Uses unit manager default if omitted."}, "roughness": {"type": "number", "description": "New roughness value."}, "roughness_unit": {"type": "string", "description": "Roughness unit (e.g. 'mm', 'um'). Uses unit manager default if omitted."}, "resistance": {"type": "number", "description": "New resistance coefficient."}, "resistance_unit": {"type": "string", "description": "Resistance unit. Uses unit manager default if omitted."}, "alias": {"type": "string", "description": "New alias."}, "subsystem": {"type": "string", "description": "Assign to subsystem."}, }, "required": ["name"], }, }, { "name": "create_subsystem", "description": "Create a new subsystem in the network.", "input_schema": { "type": "object", "properties": { "name": {"type": "string", "description": "Subsystem name."}, }, "required": ["name"], }, }, { "name": "save_network", "description": "Save the current network model to the edited network file. Use activate_network afterwards to make it the active version.", "input_schema": { "type": "object", "properties": {}, }, }, { "name": "activate_network", "description": "Activate the edited network — make it the active version used by SIMONE for simulations. Call save_network first.", "input_schema": { "type": "object", "properties": {}, }, }, { "name": "load_scenario", "description": "Load/include a scenario by name into the current session.", "input_schema": { "type": "object", "properties": { "name": {"type": "string", "description": "Scenario name."}, }, "required": ["name"], }, }, { "name": "save_scenario_as", "description": "Save the current scenario with a new name. Can change scenario type (DYN/STA/REC/FIL/S_O/C_O/PER/CPO/CPS/LTO/VOP). For dynamic (DYN) scenarios, set duration_hours or start_time/end_time. Use overwrite=true if the target name already exists.", "input_schema": { "type": "object", "properties": { "name": {"type": "string", "description": "New scenario name."}, "comment": {"type": "string", "description": "Optional comment."}, "overwrite": {"type": "boolean", "description": "If true, remove existing scenario with same name first."}, "scenario_type": {"type": "string", "description": "Scenario type code: DYN, STA, REC, FIL, S_O, C_O, PER, CPO, CPS, LTO, VOP. Omit to keep same type."}, "duration_hours": {"type": "number", "description": "Simulation duration in hours (for DYN scenarios). Used with existing start time or with start_time."}, "start_time": {"type": "string", "description": "Start time as ISO datetime (e.g. '2025-01-15 06:00'). Optional."}, "end_time": {"type": "string", "description": "End time as ISO datetime. Alternative to duration_hours."}, }, "required": ["name"], }, }, { "name": "write_scenario_values", "description": "Write scenario parameter values (offtakes, pressures, setpoints, etc.). Scenario must be open in WRITE mode (use load_scenario first). Variable names use format: OBJECT.EXTENSION (e.g. OUT1.Q for flow, S1.P for pressure, VA_001.MODE for valve mode). Always specify a unit — the SIMONE unit manager converts automatically.", "input_schema": { "type": "object", "properties": { "entries": { "type": "array", "items": { "type": "object", "properties": { "name": {"type": "string", "description": "Variable name (e.g. OUT1.Q, PIPE1.BP)."}, "value": {"type": "number", "description": "Value to set."}, "unit": {"type": "string", "description": "Unit abbreviation (e.g. 'bar', '1000Nm3/h', 'C', 'mm', 'm/s', 'kW'). Use SIMONE input abbreviations. The unit manager handles conversion."}, }, "required": ["name", "value"], }, "description": "List of {name, value, unit} entries to write.", }, "rtime": { "type": "integer", "description": "Time for the parameter (0 = valid from start / static). Default: 0.", }, }, "required": ["entries"], }, }, # ── Scenario execution ── { "name": "execute_scenario", "description": "Execute/calculate a SIMONE scenario. This runs the simulation and produces results. Blocking — waits until simulation completes. Requires extended API license.", "input_schema": { "type": "object", "properties": { "name": {"type": "string", "description": "Scenario name to execute."}, }, "required": ["name"], }, }, { "name": "get_calculation_status", "description": "Check if a scenario has been calculated (without opening it). Returns 'RUNOK' if successfully calculated.", "input_schema": { "type": "object", "properties": { "name": {"type": "string", "description": "Scenario name to check."}, }, "required": ["name"], }, }, # ── Generic API ── { "name": "call_simone_api", "description": ( "Call ANY SIMONE API function by name. Use this for functions not covered " "by other tools — profiles, functions, object sets, layers, gas quality, " "reference conditions, user attributes, messages, configuration, request lists, " "time conversion, etc. Pass SIMONE_* constant names as strings — they are " "auto-resolved. For topo functions needing h_nw, pass session.h_nw. " "Check list_api_functions first to see available functions and their arguments." ), "input_schema": { "type": "object", "properties": { "function_name": { "type": "string", "description": "Exact SIMONE API function name (e.g. 'simone_read', 'simone_nw_layer_get_first').", }, "args": { "type": "array", "items": {}, "description": ( "Arguments to pass. Use SIMONE_* constant names as strings " "(e.g. 'SIMONE_NO_FLAG', 'SIMONE_MODE_READ'). " "Numbers and strings are auto-coerced." ), }, }, "required": ["function_name"], }, }, { "name": "list_api_functions", "description": ( "List all available SIMONE API functions with argument counts and descriptions. " "Use category to filter (e.g. 'profile', 'layer', 'gasqual', 'unit', 'time', " "'message', 'config', 'attr', 'object_set', 'refcond', 'function')." ), "input_schema": { "type": "object", "properties": { "category": { "type": "string", "description": "Filter keyword (searches function name and description).", }, }, }, }, # ── Scenario result reading ── { "name": "open_scenario_results", "description": "Open a scenario for reading simulation results. Must be called before reading values.", "input_schema": { "type": "object", "properties": { "name": {"type": "string", "description": "Scenario name to open."}, }, "required": ["name"], }, }, { "name": "close_scenario_results", "description": "Close the currently open scenario (free resources).", "input_schema": { "type": "object", "properties": {}, }, }, { "name": "get_scenario_timesteps", "description": "Get available simulation time steps in the open scenario.", "input_schema": { "type": "object", "properties": { "limit": {"type": "integer", "description": "Max timesteps to return (default 100)."}, }, }, }, { "name": "list_scenario_objects", "description": "List objects (nodes, pipes, compressor stations, etc.) in the open scenario.", "input_schema": { "type": "object", "properties": { "object_type": { "type": "string", "description": "Object type: NO (nodes), PIPE, CS (compressor station), VA (valve), RG (regulator), ALL. Default: NO.", }, "limit": {"type": "integer", "description": "Max objects to return (default 200)."}, }, }, }, { "name": "read_scenario_values", "description": "Read simulation result values for given objects at a specific time. Common variable suffixes: .Q (flow), .P (pressure), .T (temperature), .RHO (density), .V (velocity).", "input_schema": { "type": "object", "properties": { "object_names": { "type": "array", "items": {"type": "string"}, "description": "List of object names to read values for.", }, "variable_suffix": { "type": "string", "description": "Variable suffix, e.g. '.Q' for flow, '.P' for pressure. Default: '.Q'.", }, "rtime": { "type": "integer", "description": "Specific rtime (seconds since epoch). If omitted, uses first available timestep.", }, "last_timestep": { "type": "boolean", "description": "If true, read values at the last available timestep.", }, }, "required": ["object_names"], }, }, ] # ━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━ # Tool Dispatcher # ━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━ def dispatch_tool(session, tool_name, tool_input): """Execute a tool and return the result as a JSON string.""" try: if tool_name == "list_networks": result = session.list_networks() elif tool_name == "select_network": result = session.select_network(tool_input["name"]) elif tool_name == "get_network_info": result = session.get_network_info() elif tool_name == "list_nodes": result = session.list_nodes( pattern=tool_input.get("pattern"), limit=tool_input.get("limit", 200), ) elif tool_name == "get_node_details": result = session.get_node_details(tool_input["name"]) elif tool_name == "list_elements": result = session.list_elements( pattern=tool_input.get("pattern"), element_type=tool_input.get("element_type"), limit=tool_input.get("limit", 200), ) elif tool_name == "get_element_details": result = session.get_element_details(tool_input["name"]) elif tool_name == "list_subsystems": result = session.list_subsystems() elif tool_name == "list_scenarios": result = session.list_scenarios() elif tool_name == "get_scenario_info": result = session.get_scenario_info(tool_input["name"]) elif tool_name == "find_objects": result = session.find_objects(tool_input["pattern"]) elif tool_name == "create_node": result = session.create_node( tool_input["name"], x=tool_input.get("x", 0.0), y=tool_input.get("y", 0.0), ) elif tool_name == "remove_node": result = session.remove_node(tool_input["name"]) elif tool_name == "modify_node": result = session.modify_node( tool_input["name"], new_name=tool_input.get("new_name"), x=tool_input.get("x"), y=tool_input.get("y"), height=tool_input.get("height"), is_supply=tool_input.get("is_supply"), alias=tool_input.get("alias"), ) elif tool_name == "create_element": result = session.create_element( tool_input["name"], tool_input["element_type"], tool_input["start_node"], tool_input["end_node"], subsystem=tool_input.get("subsystem"), ) elif tool_name == "remove_element": result = session.remove_element(tool_input["name"]) elif tool_name == "modify_element": result = session.modify_element( tool_input["name"], new_name=tool_input.get("new_name"), diameter=tool_input.get("diameter"), length=tool_input.get("length"), roughness=tool_input.get("roughness"), resistance=tool_input.get("resistance"), alias=tool_input.get("alias"), subsystem=tool_input.get("subsystem"), diameter_unit=tool_input.get("diameter_unit"), length_unit=tool_input.get("length_unit"), roughness_unit=tool_input.get("roughness_unit"), resistance_unit=tool_input.get("resistance_unit"), ) elif tool_name == "create_subsystem": result = session.create_subsystem(tool_input["name"]) elif tool_name == "save_network": result = session.save_network() elif tool_name == "activate_network": result = session.activate_network() elif tool_name == "load_scenario": result = session.load_scenario(tool_input["name"]) elif tool_name == "save_scenario_as": result = session.save_scenario_as( tool_input["name"], comment=tool_input.get("comment", ""), overwrite=tool_input.get("overwrite", False), scenario_type=tool_input.get("scenario_type"), duration_hours=tool_input.get("duration_hours"), start_time=tool_input.get("start_time"), end_time=tool_input.get("end_time"), ) elif tool_name == "write_scenario_values": result = session.write_scenario_values( tool_input["entries"], rtime=tool_input.get("rtime", 0), ) # ── Scenario execution ── elif tool_name == "execute_scenario": result = session.execute_scenario(tool_input["name"]) elif tool_name == "get_calculation_status": result = session.get_calculation_status(tool_input["name"]) elif tool_name == "open_scenario_results": result = session.open_scenario_results(tool_input["name"]) elif tool_name == "close_scenario_results": result = session.close_scenario_results() elif tool_name == "get_scenario_timesteps": result = session.get_scenario_timesteps( limit=tool_input.get("limit", 100), ) elif tool_name == "list_scenario_objects": result = session.list_scenario_objects( object_type=tool_input.get("object_type", "NO"), limit=tool_input.get("limit", 200), ) elif tool_name == "read_scenario_values": result = session.read_scenario_values( tool_input["object_names"], variable_suffix=tool_input.get("variable_suffix", ".Q"), rtime=tool_input.get("rtime"), last_timestep=tool_input.get("last_timestep", False), ) # ── Generic API ── elif tool_name == "call_simone_api": result = session.call_simone_api( tool_input["function_name"], args=tool_input.get("args", []), ) elif tool_name == "list_api_functions": result = session.list_api_functions( category=tool_input.get("category"), ) else: result = {"error": f"Unknown tool: {tool_name}"} except Exception as e: result = {"error": str(e), "traceback": traceback.format_exc()} return json.dumps(result, ensure_ascii=False, default=str) # ━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━ # Chat Loop # ━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━ SYSTEM_PROMPT = """\ You are a SIMONE gas pipeline network management assistant. You help the user \ query, inspect, and modify SIMONE network models and scenarios through natural \ language conversation. Current network: {network_name} Network directory: {net_dir} You have tools to: - NETWORKS: list available networks, switch between networks (select_network releases the lock) - QUERY: list/inspect nodes, elements, subsystems, scenarios, get network stats - MODIFY: create/remove/rename nodes and elements, change parameters (diameter, \ length, roughness, etc.), manage subsystems - SCENARIOS: list scenarios (with type info), get detailed scenario info, load, save, \ write scenario parameters (offtakes, pressures, setpoints) - EXECUTE: run/calculate a scenario (requires extended API license) - RESULTS: open a scenario, read simulation result values (flow .Q, pressure .P, \ temperature .T, density .RHO, velocity .V), list timesteps and objects - SAVE: store network changes to edited network file - GENERIC API: call ANY SIMONE API function (base + topo) via call_simone_api — \ profiles, functions, object sets, layers, gas quality, reference conditions, user \ attributes, messages, configuration, request lists, time conversion, and more. \ Use list_api_functions to discover available functions and their correct arguments. Scenario types (SIMONE runtype): - DYN = Dynamic simulation — time-dependent transient flow simulation - STA = Static (steady-state) — single time-point equilibrium calculation - REC = Reconstruction — state estimation from measured data - FIL = Filter — data reconciliation (measurement correction) - S_O = Set-point optimization — optimal operating point calculation - C_O = Configuration optimization — optimal network configuration - PER = Periodic steady-state — cyclic equilibrium simulation - CPO = Compressor optimization — optimal compressor settings - CPS = Compressor scheduling — time-based compressor planning - LTO = Long-term optimization — extended planning horizon - VOP = Virtual online prediction — real-time forecasting == SIMONE Unit System == Units are CONFIGURABLE via the SIMONE unit manager — they are NOT fixed. Each variable \ has a unit type (pressure, flow, temperature, etc.) and the user may have any unit selected. When READING values, the tool returns the current unit alongside the value — ALWAYS show it. When WRITING values, ALWAYS specify a unit in each entry. Use any valid SIMONE input \ abbreviation for the correct type — the unit manager converts automatically. Unit types and common input abbreviations: - Pressure (UNIT_TYPE_P): bar, MPa, barg, kPag, atm, psia, psig, Pa - Flow/standard (UNIT_TYPE_Q): 1000Nm3/h, Nm3/h, Nm3/s, Nm3/min, MNm3/d, Mcf/d, MMcf/d - Temperature (UNIT_TYPE_T): C, K, F - Velocity (UNIT_TYPE_V): m/s, km/h, ft/s - Power (UNIT_TYPE_PWR): W, kW, MW, hp - Density (UNIT_TYPE_RHO): kg/m3, lb/ft3 (or 1 for relative density) - Length (UNIT_TYPE_L): m, km, ft, mi - Diameter (UNIT_TYPE_D): mm, um, in - Roughness (UNIT_TYPE_RR): mm, um, in - Height (UNIT_TYPE_H): m, ft - HAD/adiabatic head (UNIT_TYPE_HAD): kJ/kg, m, km, ft - Flow/operating (UNIT_TYPE_QVOL): m3/s, m3/min, m3/h, 1000m3/h, cft/s - Linepack/volume (UNIT_TYPE_AC): 1000Nm3, MNm3, Nm3, Mcf, MMcf - Linepack/energy (UNIT_TYPE_GWH): kWh, kJ, Btu - Energy (UNIT_TYPE_ENERGY): kWh, kJ, Btu - Calorific value (UNIT_TYPE_CV): MJ/Nm3, kWh/Nm3, Btu/cf - Time (UNIT_TYPE_TIME): dt_s (seconds), dt_ds (date), dt_dt (datetime) == SIMONE Variable & Parameter Reference == Variable notation: ., e.g. NODE1.Q, PIPE1.PD Object types: NS (supply node), NO (other node), CS (compressor station), CV (control valve), \ VA (valve), RE (resistor), PIPE (pipe), NRV (non-return valve), MS (metering station) CRITICAL DISTINCTION — Topology vs Scenario data: • TOPOLOGY PARAMETERS (D, L, RR, height, etc.) = permanent network properties stored in \ the topology file. Read via get_element_details / get_node_details. Write via modify_element \ / modify_node. These use the TOPO API (simone_nw_element_set_parameter etc.). • SCENARIO ENTRIES (Q, P, T, MODE, CONF, setpoints etc.) = boundary conditions and input \ data for a specific scenario. Write via write_scenario_values (uses simone_write). • SCENARIO RESULTS (M, PD, PR, V, RHO, etc.) = computed simulation outputs. Read via \ read_scenario_values (uses simone_read). Requires opening the scenario first. NEVER use write_scenario_values for topology parameters (D, L, RR). \ NEVER use modify_element for scenario entries (Q, P, setpoints). --- TOPOLOGY PARAMETERS (use get_element_details / modify_element) --- PIPE topo: D (diameter, UNIT_TYPE_D), L (length, UNIT_TYPE_L), \ RR (roughness multiplier — factor applied to network editor value), \ COR_LAM (pipe efficiency factor, e.g. 0.9 = 90%), \ GT (ground temperature, UNIT_TYPE_T — DYN heat dynamics only, default 10), \ HTC (heat transfer coefficient, W/(m²K) — DYN heat dynamics only, default 2) NODE topo: height (UNIT_TYPE_H), is_supply (bool), coordinates (x, y) --- SCENARIO RESULTS (use read_scenario_values after opening scenario) --- PIPE results: M (flow rate, UNIT_TYPE_Q), PD (pressure drop, UNIT_TYPE_P), \ PR (pressure ratio, dimensionless), V (speed of flow, UNIT_TYPE_V), \ RHO (operating density, UNIT_TYPE_RHO), ANM3 (line pack volume, UNIT_TYPE_AC), \ GWH (line pack energy, UNIT_TYPE_GWH), ED (roughness of pipe element, UNIT_TYPE_RR), \ HFR (hydrate formation risk: 1=none, 2=water condensation, 3=hydrate — HFR option only), \ DTDP (gas subcooling vs water dew point, UNIT_TYPE_T), \ DTHE (gas subcooling vs hydrate equilibrium, UNIT_TYPE_T), \ DTHF (gas subcooling vs hydrate formation, UNIT_TYPE_T) --- SCENARIO ENTRIES (use write_scenario_values / read via read_scenario_values) --- NODE entries: Q (supply/offtake quantity, UNIT_TYPE_Q — positive=offtake, negative=supply), \ QCORR (flow correction factor), PSET (pressure setpoint, UNIT_TYPE_P), \ PSETDP (pressure setpoint range, UNIT_TYPE_P), \ MAXQP (max fictitious offtake for pressure condition, UNIT_TYPE_Q), \ T (temperature, UNIT_TYPE_T), Q.T (source temperature — supply nodes only, UNIT_TYPE_T), \ RUPT (rupture — equivalent hole diameter, UNIT_TYPE_D), RUPTQ0 (rupture ignoring offtake), \ PM (metered pressure — REC only, UNIT_TYPE_P), PMDP (metered press accuracy, UNIT_TYPE_P), \ SIGMA (relative accuracy of metered Q — REC only) NODE results: H (geodetic height, UNIT_TYPE_H), P (pressure, UNIT_TYPE_P), \ PM (metered/set pressure, UNIT_TYPE_P), QP (fictitious offtake, UNIT_TYPE_Q), \ T (temperature, UNIT_TYPE_T), TDP (water dew point temperature, UNIT_TYPE_T), \ THE (hydrate equilibrium temperature, UNIT_TYPE_T) VALVE (VA) entries: ON (open), OFF (close), BP (by-pass), \ BPDIAM (by-pass diameter, UNIT_TYPE_D), BPDP (by-pass pressure diff, UNIT_TYPE_P), \ RE (resistance coefficient — operate as resistor), REPD (fixed pressure drop, UNIT_TYPE_P) VALVE results: MODE (operating mode: ON/OFF/BP/RE), M (flow, UNIT_TYPE_Q), \ V (speed of flow, UNIT_TYPE_V), PD (pressure drop, UNIT_TYPE_P), PR (pressure ratio) RESISTOR (RE) entries: RE (resistance coefficient, range 0-20000), \ REPD (fixed pressure drop, UNIT_TYPE_P), OFF (close), ON (open) RESISTOR results: MODE (OFF/RE), M (flow, UNIT_TYPE_Q), V (speed, UNIT_TYPE_V), \ PD (pressure drop, UNIT_TYPE_P), PR (pressure ratio) CONTROL VALVE (CV) entries: SM (flow setpoint, UNIT_TYPE_Q), \ SPO (outlet pressure setpoint, UNIT_TYPE_P), SPI (inlet pressure setpoint, UNIT_TYPE_P), \ MMAX (flow limit, UNIT_TYPE_Q), SQVOL (volumetric flow setpoint, UNIT_TYPE_QVOL), \ SCVO (percent opening setpoint, %), SR (pressure ratio setpoint), \ VMAX (speed limit, UNIT_TYPE_V), BP (by-pass), OFF (close), \ RIN (inlet resistance), RINPD (inlet pressure drop, UNIT_TYPE_P), \ ROUT (outlet resistance), ROUTPD (outlet pressure drop, UNIT_TYPE_P), \ PIMIN (min inlet pressure, UNIT_TYPE_P), POMAX (max outlet pressure, UNIT_TYPE_P), \ POMIN (min outlet pressure, UNIT_TYPE_P), PDMIN (min pressure drop, UNIT_TYPE_P), \ CVPMAX (pre-heating power, UNIT_TYPE_PWR), CVTO (required outlet temp, UNIT_TYPE_T), \ MM (measured flow — REC only, UNIT_TYPE_Q), SIGMA (metered flow accuracy — REC only) CV results: MODE, M (flow, UNIT_TYPE_Q), PI (inlet pressure, UNIT_TYPE_P), \ PO (outlet pressure, UNIT_TYPE_P), PD (pressure drop, UNIT_TYPE_P), PR (pressure ratio), \ IPI (internal inlet pressure, UNIT_TYPE_P), IPO (internal outlet pressure, UNIT_TYPE_P), \ V (speed, UNIT_TYPE_V), TO (outlet temperature, UNIT_TYPE_T), \ POWER (pre-heating power, UNIT_TYPE_PWR), OPEN (percent opening, %) COMPRESSOR STATION (CS) entries: CONF (configuration string), \ TAMB (ambient temperature, UNIT_TYPE_T, default 12), \ SM (flow setpoint, UNIT_TYPE_Q), SPO (outlet pressure setpoint, UNIT_TYPE_P), \ SPI (inlet pressure setpoint, UNIT_TYPE_P), MMAX (flow limit, UNIT_TYPE_Q), \ SQVOL (volumetric flow setpoint, UNIT_TYPE_QVOL), SR (pressure ratio setpoint), \ BP (by-pass), OFF (close), \ RIN (inlet resistance), RINPD (inlet pressure drop, UNIT_TYPE_P), \ ROUT (outlet resistance), ROUTPD (outlet pressure drop, UNIT_TYPE_P), \ PIMIN (min inlet pressure, UNIT_TYPE_P), POMAX (max outlet pressure, UNIT_TYPE_P), \ SRPM (RPM setpoint), SRRPM (relative RPM setpoint), \ MAX (max power mode), OPER (force operated — STA optimization only), \ PRMAX (max pressure ratio, default 2), QVOLMAX (max volumetric flow, UNIT_TYPE_QVOL), \ POWERMAX (max power, UNIT_TYPE_PWR), EFFCS (compressor efficiency, default 0.8), \ EFFDRIVE (drive efficiency, default 0.3), \ MM (measured flow — REC only, UNIT_TYPE_Q), \ CVPMAX (cooling power, UNIT_TYPE_PWR), CVTO (discharge temp limit, UNIT_TYPE_T), \ PRMIN (min pressure ratio, default 1) CS results: QVOL (volumetric flow at suction, UNIT_TYPE_QVOL), \ HAD (adiabatic head, UNIT_TYPE_HAD), PERF (total shaft power, UNIT_TYPE_PWR), \ CONS (total energy consumption, UNIT_TYPE_PWR), MODE, \ CONR (reached configuration), SUCC (set-point success: OK/FB/LBP/MIN/MAX), \ COMB (combined cycle status: OFF/START/RUN/LBP/UNSUCC), \ TI (inlet temperature, UNIT_TYPE_T), TO (outlet temperature, UNIT_TYPE_T), \ POWER (cooling power, UNIT_TYPE_PWR), \ PI (suction pressure, UNIT_TYPE_P), PO (discharge pressure, UNIT_TYPE_P), \ PD (pressure drop, UNIT_TYPE_P), PR (pressure ratio), \ IPI (internal inlet pressure, UNIT_TYPE_P), IPO (internal outlet pressure, UNIT_TYPE_P), \ MCF (multiple configuration factor, integer 1-15) CS UNIT results: QVOL (UNIT_TYPE_QVOL), HAD (UNIT_TYPE_HAD), \ PI (suction pressure, UNIT_TYPE_P), PO (discharge pressure, UNIT_TYPE_P), \ TAC (cooler inlet temp, UNIT_TYPE_T), TBC (cooler outlet temp, UNIT_TYPE_T), \ EFF (efficiency), PMA (shaft power, UNIT_TYPE_PWR), PTU (max shaft power, UNIT_TYPE_PWR), \ CSM (energy consumption, UNIT_TYPE_PWR), RPM (speed), \ STAT (status: OFF/OK/MIN/CHOKE/MAX/SURGE), QBP (bypass flow, UNIT_TYPE_QVOL), \ STAG (stage assignment, integer) SUBSYSTEM entries: D (UNIT_TYPE_D), L (UNIT_TYPE_L), RR (roughness factor), \ QCORR (flow correction factor), COR_LAM (pipe efficiency factor), \ GT (ground temp, UNIT_TYPE_T — DYN only), HTC (heat transfer coeff — DYN only) SUBSYSTEM results: ACCU (line pack mass-based, UNIT_TYPE_AC), \ ANM3 (line pack standard-volume, UNIT_TYPE_AC), GWH (line pack energy, UNIT_TYPE_GWH), \ AQP (sum fictitious offtakes, UNIT_TYPE_Q), ASUPP (sum supplies, UNIT_TYPE_Q), \ AOFF (sum offtakes, UNIT_TYPE_Q), \ APMIN (min pressure, UNIT_TYPE_P), APMAX (max pressure, UNIT_TYPE_P), \ AVMAX (max speed, UNIT_TYPE_V), \ AIPMIN (node with min pressure), AIPMAX (node with max pressure), AIVMAX (node with max speed) SYSTEM entries (_SYS prefix): DT (simulation time step, UNIT_TYPE_TIME), \ TW (recording time step, UNIT_TYPE_TIME), QT (quality tracking: on/off), \ HD (heat dynamics: SIMPLE/FULL/OFF), JTEP (Joule-Thomson: No/Yes), \ ZET (compressibility equation), LAMBDA (pressure drop equation), \ TA (max time span for control change, UNIT_TYPE_TIME, default 180s), \ TQ (max time span for quality change, UNIT_TYPE_TIME, default 900s), \ SB (steady-state boundary mode: on/off), SBDT (SB time step, UNIT_TYPE_TIME), \ THETAEQ (discharge temp formula: Basic/Isentropic/RG1991), \ THETACOR (discharge temp correction: No/Basic/RG1991), \ KAPPA (isentropic exponent method: Constant/RG Equation/Equation of state), \ RECYC (recycle temperature handling: on/off), \ HF (hydrate formation risk model: No/Motiee/Kvs-model/Ponomarev), \ IT (hydrate inhibitor type: No/MeOH/EG/DEG), \ WCC (water dew point calculation: Bukacek) GENERAL results (all objects): NAME, ALIAS, TYPE, SUBSYS (subsystem membership) == SIMONE Built-in Functions (for expressions, labels, scenario definitions) == Functions can be used in scenario parameter values, labels, graphs, and anywhere \ expressions are accepted. Syntax: FUNCNAME(args). Expressions use standard operators: \ +, -, *, /, ** (power), relational (<, <=, ==, !=, >, >=), logical (!, &&, ||). Variables in expressions use . syntax, e.g. NODE1.P, PIPE1.M. Mathematical functions: ABS(x), TRUNC(x), ROUND(x), SQRT(x), SIN(x), COS(x), TAN(x), LOG(x), LOG10(x), \ MIN(x,y,...), MAX(x,y,...), SUM(x,y,...), RAND(x), RANDN(x) Special functions: IF(expr,x1,x2) — returns x1 if expr is true, else x2 SELECT(i,v0,...,vn) — returns value vi PWLF(x,x0,y0,x1,y1,...,xn,yn) — piecewise linear function with interpolation TFNUL(x) — returns 0 if x is undefined, else x (avoids undefined propagation) ISFNUL(obj.ext) — returns 1 if value is undefined, else 0 FNUL — returns the undefined/fnul value VALID(x) — for real-time data, yields 0 if value is INVALID VALT(x,t0) — value of x at time t0 (hours as float) VALX(obj,ext) — returns value for obj.ext (string arguments) IF(expr,x1,x2) — conditional DIST(node,name) — shortest distance (km) from node to another node or object set MATCH(string,pattern) — wildcard match, returns 1.0 if matches XTYPE(object) — returns object type (differentiates NRV, MS, fuel nodes) POSX(node), POSY(node) — internal X/Y coordinates of a node INSET(object[,setname]) — returns 1.0 if object is in the specified object set Time-based functions (compute over time periods): INTG(x[,t0]) — integral of x since initial time (or t0) INTG2(x[,t0]) — integral using step function (right-hand values) DER(x) — time derivative of x MEANH(x), MEAND(x), MEANT(x) — hourly/daily/total average MEANX(x,dt) — floating average over past dt hours MINH(x), MIND(x), MINT(x) — hourly/daily/total minimum MAXH(x), MAXD(x), MAXT(x) — hourly/daily/total maximum Limit checking functions: TESTLIM(o) — returns TRUE if any limit is defined for object o CHECKLIM(o,x) — returns +/-1 if upper/lower limit for param x at object o was breached CHECKLIMT(o,x) — returns time of limit breach (decimal hours), else 0 Functions are managed via the API: simone_define_function (base API Ch 13.10), \ simone_define_function_ex, simone_get_function, simone_remove_function, \ simone_rename_function. Use call_simone_api to access these. == Generic API Access (call_simone_api) == For any SIMONE API function not covered by the dedicated tools above, use \ call_simone_api(function_name, args). This gives access to ALL ~220 base API and \ ~220 topo API functions. Use list_api_functions(category) to discover functions. Key categories: profile, function, object_set, layer, gasqual, refcond, attr, \ message, config, req, time, unit, execute, label, image. SIMONE_* constants can be passed as string args — they are auto-resolved. For topo functions needing h_nw (network handle), it is available via the session. Handles returned by get/find functions can be used in subsequent calls. ALWAYS release handles with simone_nw_handle_release when done. IMPORTANT: Always use the exact variable/parameter extensions listed above. \ Do NOT guess or invent variable names. If the user asks about a variable not listed here, \ tell them it is not in the reference and suggest the closest match. When displaying read values, ALWAYS include the unit returned by the tool. \ When writing values, ALWAYS specify the unit field in each entry. Guidelines: - Before any DESTRUCTIVE operation (create, remove, modify), clearly state what \ you are about to do and ask the user to confirm unless they explicitly asked for it. - When listing large numbers of objects, use patterns to filter. - Respond in the same language the user uses. - Be concise and practical. - After modifications, remind the user to save if they haven't. - Network changes are saved to the EDITED version first (save_network). Then use \ activate_network to make it the active version. Always confirm with the user before activating. - When saving a scenario as DYN (dynamic) type without duration or time range, \ ALWAYS ask the user how long the simulation should run (duration_hours) before saving. \ Dynamic scenarios require a time range. For STA (static) no duration is needed. """ def run_chat(session): """Main interactive chat loop.""" api_key = os.environ.get("ANTHROPIC_API_KEY", _DEFAULT_API_KEY) model = os.environ.get("SIMONE_CHATBOT_MODEL", "claude-sonnet-4-6") system = SYSTEM_PROMPT.format( network_name=session.network_name or "(none)", net_dir=session.net_dir or "(unknown)", ) messages = [] print() print("Type your instructions (or 'quit' to exit, 'save' to save network).") print() while True: # Ctrl+C anywhere in this loop cancels the current action, never exits try: try: user_input = input("You> ").strip() except EOFError: print() break if not user_input: continue # Special commands if user_input.lower() in ("quit", "exit", "q"): if session._dirty: print("Warning: You have unsaved changes!") try: ans = input("Save before exit? (y/n): ").strip().lower() except (EOFError, KeyboardInterrupt): ans = "n" if ans in ("y", "yes", "a", "ano"): result = session.save_network() print(json.dumps(result, indent=2)) break if user_input.lower() == "save": result = session.save_network() print(json.dumps(result, indent=2, ensure_ascii=False)) continue if user_input.lower() == "status": info = { "network": session.network_name, "directory": session.net_dir, "editing": session._edit_started, "unsaved_changes": session._dirty, } print(json.dumps(info, indent=2, ensure_ascii=False)) continue messages.append({"role": "user", "content": user_input}) # Conversation loop (handle tool calls) cancelled = False while True: api_messages = _serialize_messages(messages) with ThinkingIndicator(): resp = claude_api_call( api_key, model, system, TOOLS, api_messages ) content_blocks = resp.get("content", []) stop_reason = resp.get("stop_reason", "end_turn") # Store raw content blocks for conversation history messages.append({"role": "assistant", "content": content_blocks}) # Check for tool use tool_uses = [b for b in content_blocks if b.get("type") == "tool_use"] if not tool_uses: for block in content_blocks: if block.get("type") == "text" and block.get("text"): print(f"\nSIMONE> {block['text']}") print() break # Execute tools and collect results tool_results = [] for tu in tool_uses: t_name = tu["name"] t_input = tu.get("input", {}) t_id = tu["id"] print(f" [calling {t_name}({json.dumps(t_input, ensure_ascii=False)})]") result_str = dispatch_tool(session, t_name, t_input) tool_results.append({ "type": "tool_result", "tool_use_id": t_id, "content": result_str, }) messages.append({"role": "user", "content": tool_results}) # Print any text blocks alongside tool uses for block in content_blocks: if block.get("type") == "text" and block.get("text", "").strip(): print(f"\nSIMONE> {block['text']}") if stop_reason != "tool_use": print() break except KeyboardInterrupt: # Cancel current turn — clean up messages to consistent state print("\n [cancelled]") # Remove any partial messages from this turn while messages and messages[-1]["role"] != "user": messages.pop() # Remove the user message that started this turn (if any content was typed) if messages and messages[-1]["role"] == "user" and isinstance(messages[-1]["content"], str): messages.pop() continue except Exception as e: print(f"\nError: {e}") # Clean up messages while messages and messages[-1]["role"] != "user": messages.pop() if messages and messages[-1]["role"] == "user" and isinstance(messages[-1]["content"], str): messages.pop() continue def _serialize_messages(messages): """Ensure messages are JSON-serializable for the API.""" out = [] for msg in messages: role = msg["role"] content = msg["content"] # Already a string if isinstance(content, str): out.append({"role": role, "content": content}) # List of blocks (from API response or tool results) elif isinstance(content, list): out.append({"role": role, "content": content}) else: out.append({"role": role, "content": str(content)}) return out # ━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━ # Entry Point # ━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━ def main(): print("=" * 60) print(" SIMONE Chatbot v1.6") print(" AI-powered network management for SIMONE") print(" No pip required — stdlib only") print("=" * 60) # Find and init SIMONE print("\nInitializing SIMONE SDK...") root = find_simone_root() if not root: print("ERROR: Cannot find SIMONE installation.") print("Set SIMONE_ROOT environment variable to your SIMONE directory.") sys.exit(1) print(f" SIMONE root: {root}") try: api = init_simone_sdk(root) except ImportError as e: print(f"ERROR: Cannot load SimoneApi: {e}") print("Make sure SIMONE is properly installed with the API SDK.") sys.exit(1) session = SimoneSession(api, root) try: network = session.connect() if network: print(f" Connected to network: {network}") if session.net_dir: print(f" Network directory: {session.net_dir}") else: print(" Warning: No network currently selected in SIMONE.") print(" Open a network in SIMONE first, then restart this chatbot.") print(f"\n Tools: {len(TOOLS)} ({', '.join(t['name'] for t in TOOLS[-5:])} ...)") print(f" Available element types: {', '.join(sorted(session._element_type_map.keys()))}") print(f" Available parameters: {', '.join(sorted(session._param_map.keys()))}") run_chat(session) except Exception as e: print(f"\nERROR: {e}") traceback.print_exc() finally: print("\nCleaning up...") session.close() print("Goodbye!") if __name__ == "__main__": main()