Add now/random/assert-var flow commands and QR reading

Four additions:
- AC_now_to_var: current time (strftime format) into a variable.
- AC_random_to_var: a seeded random int / float / choice into a variable.
- assert_variable / AC_assert_var: fail when a flow variable doesn't
  satisfy eq/ne/lt/gt/contains/regex/... — the assertion-DSL companion to
  if_var, pairing with ocr_to_var / shell_to_var.
- read_qr_codes / AC_read_qr: decode QR codes in a screen region via
  OpenCV's QRCodeDetector (no new dependency; decoder injectable).

All exposed via the facade / executor and the visual script builder.

Author: JE-Chen

je_auto_control/__init__.py

@@ -79,6 +79,8 @@
 from je_auto_control.utils.scroll_find import scroll_until_visible
 # Recoverable deletion (move files to the OS recycle bin).
 from je_auto_control.utils.trash import move_to_trash
+# QR code decoding from a screen region / image.
+from je_auto_control.utils.qr import read_qr_codes
 # WebRunner bridge (headless: optional je_web_runner dependency)
 from je_auto_control.utils.webrunner_bridge import (
     WebRunnerBridgeError, is_webrunner_available, list_webrunner_commands,
@@ -155,7 +157,8 @@
     AssertionResult, GroupAssertionResult, assert_all, assert_any,
     assert_by_description, assert_clipboard, assert_duration,
     assert_eventually, assert_file, assert_http, assert_image, assert_pixel,
-    assert_process, assert_text, assert_window, run_assertion_spec,
+    assert_process, assert_text, assert_variable, assert_window,
+    run_assertion_spec,
 )
 # Data-driven execution (load rows from CSV / JSON / SQLite / Excel)
 from je_auto_control.utils.data_source import data_source_kinds, load_rows
@@ -549,6 +552,7 @@ def start_autocontrol_gui(*args, **kwargs):
     "AssertionResult", "assert_image", "assert_pixel",
     "assert_text", "assert_window", "assert_clipboard", "assert_process",
     "assert_file", "assert_http", "assert_by_description", "assert_duration",
+    "assert_variable",
     # Assertion combinators (soft groups + eventual polling)
     "GroupAssertionResult", "assert_all", "assert_any", "assert_eventually",
     "run_assertion_spec",
@@ -602,6 +606,8 @@ def start_autocontrol_gui(*args, **kwargs):
     "scroll_until_visible",
     # Recoverable deletion (recycle bin)
     "move_to_trash",
+    # QR code decoding
+    "read_qr_codes",
     # WebRunner bridge (browser automation via je_web_runner)
     "WebRunnerBridgeError", "is_webrunner_available",
     "list_webrunner_commands", "run_webrunner_action",

je_auto_control/gui/script_builder/command_schema.py

@@ -224,6 +224,14 @@ def _add_ocr_specs(specs: List[CommandSpec]) -> None:
                       default=60.0, min_value=0.0, max_value=100.0),
         ),
     ))
+    specs.append(CommandSpec(
+        "AC_read_qr", "OCR", "Read QR Codes",
+        fields=(
+            FieldSpec("region", FieldType.STRING, optional=True,
+                      placeholder="[0, 0, 400, 400]"),
+        ),
+        description="Decode QR codes in a screen region (OpenCV).",
+    ))
     specs.append(CommandSpec(
         "AC_scroll_to_find", "OCR", "Scroll Until Visible",
         fields=(
@@ -448,6 +456,39 @@ def _add_flow_specs(specs: List[CommandSpec]) -> None:
         ),
         description="String-transform a variable (upper/strip/replace/regex/...).",
     ))
+    specs.append(CommandSpec(
+        "AC_now_to_var", "Flow", "Timestamp into Variable",
+        fields=(
+            FieldSpec("var", FieldType.STRING, default="now"),
+            FieldSpec("format", FieldType.STRING, optional=True,
+                      default="%Y-%m-%d %H:%M:%S"),
+        ),
+        description="Store the current time (strftime format) in a variable.",
+    ))
+    specs.append(CommandSpec(
+        "AC_random_to_var", "Flow", "Random into Variable",
+        fields=(
+            FieldSpec("var", FieldType.STRING, default="random"),
+            FieldSpec("kind", FieldType.ENUM,
+                      choices=("int", "float", "choice"), default="int"),
+            FieldSpec("min", FieldType.FLOAT, optional=True, default=0.0),
+            FieldSpec("max", FieldType.FLOAT, optional=True, default=100.0),
+            FieldSpec("seed", FieldType.INT, optional=True),
+        ),
+        description="Store a random int / float / choice in a variable.",
+    ))
+    specs.append(CommandSpec(
+        "AC_assert_var", "Flow", "Assert Variable",
+        fields=(
+            FieldSpec("name", FieldType.STRING),
+            FieldSpec("op", FieldType.ENUM,
+                      choices=("eq", "ne", "lt", "le", "gt", "ge",
+                               "contains", "startswith", "endswith",
+                               "regex"), default="eq"),
+            FieldSpec("value", FieldType.STRING, optional=True),
+        ),
+        description="Fail if a flow variable doesn't satisfy the condition.",
+    ))
     specs.append(CommandSpec(
         "AC_assert_duration", "Flow", "Assert Duration (perf budget)",
         fields=(

je_auto_control/utils/assertion/__init__.py

@@ -19,6 +19,7 @@
     assert_pixel,
     assert_process,
     assert_text,
+    assert_variable,
     assert_window,
 )
 from je_auto_control.utils.assertion.combinators import (
@@ -45,6 +46,7 @@
     "assert_pixel",
     "assert_process",
     "assert_text",
+    "assert_variable",
     "assert_window",
     "run_assertion_spec",
 ]

je_auto_control/utils/assertion/assertions.py

@@ -503,6 +503,56 @@ def assert_by_description(description: str,
     )
 
 
+def _variable_satisfies(value: Any, op: str, expected: Any) -> bool:
+    """Return True when ``value op expected`` holds (eq/ne/contains/...)."""
+    import re
+    comparators = {
+        "eq": lambda a, b: a == b,
+        "ne": lambda a, b: a != b,
+        "lt": lambda a, b: a < b,
+        "le": lambda a, b: a <= b,
+        "gt": lambda a, b: a > b,
+        "ge": lambda a, b: a >= b,
+        "contains": lambda a, b: b in a,
+        "startswith": lambda a, b: isinstance(a, str) and a.startswith(b),
+        "endswith": lambda a, b: isinstance(a, str) and a.endswith(b),
+        "regex": lambda a, b: re.search(str(b), str(a)) is not None,
+    }
+    comparator = comparators.get(op)
+    if comparator is None:
+        raise AutoControlAssertionException(
+            f"assert_var: unknown op {op!r}; expected one of "
+            f"{sorted(comparators)}"
+        )
+    try:
+        return bool(comparator(value, expected))
+    except TypeError:
+        return False
+
+
+def assert_variable(value: Any, op: str = "eq", expected: Any = None,
+                    name: str = "variable",
+                    raise_on_fail: bool = True) -> AssertionResult:
+    """Assert that ``value`` satisfies ``op expected`` (eq/ne/contains/regex/…).
+
+    The assertion-DSL companion to the flow ``if_var`` / ``while_var``
+    conditions: instead of branching, fail loudly when a variable doesn't
+    hold the expected value — handy after ``ocr_to_var`` / ``shell_to_var``.
+    """
+    passed = _variable_satisfies(value, op, expected)
+    message = (
+        f"assert_var passed: {name}={value!r} {op} {expected!r}"
+        if passed else
+        f"assert_var failed: expected {name}={value!r} to satisfy "
+        f"{op} {expected!r}"
+    )
+    return _finalize(
+        "variable", passed, message,
+        expected={"op": op, "value": expected}, actual=value,
+        raise_on_fail=raise_on_fail, capture_on_fail=False,
+    )
+
+
 def assert_duration(action: Callable[[], Any], max_ms: float,
                     min_ms: float = 0.0,
                     raise_on_fail: bool = True,

je_auto_control/utils/executor/action_executor.py

@@ -2015,6 +2015,31 @@ def _move_to_trash(path: str) -> Dict[str, Any]:
     return {"trashed": move_to_trash(path)}
 
 
+def _read_qr(region: Optional[Union[List[int], str]] = None) -> Dict[str, Any]:
+    """Executor adapter: decode QR codes in a screen region.
+
+    ``region`` is ``[x1, y1, x2, y2]`` (or a JSON string for the builder);
+    omit it to scan the whole screen.
+    """
+    import json
+    import os
+    import tempfile
+    from je_auto_control.utils.qr import read_qr_codes
+    from je_auto_control.wrapper.auto_control_screen import screenshot
+    if isinstance(region, str):
+        region = json.loads(region) if region.strip() else None
+    handle, tmp = tempfile.mkstemp(prefix="qr_", suffix=".png")
+    os.close(handle)
+    try:
+        screenshot(tmp, screen_region=region)
+        return {"codes": read_qr_codes(tmp)}
+    finally:
+        try:
+            os.unlink(tmp)
+        except OSError:
+            pass
+
+
 def _scroll_to_find(target: str, kind: str = "image", direction: str = "down",
                     max_scrolls: int = 10,
                     scroll_amount: int = 3) -> Dict[str, Any]:
@@ -2484,6 +2509,9 @@ def __init__(self):
             # Recoverable deletion (move a file to the OS recycle bin)
             "AC_move_to_trash": _move_to_trash,
 
+            # QR code decoding from a screen region
+            "AC_read_qr": _read_qr,
+
             # Scroll until a target image / text is visible
             "AC_scroll_to_find": _scroll_to_find,
 

je_auto_control/utils/executor/flow_control.py

@@ -389,6 +389,48 @@ def exec_shell_to_var(executor: Any, args: Mapping[str, Any]) -> Dict[str, Any]:
             "returncode": completed.returncode}
 
 
+def _now():
+    import datetime as _dt
+    return _dt.datetime.now()
+
+
+def exec_now_to_var(executor: Any, args: Mapping[str, Any]) -> Dict[str, Any]:
+    """Store the current local time (strftime format) in a flow variable."""
+    value = _now().strftime(str(args.get("format", "%Y-%m-%d %H:%M:%S")))
+    var_name = args.get("var", "now")
+    executor.variables.set(var_name, value)
+    return {"var": var_name, "value": value}
+
+
+def exec_random_to_var(executor: Any,
+                       args: Mapping[str, Any]) -> Dict[str, Any]:
+    """Store a random value (int / float / choice) in a flow variable."""
+    import random
+    rng = random.Random(args.get("seed"))
+    kind = str(args.get("kind", "int"))
+    if kind == "choice":
+        value: Any = rng.choice(list(args.get("choices") or [None]))
+    elif kind == "float":
+        value = rng.uniform(float(args.get("min", 0.0)),
+                            float(args.get("max", 1.0)))
+    else:
+        value = rng.randint(int(args.get("min", 0)), int(args.get("max", 100)))
+    var_name = args.get("var", "random")
+    executor.variables.set(var_name, value)
+    return {"var": var_name, "value": value}
+
+
+def exec_assert_var(executor: Any, args: Mapping[str, Any]) -> Dict[str, Any]:
+    """Assert a flow variable satisfies a condition (assertion DSL)."""
+    from je_auto_control.utils.assertion import assert_variable
+    name = args["name"]
+    return assert_variable(
+        executor.variables.get_value(name), op=str(args.get("op", "eq")),
+        expected=args.get("value"), name=name,
+        raise_on_fail=bool(args.get("raise_on_fail", True)),
+    ).to_dict()
+
+
 def exec_read_file_to_var(executor: Any,
                           args: Mapping[str, Any]) -> Dict[str, Any]:
     """Read a file's text content into a flow variable."""
@@ -603,4 +645,7 @@ def exec_call_macro(executor: Any, args: Mapping[str, Any]) -> Any:
     "AC_read_file_to_var": exec_read_file_to_var,
     "AC_http_to_var": exec_http_to_var,
     "AC_transform_var": exec_transform_var,
+    "AC_now_to_var": exec_now_to_var,
+    "AC_random_to_var": exec_random_to_var,
+    "AC_assert_var": exec_assert_var,
 }

je_auto_control/utils/qr/__init__.py

@@ -0,0 +1,4 @@
+"""Decode QR codes from an image or screen region (OpenCV)."""
+from je_auto_control.utils.qr.qr import read_qr_codes
+
+__all__ = ["read_qr_codes"]

je_auto_control/utils/qr/qr.py

@@ -0,0 +1,49 @@
+"""Decode QR codes from an image or screen region using OpenCV.
+
+OpenCV's ``QRCodeDetector`` ships with ``je_open_cv`` (already a
+dependency), so no extra package is needed. The decoder is injectable so
+the wrapper is unit-testable without a real QR image. GUI-free.
+"""
+import io
+from pathlib import Path
+from typing import Any, Callable, List, Optional, Sequence, Union
+
+from PIL import Image
+
+ImageSource = Union[str, Path, bytes, Image.Image]
+QRDecoder = Callable[[Any], List[str]]
+
+
+def _load_np(source: ImageSource, region: Optional[Sequence[int]]):
+    import numpy as np
+    if isinstance(source, Image.Image):
+        image = source
+    elif isinstance(source, bytes):
+        image = Image.open(io.BytesIO(source))
+    else:
+        image = Image.open(str(source))
+    image = image.convert("RGB")
+    if region is not None:
+        image = image.crop(tuple(int(v) for v in region))
+    return np.array(image)
+
+
+def _default_decoder(image_np: Any) -> List[str]:
+    import cv2
+    detector = cv2.QRCodeDetector()
+    ok, decoded, _points, _straight = detector.detectAndDecodeMulti(image_np)
+    if not ok or decoded is None:
+        return []
+    return [text for text in decoded if text]
+
+
+def read_qr_codes(source: ImageSource,
+                  region: Optional[Sequence[int]] = None, *,
+                  decoder: Optional[QRDecoder] = None) -> List[str]:
+    """Decode every QR code in ``source`` (or a sub-region); return the texts.
+
+    ``source`` is a path, PNG bytes, or a PIL image. ``decoder`` is
+    injectable for tests; the default uses OpenCV's ``QRCodeDetector``.
+    """
+    image_np = _load_np(source, region)
+    return (decoder or _default_decoder)(image_np)

test/unit_test/headless/test_flow_var_commands.py

@@ -1,8 +1,17 @@
-"""Tests for flow variable commands: read-file, http, transform."""
+"""Tests for flow variable commands: read-file, http, transform, now,
+random, assert-var."""
+import datetime
+
+import pytest
+
+from je_auto_control.utils.exception.exceptions import (
+    AutoControlAssertionException,
+)
 from je_auto_control.utils.executor import flow_control
 from je_auto_control.utils.executor.action_executor import Executor
 from je_auto_control.utils.executor.flow_control import (
-    exec_http_to_var, exec_read_file_to_var, exec_transform_var,
+    exec_assert_var, exec_http_to_var, exec_now_to_var, exec_random_to_var,
+    exec_read_file_to_var, exec_transform_var,
 )
 
 
@@ -62,3 +71,44 @@ def test_transform_var_replace_and_slice():
     exec_transform_var(executor, {"name": "v", "op": "slice",
                                   "start": 0, "end": 3, "into": "s"})
     assert executor.variables.get_value("s") == "foo"
+
+
+def test_now_to_var_formats_injected_clock(monkeypatch):
+    monkeypatch.setattr(flow_control, "_now",
+                        lambda: datetime.datetime(2026, 1, 2, 3, 4, 5))
+    executor = Executor()
+    result = exec_now_to_var(executor, {"format": "%Y-%m-%d", "var": "d"})
+    assert result["value"] == "2026-01-02"
+    assert executor.variables.get_value("d") == "2026-01-02"
+
+
+def test_random_to_var_int_within_fixed_range():
+    executor = Executor()
+    exec_random_to_var(executor, {"kind": "int", "min": 1, "max": 1, "var": "r"})
+    assert executor.variables.get_value("r") == 1
+
+
+def test_random_to_var_choice():
+    executor = Executor()
+    exec_random_to_var(executor, {"kind": "choice", "choices": ["only"],
+                                  "var": "c"})
+    assert executor.variables.get_value("c") == "only"
+
+
+def test_assert_var_passes_then_raises():
+    executor = Executor()
+    executor.variables.set("v", "Sam")
+    result = exec_assert_var(executor, {"name": "v", "op": "eq",
+                                        "value": "Sam"})
+    assert result["passed"] is True
+    with pytest.raises(AutoControlAssertionException):
+        exec_assert_var(executor, {"name": "v", "op": "eq", "value": "Nope"})
+
+
+def test_assert_var_regex_match():
+    executor = Executor()
+    executor.variables.set("v", "Order #12345")
+    result = exec_assert_var(executor, {"name": "v", "op": "regex",
+                                        "value": r"#\d+",
+                                        "raise_on_fail": False})
+    assert result["passed"] is True