ibm3270

60class ConnectionState(StrEnum):
61    Connected = "C"
62    NotConnected = "N"

65class EmulatorMode(StrEnum):
66    Mode3270 = "I"
67    NVTLine = "L"
68    NVTCharacter = "C"
69    Unnegotiated = "P"
70    NotConnected = "N"

168@dataclass
169class FieldDefinition:
170    """Descriptor for a single screen field used with `Terminal.read_many`.
171
172    Attributes:
173        row: 1-based row.
174        col: 1-based column.
175        length: Number of characters to read.
176        type: `"string"` (default) or `"number"` for float coercion.
177        trim: Strip trailing whitespace. Defaults to `True`.
178        name: Optional key in the result dict. Defaults to `"row,col"`.
179    """
180
181    row: int
182    col: int
183    length: int
184    type: Literal["string", "number"] = "string"
185    trim: bool = True
186    name: str | None = None

55class FieldProtection(StrEnum):
56    Protected = "P"
57    Unprotected = "U"

44class KeyboardState(StrEnum):
45    Unlocked = "U"
46    Locked = "L"
47    Error = "E"

50class ScreenFormatting(StrEnum):
51    Formatted = "F"
52    Unformatted = "U"

118@dataclass
119class ScreenPosition:
120    """1-based (row, col) coordinate on the terminal screen."""
121
122    row: int
123    col: int

126@dataclass
127class ScreenSize:
128    """Terminal screen dimensions in rows and columns."""
129
130    rows: int
131    cols: int

19class SessionBusyError(SessionError):
20    """Raised when a second operation is attempted while one is in flight."""

15class SessionDisconnectedError(SessionError):
16    """Raised when an operation requires a running session."""

7class SessionError(Exception):
8    """Base exception for session-level failures."""

13class SessionManager:
14    """Manage multiple isolated terminal sessions.
15
16    Thread-safe registry of `Terminal` instances keyed
17    by auto-generated UUID session IDs.
18    """
19
20    def __init__(self) -> None:
21        self._sessions: dict[str, Terminal] = {}
22        self._lock = Lock()
23
24    def create_session(self, options: TerminalOptions | None = None, *, start: bool = False) -> str:
25        """Create a new terminal session and register it.
26
27        Args:
28            options: Configuration for the underlying s3270 process. Defaults to `TerminalOptions()`.
29            start: If `True`, call `Terminal.start` immediately.
30
31        Returns:
32            The new session ID (UUID string).
33        """
34        session_id = str(uuid4())
35        session = Terminal(options=options, session_id=session_id)
36        if start:
37            session.start()
38        with self._lock:
39            self._sessions[session_id] = session
40        return session_id
41
42    def get_session(self, session_id: str) -> Terminal:
43        """Retrieve a session by ID.
44
45        Args:
46            session_id: The UUID string returned by `create_session`.
47
48        Returns:
49            The `Terminal` for that session.
50
51        Raises:
52            KeyError: If *session_id* is not registered.
53        """
54        with self._lock:
55            session = self._sessions.get(session_id)
56        if session is None:
57            raise KeyError(f"Unknown session id: {session_id}")
58        return session
59
60    def close_session(self, session_id: str) -> None:
61        """Stop and remove a session.
62
63        Args:
64            session_id: The UUID string of the session to close.
65
66        No-op if the session ID is unknown.
67        """
68        with self._lock:
69            session = self._sessions.pop(session_id, None)
70        if session is not None:
71            session.stop()
72
73    def close_all(self) -> None:
74        """Stop and remove all registered sessions."""
75        with self._lock:
76            sessions = list(self._sessions.values())
77            self._sessions.clear()
78        for session in sessions:
79            session.stop()
80
81    def list_sessions(self) -> list[str]:
82        """Return the IDs of all registered sessions."""
83        with self._lock:
84            return list(self._sessions.keys())

23class SessionProcessError(SessionError):
24    """Raised when the emulator process dies or I/O fails unexpectedly."""

73class SessionState(StrEnum):
74    """Lifecycle state of a `Terminal` session."""
75
76    Initial = "INITIAL"
77    Started = "STARTED"
78    Connected = "CONNECTED"
79    Disconnected = "DISCONNECTED"
80    Stopped = "STOPPED"
81    Failed = "FAILED"

11class SessionTimeoutError(SessionError, TimeoutError):
12    """Raised when a blocking session operation exceeds its timeout."""

34class StatusFlag(StrEnum):
35    """Boolean status fields readable via `Terminal.is_`."""
36
37    Formatted = "Formatted"
38    KeyboardLock = "KeyboardLock"
39    Printer = "Printer"
40    Secure = "Secure"
41    Tn3270e = "Tn3270e"

134@dataclass
135class StatusInfo:
136    """Parsed s3270 status line, updated after each successful command.
137
138    Attributes:
139        keyboard_state: Whether the keyboard is unlocked, locked, or in error.
140        screen_formatting: Formatted or unformatted screen.
141        field_protection: Protection state of the current field.
142        connection_state: Connected or not connected.
143        host: Hostname from the status line, or `None` if not connected.
144        emulator_mode: Current emulator negotiation mode.
145        model_number: Active 3270 model number.
146        rows: Screen height in rows.
147        cols: Screen width in columns.
148        cursor_row: 0-based cursor row from s3270.
149        cursor_col: 0-based cursor column from s3270.
150        window_id: X11 window ID or `"0"` on non-X platforms.
151        command_execution_time: Last command latency in seconds, or `None`.
152    """
153    keyboard_state: KeyboardState
154    screen_formatting: ScreenFormatting
155    field_protection: FieldProtection
156    connection_state: ConnectionState
157    host: str | None
158    emulator_mode: EmulatorMode
159    model_number: int
160    rows: int
161    cols: int
162    cursor_row: int
163    cursor_col: int
164    window_id: str
165    command_execution_time: float | None

 37class Terminal:
 38    """High-level synchronous wrapper around a single s3270 process.
 39
 40    Manages the full lifecycle of a terminal session: start, connect, interact,
 41    disconnect, and stop. All operations are synchronous and blocking.
 42
 43    Args:
 44        options: Configuration for the s3270 process. Defaults to `TerminalOptions()`.
 45        session_id: Stable identifier for this session. Auto-generated if omitted.
 46    """
 47
 48    def __init__(
 49        self,
 50        options: TerminalOptions | None = None,
 51        *,
 52        session_id: str | None = None,
 53    ) -> None:
 54        self._options = options or TerminalOptions()
 55        self._session_id = session_id or str(uuid4())
 56        self._transport = Transport(default_timeout_ms=self._options.timeout)
 57        self._connected: bool = False
 58        self._screen_buffer: list[str] = []
 59        self._last_status_info: StatusInfo | None = None
 60        self._state = SessionState.Initial
 61
 62    # ------------------------------------------------------------------
 63    # Internal helpers
 64    # ------------------------------------------------------------------
 65
 66    def _assert_running(self) -> None:
 67        if not self._transport.available():
 68            self._state = SessionState.Failed
 69            raise SessionDisconnectedError("Terminal is not running")
 70
 71    def _get_status_field(self, flag: StatusFlag) -> str:
 72        if self._last_status_info is None:
 73            return "false"
 74        s = self._last_status_info
 75        if flag == StatusFlag.Formatted:
 76            return "true" if s.screen_formatting == ScreenFormatting.Formatted else "false"
 77        if flag == StatusFlag.KeyboardLock:
 78            return "false" if s.keyboard_state == KeyboardState.Unlocked else "true"
 79        if flag == StatusFlag.Tn3270e:
 80            return "true" if s.emulator_mode == EmulatorMode.Mode3270 else "false"
 81        return "false"
 82
 83    def _to_emulator_coordinate(self, row: int, col: int) -> tuple[int, int]:
 84        if row < 1:
 85            raise ValueError(f"row must be >= 1, got {row}")
 86        if col < 1:
 87            raise ValueError(f"col must be >= 1, got {col}")
 88        return row - 1, col - 1
 89
 90    def _to_display_position(self, row: int, col: int) -> ScreenPosition:
 91        return ScreenPosition(row + 1, col + 1)
 92
 93    def _resolved_start_args(self) -> list[str]:
 94        args = list(self._options.args)
 95        if "-model" in args:
 96            return args
 97        for index, arg in enumerate(args[:-1]):
 98            if arg == "-xrm" and "s3270.model:" in args[index + 1]:
 99                return args
100        return ["-model", _DEFAULT_MODEL, *args]
101
102    # ------------------------------------------------------------------
103    # Lifecycle
104    # ------------------------------------------------------------------
105
106    def available(self) -> bool:
107        """Return `True` if the underlying s3270 process is running."""
108        return self._transport.available()
109
110    @property
111    def state(self) -> SessionState:
112        """Current lifecycle state of the session."""
113        return self._state
114
115    @property
116    def session_id(self) -> str:
117        """Unique identifier for this session."""
118        return self._session_id
119
120    def start(self) -> None:
121        """Launch the s3270 process. No-op if already running."""
122        if self.available():
123            return
124        self._transport.start(self._options.executable, self._resolved_start_args())
125        self._state = SessionState.Started
126
127    def stop(self) -> None:
128        """Terminate the s3270 process and release resources."""
129        self._transport.stop()
130        self._connected = False
131        self._state = SessionState.Stopped
132
133    # ------------------------------------------------------------------
134    # Command dispatch
135    # ------------------------------------------------------------------
136
137    def command(self, cmd: str, *, timeout: int | None = None) -> TerminalResponse:
138        """Send a raw s3270 command and return the response.
139
140        Args:
141            cmd: The s3270 command string, e.g. `"Enter"` or `"String(text)"`.
142            timeout: Milliseconds to wait for a response. Uses the session default if omitted.
143
144        Returns:
145            A `TerminalResponse` with `ok`, `data`, and `status`.
146
147        Raises:
148            SessionDisconnectedError: If the s3270 process is not running.
149            SessionTimeoutError: If the command does not complete within *timeout*.
150        """
151        self._assert_running()
152        started_at = time.monotonic()
153        try:
154            response = self._transport.execute(cmd, timeout=timeout)
155        except (SessionDisconnectedError, SessionProcessError):
156            self._state = SessionState.Failed
157            raise
158
159        if response.status:
160            self._last_status_info = parse_status(response.status)
161
162        if response.ok:
163            elapsed_ms = int((time.monotonic() - started_at) * 1000)
164            _LOG.debug(
165                "session=%s pid=%s cmd=%s elapsed_ms=%d ok=true",
166                self._session_id,
167                self._transport.pid(),
168                cmd,
169                elapsed_ms,
170            )
171        else:
172            elapsed_ms = int((time.monotonic() - started_at) * 1000)
173            _LOG.debug(
174                "session=%s pid=%s cmd=%s elapsed_ms=%d ok=false data=%s",
175                self._session_id,
176                self._transport.pid(),
177                cmd,
178                elapsed_ms,
179                response.data,
180            )
181        return response
182
183    def run_step(self, cmd: str, *, timeout: int | None = None) -> TerminalResponse:
184        """Alias for `command`. Sends a single raw s3270 command.
185
186        Args:
187            cmd: The s3270 command string.
188            timeout: Milliseconds to wait. Uses the session default if omitted.
189        """
190        return self.command(cmd, timeout=timeout)
191
192    def run_workflow(self, commands: list[str], *, timeout: int | None = None) -> list[TerminalResponse]:
193        """Execute a sequence of raw s3270 commands in order.
194
195        Args:
196            commands: List of s3270 command strings to execute sequentially.
197            timeout: Per-command timeout in milliseconds. Uses the session default if omitted.
198
199        Returns:
200            List of `TerminalResponse` objects, one per command.
201        """
202        return [self.command(cmd, timeout=timeout) for cmd in commands]
203
204    # ------------------------------------------------------------------
205    # Connectivity
206    # ------------------------------------------------------------------
207
208    def connect(
209        self,
210        hostname: str,
211        port: int,
212        *,
213        mode: TerminalMode | None = None,
214        lu_name: str | None = None,
215    ) -> TerminalResponse:
216        """Connect to a TN3270 host.
217
218        Args:
219            hostname: Hostname or IP address of the mainframe.
220            port: TCP port (commonly 23 or 992 for TLS).
221            mode: Optional `TerminalMode` prefix (e.g. `NoTN3270E`).
222            lu_name: Optional LU name for LU-to-LU connections.
223
224        Returns:
225            `TerminalResponse` — `ok` if the connection was established.
226        """
227        addr = f"{hostname.lower()}:{port}"
228        if lu_name:
229            addr = f"{lu_name}@{addr}"
230        if mode:
231            addr = f"{mode.value}:{addr}"
232        resp = self.command(f"Connect({addr})")
233        self._connected = resp.ok
234        if self._connected:
235            self._state = SessionState.Connected
236        return resp
237
238    def disconnect(self) -> TerminalResponse:
239        """Disconnect from the host. No-op (returns ok) if not connected."""
240        if not self._connected:
241            return TerminalResponse(ok=True, data="", status="", raw=[])
242        resp = self.command("Disconnect")
243        self._connected = False
244        self._state = SessionState.Disconnected
245        return resp
246
247    # ------------------------------------------------------------------
248    # Query / status getters
249    # ------------------------------------------------------------------
250
251    def query(self, setting: TerminalSetting) -> TerminalResponse:
252        """Issue a `Query()` command for the given setting.
253
254        Args:
255            setting: The `TerminalSetting` to query (e.g. `Host`, `Model`).
256
257        Returns:
258            `TerminalResponse` with the value in `data`.
259        """
260        return self.command(f"Query({setting.value})")
261
262    def get(self, setting: TerminalSetting) -> str:
263        """Return the string value of a terminal setting.
264
265        Args:
266            setting: The `TerminalSetting` to retrieve.
267
268        Returns:
269            The setting value as a plain string.
270        """
271        resp = self.query(setting)
272        return resp.data
273
274    def cursor(self) -> ScreenPosition | None:
275        """Return the current cursor position (1-based row/col), or `None` if unknown."""
276        if self._last_status_info is None:
277            return None
278        s = self._last_status_info
279        return self._to_display_position(s.cursor_row, s.cursor_col)
280
281    def screen_size(self) -> ScreenSize | None:
282        """Return the current screen dimensions, or `None` if not yet known."""
283        if self._last_status_info is None:
284            return None
285        return ScreenSize(self._last_status_info.rows, self._last_status_info.cols)
286
287    def is_(self, flag: StatusFlag) -> bool:
288        """Test a boolean status flag from the last s3270 status line.
289
290        Args:
291            flag: A `StatusFlag` such as `Formatted` or `KeyboardLock`.
292
293        Returns:
294            `True` if the flag is set.
295        """
296        val = self._get_status_field(flag)
297        if flag == StatusFlag.KeyboardLock:
298            return val != "false"
299        return val in ("true", "1")
300
301    def current_field(self) -> ScreenPosition | None:
302        """Return the position of the current input field (same as `cursor`)."""
303        return self.cursor()
304
305    # ------------------------------------------------------------------
306    # Screen buffer
307    # ------------------------------------------------------------------
308
309    def refresh(self) -> TerminalResponse:
310        """Wait for the host and update the internal screen buffer via `Ascii1()`.
311
312        Returns:
313            `TerminalResponse` from the `Ascii1()` command.
314        """
315        self.wait_ready()
316        resp = self.command("Ascii1()")
317        self._screen_buffer = resp.data.splitlines()
318        return resp
319
320    def screen(self) -> str:
321        """Return the last refreshed screen as a single newline-joined string."""
322        return "\n".join(self._screen_buffer)
323
324    def get_screen_buffer(self) -> list[str]:
325        """Return the last refreshed screen as a list of row strings."""
326        return list(self._screen_buffer)
327
328    # ------------------------------------------------------------------
329    # Read / write / check
330    # ------------------------------------------------------------------
331
332    def read(self, row: int, col: int, length: int, trim: bool = True) -> str:
333        """Read a region from the screen buffer.
334
335        Args:
336            row: 1-based row number.
337            col: 1-based column number.
338            length: Number of characters to read.
339            trim: Strip trailing whitespace from the result. Defaults to `True`.
340
341        Returns:
342            The extracted string, or `""` if the row is out of range.
343        """
344        if row < 1 or row > len(self._screen_buffer):
345            return ""
346        line = self._screen_buffer[row - 1]
347        result = line[col - 1 : col - 1 + length]
348        return result.rstrip() if trim else result
349
350    def write(self, text: str, row: int, col: int, length: int | None = None) -> TerminalResponse:
351        """Move the cursor then type *text* with `String()`.
352
353        Args:
354            text: Text to send.
355            row: 1-based destination row.
356            col: 1-based destination column.
357            length: If provided, right-justify *text* in a field of this width.
358        """
359        if length is not None:
360            text = text[:length].rjust(length)
361        self.move(row, col)
362        return self.string(text)
363
364    def check(self, text: str, row: int, col: int) -> bool:
365        """Return `True` if *text* appears at the given screen position.
366
367        Args:
368            text: Expected string.
369            row: 1-based row.
370            col: 1-based column.
371        """
372        return self.read(row, col, len(text)) == text
373
374    def read_many(self, fields: list[FieldDefinition]) -> FieldDefinitionRecord:
375        """Refresh the screen and read multiple fields in one call.
376
377        Args:
378            fields: List of `FieldDefinition` descriptors.
379
380        Returns:
381            Dict mapping each field's `name` (or `"row,col"` key) to its value.
382            Numeric fields are coerced to `float`; blank numeric fields become `None`.
383        """
384        self.refresh()
385        result: FieldDefinitionRecord = {}
386        for field in fields:
387            raw = self.read(field.row, field.col, field.length, trim=field.trim)
388            key = field.name or f"{field.row},{field.col}"
389            if field.type == "number":
390                try:
391                    result[key] = float(raw) if raw else None
392                except ValueError:
393                    result[key] = raw
394            else:
395                result[key] = raw
396        return result
397
398    # ------------------------------------------------------------------
399    # Text / input helpers
400    # ------------------------------------------------------------------
401
402    def string(self, text: str) -> TerminalResponse:
403        """Send text to the emulator using the `String()` action.
404
405        Args:
406            text: The string to type, including s3270 escape sequences.
407
408        Raises:
409            ValueError: If *text* contains an invalid s3270 escape sequence.
410        """
411        validate_escape_sequences(text)
412        return self.command(f"String({text})")
413
414    def send_text(self, text: str) -> TerminalResponse:
415        """Alias for `string`."""
416        return self.string(text)
417
418    def enter(self) -> TerminalResponse:
419        """Press the Enter key and refresh the screen."""
420        response = self.command("Enter")
421        self.refresh()
422        return response
423
424    def send_enter(self) -> TerminalResponse:
425        """Alias for `enter`."""
426        return self.enter()
427
428    def tab(self) -> TerminalResponse:
429        """Press the Tab key to advance to the next input field."""
430        return self.command("Tab")
431
432    def clear(self) -> TerminalResponse:
433        """Press the Clear key and refresh the screen."""
434        response = self.command("Clear")
435        self.refresh()
436        return response
437
438    def pf(self, n: int) -> TerminalResponse:
439        """Press a PF function key and refresh the screen.
440
441        Args:
442            n: PF key number, 1–24.
443
444        Raises:
445            ValueError: If *n* is outside the range 1–24.
446        """
447        if not 1 <= n <= 24:
448            raise ValueError(f"PF key must be 1–24, got {n}")
449        response = self.command(f"PF({n})")
450        self.refresh()
451        return response
452
453    def send_pf(self, n: int) -> TerminalResponse:
454        """Alias for `pf`."""
455        return self.pf(n)
456
457    def pa(self, n: int) -> TerminalResponse:
458        """Press a PA program-attention key and refresh the screen.
459
460        Args:
461            n: PA key number, 1–3.
462
463        Raises:
464            ValueError: If *n* is outside the range 1–3.
465        """
466        if not 1 <= n <= 3:
467            raise ValueError(f"PA key must be 1–3, got {n}")
468        response = self.command(f"PA({n})")
469        self.refresh()
470        return response
471
472    def move(self, row: int, col: int) -> TerminalResponse:
473        """Move the cursor to the specified screen position.
474
475        Args:
476            row: 1-based row number.
477            col: 1-based column number.
478        """
479        emulator_row, emulator_col = self._to_emulator_coordinate(row, col)
480        return self.command(f"MoveCursor({emulator_row},{emulator_col})")
481
482    def read_screen(self) -> str:
483        """Refresh the screen buffer and return its full text content."""
484        self.refresh()
485        return self.screen()
486
487    def scrape(self, row: int, col: int, length: int, trim: bool = True) -> str:
488        """Alias for `read`."""
489        return self.read(row, col, length, trim=trim)
490
491    # ------------------------------------------------------------------
492    # Wait helpers
493    # ------------------------------------------------------------------
494
495    def wait(self, timeout: int | None = None) -> TerminalResponse:
496        """Issue `Wait()` — blocks until the emulator is ready.
497
498        Args:
499            timeout: Milliseconds to wait. Uses the session default if omitted.
500        """
501        return self.command("Wait()", timeout=timeout)
502
503    def wait_output(self, timeout: int | None = None) -> TerminalResponse:
504        """Issue `Wait(Output)` — blocks until the host sends new screen data.
505
506        Args:
507            timeout: Milliseconds to wait. Uses the session default if omitted.
508        """
509        return self.command("Wait(Output)", timeout=timeout)
510
511    def wait_unlock(self, timeout: int | None = None) -> TerminalResponse:
512        """Issue `Wait(Unlock)` — blocks until the keyboard is unlocked.
513
514        Args:
515            timeout: Milliseconds to wait. Uses the session default if omitted.
516        """
517        return self.command("Wait(Unlock)", timeout=timeout)
518
519    def wait_ready(self, timeout: int | None = None) -> None:
520        """Wait for the keyboard to unlock, then wait for host output.
521
522        Convenience wrapper around `wait_unlock` + `wait_output`.
523
524        Args:
525            timeout: Per-call timeout in milliseconds. Uses the session default if omitted.
526        """
527        self.wait_unlock(timeout)
528        self.wait_output(timeout)
529
530    def wait_for(
531        self,
532        text: str,
533        row: int | None = None,
534        col: int | None = None,
535        timeout: int = 30_000,
536    ) -> bool:
537        """Poll the screen until *text* appears, or *timeout* elapses.
538
539        Args:
540            text: The string to watch for.
541            row: 1-based row to check. Must be paired with *col*.
542            col: 1-based column to check. Must be paired with *row*.
543            timeout: Total wait time in milliseconds (capped at 300 000). Defaults to 30 000.
544
545        Returns:
546            ``True`` if *text* was found before the deadline, ``False`` otherwise.
547
548        Raises:
549            ValueError: If only one of *row* / *col* is provided.
550        """
551        if (row is None) != (col is None):
552            raise ValueError("both row and col must be provided, or neither")
553        timeout = max(0, min(timeout, 300_000))
554        deadline = time.monotonic() + timeout / 1000
555        while time.monotonic() < deadline:
556            self.refresh()
557            if row is not None:
558                if self.read(row, col, len(text)) == text:  # type: ignore[arg-type]
559                    return True
560            else:
561                if text in self.screen():
562                    return True
563            time.sleep(0.1)
564        return False