android-fs5/lib/input.sh

#!/usr/bin/env bash
# lib/input.sh — ADB input automation and screenshot functions for android-fs5
# shellcheck shell=bash

# adb_screenshot [OUTPUT_PATH] — capture device screen to PNG
# Returns path of saved file
adb_screenshot() {
  local output="${1:-/tmp/fs5-screenshot-$(date +%s).png}"
  adb_cmd shell screencap -p > "${output}"
  echo "${output}"
}

# adb_tap X Y — send tap event at coordinates
adb_tap() {
  local x="${1}" y="${2}"
  adb_cmd shell input tap "${x}" "${y}"
}

# adb_swipe X1 Y1 X2 Y2 [DURATION_MS] — send swipe gesture
adb_swipe() {
  local x1="${1}" y1="${2}" x2="${3}" y2="${4}"
  local duration="${5:-300}"
  adb_cmd shell input swipe "${x1}" "${y1}" "${x2}" "${y2}" "${duration}"
}

# adb_type TEXT — type text into focused input field
adb_type() {
  local text="${1}"
  # Escape spaces for adb input text
  local escaped="${text// /%s}"
  adb_cmd shell input text "${escaped}"
}

# adb_key KEYCODE — send Android key event
# Common: KEYCODE_HOME, KEYCODE_BACK, KEYCODE_POWER, KEYCODE_ENTER
adb_key() {
  local keycode="${1}"
  adb_cmd shell input keyevent "${keycode}"
}

# adb_launch PACKAGE — launch app by package name
adb_launch() {
  local pkg="${1}"
  local result exit_code
  result=$(adb_cmd shell monkey -p "${pkg}" -c android.intent.category.LAUNCHER 1 2>&1) || exit_code=$?
  exit_code="${exit_code:-0}"
  if echo "${result}" | grep -q "Events injected: 1"; then
    return 0
  elif [[ "${exit_code}" -ne 0 ]] || echo "${result}" | grep -q "No activities found\|monkey aborted"; then
    out_error "Package not found or has no launcher activity: ${pkg}"
    return 1
  else
    out_error "Failed to launch: ${pkg}"
    return 1
  fi
}

# adb_ui_dump [OUTPUT_PATH] — dump UI hierarchy via UIAutomator
# Returns local path to the XML file
adb_ui_dump() {
  local output="${1:-/tmp/fs5-ui-$(date +%s).xml}"
  local device_path="/sdcard/fs5_ui_dump.xml"
  # UIAutomator prints permission warnings to stderr but still works
  adb_cmd shell uiautomator dump "${device_path}" >/dev/null 2>&1
  adb_cmd pull "${device_path}" "${output}" >/dev/null 2>&1
  adb_cmd shell rm -f "${device_path}" >/dev/null 2>&1
  echo "${output}"
}

# adb_ui_find TEXT [XML_PATH] — find UI element by text/content-desc, return bounds center
# Returns "x y" coordinates of element center, or empty if not found
adb_ui_find() {
  local search_text="${1}"
  local xml_path="${2:-}"

  if [[ -z "${xml_path}" ]]; then
    xml_path=$(adb_ui_dump)
  fi

  python3 - "${xml_path}" "${search_text}" <<'PYEOF'
import sys, xml.etree.ElementTree as ET, re

xml_path = sys.argv[1]
search = sys.argv[2].lower()

try:
    tree = ET.parse(xml_path)
except Exception as e:
    sys.exit(1)

root = tree.getroot()
for node in root.iter('node'):
    text = node.get('text', '').lower()
    desc = node.get('content-desc', '').lower()
    rid  = node.get('resource-id', '').lower()
    if search in text or search in desc or search in rid:
        bounds = node.get('bounds', '')
        # Parse [x1,y1][x2,y2]
        m = re.findall(r'\d+', bounds)
        if len(m) == 4:
            cx = (int(m[0]) + int(m[2])) // 2
            cy = (int(m[1]) + int(m[3])) // 2
            print(f"{cx} {cy}")
            sys.exit(0)
sys.exit(1)
PYEOF
}

# adb_ui_elements [XML_PATH] — list all interactive UI elements with bounds
# Returns JSON array of {text, resource_id, class, bounds, clickable, center_x, center_y}
adb_ui_elements() {
  local xml_path="${1:-}"

  if [[ -z "${xml_path}" ]]; then
    xml_path=$(adb_ui_dump)
  fi

  python3 - "${xml_path}" <<'PYEOF'
import sys, xml.etree.ElementTree as ET, re, json

xml_path = sys.argv[1]
try:
    tree = ET.parse(xml_path)
except Exception as e:
    print(json.dumps([]))
    sys.exit(0)

root = tree.getroot()
elements = []
for node in root.iter('node'):
    text      = node.get('text', '')
    desc      = node.get('content-desc', '')
    rid       = node.get('resource-id', '')
    cls       = node.get('class', '')
    bounds    = node.get('bounds', '')
    clickable = node.get('clickable', 'false') == 'true'
    enabled   = node.get('enabled', 'false') == 'true'

    # Include clickable elements or elements with text/desc
    if not (clickable or text or desc):
        continue

    cx, cy = None, None
    m = re.findall(r'\d+', bounds)
    if len(m) == 4:
        cx = (int(m[0]) + int(m[2])) // 2
        cy = (int(m[1]) + int(m[3])) // 2

    elements.append({
        'text':        text,
        'content_desc': desc,
        'resource_id': rid,
        'class':       cls.split('.')[-1],  # short class name
        'bounds':      bounds,
        'clickable':   clickable,
        'enabled':     enabled,
        'center_x':    cx,
        'center_y':    cy,
    })

print(json.dumps(elements, indent=2))
PYEOF
}

# adb_ocr [SCREENSHOT_PATH] — screenshot + OCR, return extracted text
adb_ocr() {
  local screenshot
  screenshot=$(adb_screenshot)
  if ! command -v tesseract &>/dev/null; then
    out_error "tesseract not found. Install: sudo pacman -S tesseract tesseract-data-eng"
    return 1
  fi
  # Locate tessdata — try common paths
  local tessdata=""
  for d in /usr/share/tessdata /usr/share/tesseract-ocr/4.00/tessdata \
            /usr/share/tesseract-ocr/tessdata /usr/local/share/tessdata; do
    if [[ -f "${d}/eng.traineddata" ]]; then
      tessdata="${d}"
      break
    fi
  done
  if [[ -z "${tessdata}" ]]; then
    out_error "tesseract eng language data not found. Install: sudo pacman -S tesseract-data-eng"
    return 1
  fi
  local txt_base="${screenshot%.png}"
  TESSDATA_PREFIX="${tessdata}" tesseract "${screenshot}" "${txt_base}" -l eng --psm 6 quiet 2>/dev/null
  if [[ -f "${txt_base}.txt" ]]; then
    cat "${txt_base}.txt"
    rm -f "${txt_base}.txt"
  fi
  rm -f "${screenshot}"
}

# _kdeconnect_device_id — find the paired+reachable FS5 device ID via kdeconnect-cli
# Returns device ID or empty string if not paired/reachable
_kdeconnect_device_id() {
  # --list-available returns only paired AND reachable devices
  kdeconnect-cli --list-available --id-only 2>/dev/null | head -1 | tr -d '\r\n'
}

# _kdeconnect_check — verify KDE Connect is paired; print setup instructions if not
_kdeconnect_check() {
  local dev_id
  dev_id=$(_kdeconnect_device_id)
  if [[ -z "${dev_id}" ]]; then
    out_error "KDE Connect: no paired device found."
    echo "Setup steps:"
    echo "  1. Install KDE Connect on Linux: sudo pacman -S kdeconnect"
    echo "  2. Open KDE Connect app on the FS5 phone"
    echo "  3. Accept the pairing request on both devices"
    echo "  4. Grant SMS permission in KDE Connect app on phone"
    echo "  5. Re-run this command"
    return 1
  fi
  echo "${dev_id}"
}

# adb_sms_list [LIMIT] — list SMS conversations via KDE Connect DBus
# Returns JSON array of {id, message_count} (KDE Connect 21+ only provides metadata via DBus)
# Full message content requires KDE Connect app on phone; OTPs use notifications plugin
# Requires: KDE Connect paired + SMS plugin + SMS permission granted on phone
adb_sms_list() {
  local limit="${1:-10}"
  local dev_id
  dev_id=$(_kdeconnect_check) || return 1

  local dev_path="/modules/kdeconnect/devices/${dev_id}"
  local sms_path="${dev_path}/sms"

  # KDE Connect 21+ DBus SMS API:
  # requestAllConversations → fires conversationLoaded(int64 id, uint64 count) signals
  # Message body content is NOT exposed via DBus in KDE Connect 25.x
  # Use select() for non-blocking reads to avoid hanging
  python3 /dev/stdin "${dev_id}" "${limit}" <<'PYEOF'
import sys, json, subprocess, time, re, select

dev_id = sys.argv[1]
limit = int(sys.argv[2])
sms_path = f"/modules/kdeconnect/devices/{dev_id}/sms"

# Start monitoring
proc = subprocess.Popen(
    ["dbus-monitor", "--session", "sender='org.kde.kdeconnect'"],
    stdout=subprocess.PIPE, stderr=subprocess.DEVNULL, text=True
)
time.sleep(0.3)

# Request all conversations
subprocess.run(
    ["gdbus", "call", "--session", "--dest", "org.kde.kdeconnect",
     "--object-path", sms_path,
     "--method", "org.kde.kdeconnect.device.sms.requestAllConversations"],
    capture_output=True, text=True)

conv_list = []
deadline = time.time() + 6
last_signal = time.time()

while time.time() < deadline:
    ready = select.select([proc.stdout], [], [], 0.5)
    if not ready[0]:
        if conv_list and (time.time() - last_signal) > 1.5:
            break
        continue
    line = proc.stdout.readline()
    if "member=conversationLoaded" in line:
        last_signal = time.time()
        id_line = proc.stdout.readline()
        count_line = proc.stdout.readline()
        m_id = re.search(r'int64\s+(\d+)', id_line)
        m_cnt = re.search(r'uint64\s+(\d+)', count_line)
        if m_id:
            conv_list.append({
                "id": int(m_id.group(1)),
                "message_count": int(m_cnt.group(1)) if m_cnt else 0
            })
            if len(conv_list) >= limit:
                break

proc.terminate()
print(json.dumps(conv_list[:limit], indent=2))
PYEOF
}

# adb_sms_otp [WAIT_SECS] — extract OTP code from KDE Connect notifications
# Watches for SMS/messaging app notifications containing 4-8 digit codes
# If WAIT_SECS > 0, polls until new OTP arrives or timeout
adb_sms_otp() {
  local wait_secs="${1:-0}"
  local dev_id
  dev_id=$(_kdeconnect_check) || return 1

  local deadline=$(( $(date +%s) + wait_secs ))
  local notif_base="/modules/kdeconnect/devices/${dev_id}/notifications"

  # Poll KDE Connect notifications DBus for OTP patterns
  while true; do
    local otp
    # Get active notification IDs
    local notif_ids
    notif_ids=$(gdbus call --session --dest org.kde.kdeconnect \
      --object-path "${notif_base}" \
      --method org.kde.kdeconnect.device.notifications.activeNotifications \
      2>/dev/null | grep -oE "'[^']+'" | tr -d "'" | tr ',' '\n' | tr -d ' []')

    for nid in ${notif_ids}; do
      local props
      props=$(gdbus call --session --dest org.kde.kdeconnect \
        --object-path "${notif_base}/${nid}" \
        --method org.freedesktop.DBus.Properties.GetAll \
        "org.kde.kdeconnect.device.notifications.notification" 2>/dev/null)
      # Extract OTP from text/ticker fields
      otp=$(echo "${props}" | grep -oE '\b[0-9]{4,8}\b' | head -1)
      if [[ -n "${otp}" ]]; then
        echo "${otp}"
        return 0
      fi
    done

    if [[ "${wait_secs}" -le 0 ]] || [[ $(date +%s) -ge ${deadline} ]]; then
      break
    fi
    sleep 3
  done

  out_error "No OTP found in KDE Connect notifications"
  return 1
}