#!/usr/bin/env python """Shared helpers for Windows CODESYS action scripts.""" from __future__ import print_function import json import os import sys try: text_type = unicode # type: ignore[name-defined] binary_type = str PY2 = True except NameError: text_type = str binary_type = bytes PY2 = False def _argv(index): if len(sys.argv) > index: return sys.argv[index] return None def load_payload(): """Load payload from argv[1] (json string or json file path).""" raw = _argv(1) if not raw: raise ValueError("Missing payload argument.") if os.path.exists(raw): with open(raw, "r") as f: return json.loads(f.read()) try: return json.loads(raw) except ValueError as exc: raise ValueError("Invalid JSON payload: {0}".format(exc)) def _normalize_text(value): if isinstance(value, dict): normalized = {} for key, item in value.items(): normalized[_normalize_text(key)] = _normalize_text(item) return normalized if isinstance(value, list): return [_normalize_text(item) for item in value] if isinstance(value, tuple): return [_normalize_text(item) for item in value] if isinstance(value, binary_type): if PY2: for encoding in ("utf-8", "cp1252", "latin-1"): try: return value.decode(encoding).encode("ascii", "replace") except Exception: pass return value.decode("latin-1", "replace").encode("ascii", "replace") for encoding in ("utf-8", "cp1252", "latin-1"): try: return value.decode(encoding).encode("ascii", "replace").decode("ascii") except Exception: pass return value.decode("latin-1", "replace").encode("ascii", "replace").decode("ascii") if isinstance(value, text_type): if PY2: return value.encode("ascii", "replace") return value.encode("ascii", "replace").decode("ascii") return value def emit(data): """Print JSON to stdout and optionally write to argv[2] path.""" text = json.dumps(_normalize_text(data), indent=2, ensure_ascii=True) print(text) result_path = _argv(2) or os.environ.get("CODESYS_SCRIPT_RESULT_PATH") if result_path: with open(result_path, "w") as f: f.write(text)