API Reference

This page documents the public API of Pure3270.

Main Module

pure3270 package init. Exports core classes and functions for 3270 terminal emulation.

class pure3270.Session(host: str | None = None, port: int = 23, ssl_context: Any | None = None, force_mode: str | None = None, allow_fallback: bool = True, enable_trace: bool = False, terminal_type: str = 'IBM-3278-2')

Bases: object

Synchronous wrapper for AsyncSession.

This class provides a synchronous interface to the asynchronous 3270 session. All methods use asyncio.run() to execute async operations.

__init__(host: str | None = None, port: int = 23, ssl_context: Any | None = None, force_mode: str | None = None, allow_fallback: bool = True, enable_trace: bool = False, terminal_type: str = 'IBM-3278-2') → None

Initialize a synchronous session with a dedicated thread and event loop.

connect(host: str | None = None, port: int | None = None, ssl_context: Any | None = None) → None

Connect to the 3270 host synchronously.

Parameters:

• host – Optional host override.

• port – Optional port override (default 23).

• ssl_context – Optional SSL context override.

Raises:

ConnectionError – If connection fails.

send(data: bytes) → None

Send data to the session.

Parameters:

data – Bytes to send.

Raises:

SessionError – If send fails.

read(timeout: float = 5.0) → bytes

Read data from the session.

Parameters:

timeout – Read timeout in seconds.

Returns:

Received bytes.

Raises:

SessionError – If read fails.

get_aid() → int | None

Get AID synchronously (last known AID value).

close() → None

Close the session synchronously.

property connected: bool

Check if session is connected.

get_trace_events() → List[Any]

open(host: str, port: int = 23) → None

Open connection synchronously (s3270 Open() action).

property screen_buffer: ScreenBuffer

Expose screen buffer for sync Session tests.

property input_inhibited: bool

Get Input Inhibited state (keyboard locked).

property tn3270e_mode: bool

Check if session is in TN3270E mode.

Preference order: 1) Async session’s tn3270 mode flag when True 2) Handler’s negotiated_tn3270e flag as a fallback

close_script() → None

Close script synchronously (s3270 CloseScript() action).

ascii(data: bytes) → str

Convert EBCDIC data to ASCII text (s3270 Ascii() action).

Parameters:

data – EBCDIC bytes to convert.

Returns:

ASCII string representation.

ebcdic(text: str) → bytes

Convert ASCII text to EBCDIC data (s3270 Ebcdic() action).

Parameters:

text – ASCII text to convert.

Returns:

EBCDIC bytes representation.

ascii1(byte_val: int) → str

Convert a single EBCDIC byte to ASCII character.

This is a pure conversion utility and does not require a live session.

ebcdic1(char: str) → int

Convert a single ASCII character to EBCDIC byte.

This is a pure conversion utility and does not require a live session.

ascii_field(field_index: int) → str

Convert field content to ASCII text (s3270 AsciiField() action).

Parameters:

field_index – Index of field to convert.

Returns:

ASCII string representation of field content.

cursor_select() → None

Select field at cursor (s3270 CursorSelect() action).

Raises:

SessionError – If not connected.

delete_field() → None

Delete field at cursor (s3270 DeleteField() action).

Raises:

SessionError – If not connected.

string(text: str) → None

Send string to the session (s3270 String() action).

circum_not() → None

Toggle circumvention of field protection (s3270 CircumNot() action).

Raises:

SessionError – If not connected.

script(commands: str) → None

Execute script (s3270 Script() action).

Parameters:

commands – Script commands.

Raises:

SessionError – If not connected.

execute(command: str) → str

Execute external command synchronously (s3270 Execute() action).

info() → str

Get session information synchronously.

query(query_type: str = 'All') → str

Query session information synchronously.

Supported query types (s3270 protocol parity):

• All: Summary information

• CodePage: Host code page

• ConnectionState: Connection state

• Cursor: Cursor position

• Model: 3270 model information

• ScreenCurSize: Current screen dimensions

• Tn3270eOptions: Active TN3270E functions

• TelnetHostOptions: Host TELNET options

• TelnetMyOptions: Client TELNET options

• And 30+ more query types…

move_cursor1(row: int, col: int) → None

Move cursor to specified position (s3270 MoveCursor1 action).

hex_string(hex_data: str) → None

Send raw hex bytes as input (s3270 HexString() action).

set(option: str, value: str) → None

Set option synchronously.

print_text(text: str) → None

Print text synchronously.

snap() → None

Take snapshot synchronously.

show() → None

Show screen synchronously.

trace(on: bool = True) → None

Enable/disable tracing synchronously.

wait(seconds: float = 1.0) → None

Wait synchronously.

sleep(seconds: float = 1.0) → None

Sleep synchronously.

transfer(file: str) → None

Transfer file synchronously.

source(file: str) → None

Source file synchronously.

expect(pattern: str, timeout: float = 10.0) → bool

Expect pattern synchronously.

fail(message: str) → None

Fail with message synchronously.

compose(text: str) → None

Compose text synchronously.

set_field_attribute(field_index: int, attr: str, value: int) → None

Set a screen field attribute via the underlying async session.

This is a synchronous passthrough used by compatibility tests; the underlying implementation is synchronous as well.

cookie(cookie_string: str) → None

Set cookie synchronously.

interrupt() → None

Send interrupt synchronously (s3270 Interrupt() action).

key(keyname: str) → None

Send key synchronously (s3270 Key() action).

submit(aid: int) → None

Submit with AID synchronously.

home() → None

Move cursor to home position.

up() → None

Move cursor up.

down() → None

Move cursor down.

tab() → None

Move cursor to next tab stop.

backtab() → None

Move cursor to previous tab stop.

backspace() → None

Send backspace key.

enter() → None

Send Enter key.

pf(n: str) → None

Send PF key.

pa(n: str) → None

Send PA key.

erase() → None

Erase entire screen.

clear() → None

Clear entire screen (alias for erase).

newline() → None

Move cursor to start of next line.

erase_eof() → None

Erase to end of field.

erase_input() → None

Erase all input fields.

field_end() → None

Move cursor to end of field.

field_mark() → None

Set field mark.

dup() → None

Duplicate character.

pause(seconds: float = 1.0) → None

Pause session.

bell() → None

Ring bell.

left2() → None

Move cursor left by 2.

right() → None

Move cursor right.

right2() → None

Move cursor right by 2.

reset() → None

Reset session.

field_exit() → None

Exit field.

sysreq() → None

Send SysReq key.

attn() → None

Send Attention key.

test() → None

Send Test key.

left() → None

Move cursor left.

class pure3270.AsyncSession(host: str | None = None, port: int = 23, ssl_context: Any | None = None, force_mode: str | None = None, allow_fallback: bool = True, enable_trace: bool = False, terminal_type: str = 'IBM-3278-2', is_printer_session: bool = False)

Bases: object

Asynchronous 3270 session handler.

This class provides an asynchronous interface to the 3270 session. All operations are async and use asyncio for non-blocking I/O.

AID_MAP: Dict[str, int] = {'Attn': 241, 'BackSpace': 248, 'CLEAR': 109, 'Dup': 245, 'Enter': 125, 'PA(1)': 108, 'PA(2)': 110, 'PA(3)': 107, 'PF(1)': 241, 'PF(10)': 122, 'PF(11)': 123, 'PF(12)': 124, 'PF(13)': 193, 'PF(14)': 194, 'PF(15)': 195, 'PF(16)': 196, 'PF(17)': 197, 'PF(18)': 198, 'PF(19)': 199, 'PF(2)': 242, 'PF(20)': 200, 'PF(21)': 201, 'PF(22)': 74, 'PF(23)': 75, 'PF(24)': 76, 'PF(3)': 243, 'PF(4)': 244, 'PF(5)': 245, 'PF(6)': 246, 'PF(7)': 247, 'PF(8)': 248, 'PF(9)': 249, 'RESET': 106, 'SysReq': 240, 'TEST': 17, 'Test': 17}

__init__(host: str | None = None, port: int = 23, ssl_context: Any | None = None, force_mode: str | None = None, allow_fallback: bool = True, enable_trace: bool = False, terminal_type: str = 'IBM-3278-2', is_printer_session: bool = False) → None

Initialize the AsyncSession.

Parameters:

• host – Host to connect to.

• port – Port to connect to.

• ssl_context – SSL context for secure connections.

• force_mode – Force specific TN3270 mode.

• allow_fallback – Allow fallback to TN3270 if TN3270E fails.

• enable_trace – Enable tracing.

• terminal_type – Terminal model type.

• is_printer_session – True if this is a printer session.

property host: str | None

Public host accessor for tests.

property port: int

Public port accessor for tests.

property model: str

property screen_buffer: ScreenBuffer

Get the screen buffer.

property screen: ScreenBuffer

Get the screen buffer (alias for screen_buffer).

property input_inhibited: bool

Get Input Inhibited state (keyboard locked).

property handler: TN3270Handler | None

Get the underlying TN3270 handler (may be None if not connected).

property tn3270e_mode: bool

Check if session is in TN3270E mode.

property connected: bool

Check if session is connected.

async connect(host: str | None = None, port: int | None = None, ssl_context: Any | None = None) → None

Connect to the 3270 host asynchronously.

async send(data: bytes) → None

Send data to the session with retry logic.

async send_data(data: bytes) → None

Send data to the session (alias for send).

async read(timeout: float = 5.0) → bytes

Read data from the session with retry logic.

async receive_data(timeout: float = 5.0) → bytes

Receive data from the session (alias for read).

get_aid() → int | None

Get the last AID value.

async close() → None

Close the session.

async close_script() → None

Close script (s3270 CloseScript() action).

get_trace_events() → List[Any]

Get trace events.

managed() → AsyncIterator[AsyncSession]

Provide a managed async context that guarantees close() on exit.

ascii(data: bytes) → str

Convert EBCDIC to ASCII.

ebcdic(text: str) → bytes

Convert ASCII to EBCDIC.

ascii1(byte_val: int) → str

Convert single EBCDIC byte to ASCII.

ebcdic1(char: str) → int

Convert single ASCII char to EBCDIC.

ascii_field(field_index: int) → str

Convert field to ASCII.

async cursor_select() → None

Select field at cursor.

async delete_field() → None

Delete field at cursor.

async insert_text(text: str) → None

Insert text at cursor.

async string(text: str) → None

Send string to the session.

async circum_not() → None

Toggle circumvention.

async script(commands: str) → None

Execute script command(s).

Supports both simple method names and function-style commands like: - Simple: “connect”, “disconnect” - With args: “String(test)”, “MoveCursor(5,10)”

async execute(command: str) → str

Execute external command.

async interrupt() → None

Send interrupt.

async key(keyname: str) → None

Send key synchronously (s3270 Key() action).

capabilities() → str

Get capabilities.

async set_option(option: str, value: str) → None

Set option (placeholder for future functionality).

async query(query_type: str = 'All') → str

Query screen.

async set(option: str, value: str) → None

Set option (alias).

async exit() → None

Exit session.

async bell() → None

Ring bell.

async pause(seconds: float = 1.0) → None

Pause.

async ansi_text(data: bytes) → str

Convert EBCDIC to ANSI.

async hex_string(hex_str: str) → bytes

Convert hex to bytes.

async show() → None

Show screen.

async snap() → None

Save snapshot.

async newline() → None

Move cursor to start of next line.

async page_down() → None

Page down: wrap to top row for tests.

async page_up() → None

Page up: move to top row for tests.

async paste_string(text: str) → None

async end() → None

Move cursor to end of current line.

async move_cursor(row: int, col: int) → None

async move_cursor1(row1: int, col1: int) → None

async next_word() → None

async previous_word() → None

async toggle_insert() → None

async flip() → None

async insert() → None

async delete() → None

Delete character at cursor by shifting remainder left and clearing last.

async disconnect() → None

async left() → None

Move cursor left.

async right() → None

Move cursor right.

async left2() → None

Move cursor left by 2.

async right2() → None

Move cursor right by 2.

async mono_case() → None

Toggle monocase.

async nvt_text(text: str) → None

Send NVT text.

async print_text(text: str) → None

Print text.

async prompt(message: str) → str

Prompt for input.

async read_buffer() → bytes

Read buffer.

async reconnect() → None

Reconnect.

async info() → str

Get session information.

async quit() → None

Quit the session.

async home() → None

Move cursor to home position.

async up() → None

Move cursor up.

async down() → None

Move cursor down.

async tab() → None

Send Tab key.

async backtab() → None

Move cursor to previous tab stop.

async backspace() → None

Send backspace key.

async enter() → None

Send Enter key.

async clear() → None

Clear screen locally.

async pf(n: str) → None

Send PF key.

async pa(n: str) → None

Send PA key.

async macro(commands: List[str]) → None

Execute a sequence of commands.

async erase() → None

Erase entire screen (local).

async erase_eof() → None

Erase to end of field.

async erase_input() → None

Erase all input fields.

async field_end() → None

Move cursor to end of field.

async field_mark() → None

Set field mark.

async dup() → None

Duplicate character.

async field_exit() → None

Exit field.

async sysreq() → None

Send SysReq key.

async attn() → None

Send Attention key.

async test() → None

Send Test key.

async screen_trace() → None

Screen trace.

async source(file: str) → None

Source script.

async subject_names() → None

Subject names.

async sys_req(command: str | None = None) → None

Send SysReq.

async send_break() → None

Send Telnet BREAK.

async send_soh_message(status_code: int) → None

Send SOH message.

async toggle_option(option: str) → None

Toggle option.

async trace(on: bool) → None

Enable/disable tracing.

async transfer(file: str) → None

Transfer file.

async send_file(local_path: str, remote_name: str) → None

Send a file to the host using IND$FILE protocol.

async receive_file(remote_name: str, local_path: str) → None

Receive a file from the host using IND$FILE protocol.

async wait_condition(condition: str) → None

Wait for condition.

async compose(text: str) → None

Compose text.

async cookie(cookie_string: str) → None

Set a cookie.

This sets the cookie immediately and returns None. Tests can call this without awaiting since the side-effect happens immediately.

async expect(pattern: str, timeout: float = 10.0) → bool

Wait for pattern.

async fail(message: str) → None

Fail with message.

async select_light_pen(row: int, col: int) → None

Select light pen.

async start_lu_lu_session(lu_name: str) → None

Start LU-LU session.

property lu_lu_session: Any

Get the current LU-LU session if active.

async load_resource_definitions(file_path: str) → None

Load resources.

async apply_resources() → None

Apply resources.

set_field_attribute(field_index: int, attr: str, value: int) → None

Set field attribute.

async submit(aid: int) → None

Submit with AID.

class pure3270.PrinterSession

Bases: object

Advanced TN3270E printer session handler with comprehensive SCS support.

__init__() → None

Initialize the printer session with thread safety.

activate() → None

Activate the printer session with thread safety.

deactivate() → None

Deactivate the printer session with thread safety.

start_new_job(job_id: str = '') → PrinterJob

Start a new printer job.

add_scs_data(data: bytes, holding_lock: bool = False) → None

Add SCS character data to the current job.

Parameters:

• data – SCS data to add

• holding_lock – True if caller already holds self.lock (internal use only)

handle_print_eoj() → None

Handle PRINT-EOJ (End of Job) command.

prune() → None

Prune completed jobs to the last max_jobs.

get_current_job() → PrinterJob | None

Get the current active job.

get_completed_jobs() → List[PrinterJob]

Get the list of completed jobs.

get_job_statistics() → Dict[str, Any]

Get printer job statistics.

clear_completed_jobs() → None

Clear the list of completed jobs.

close() → None

Close the printer session and clear all jobs.

handle_scs_control_code(scs_code: int) → None

Handle SCS control codes with comprehensive support.

process_tn3270e_message(header: TN3270EHeader, data: bytes) → None

Process a TN3270E message for printer session with comprehensive support.

get_session_info() → Dict[str, Any]

Get comprehensive session information.

set_device_type(device_type: str) → None

Set the device type for the printer session.

add_tn3270e_function(function: int) → None

Add a supported TN3270E function.

supports_tn3270e_function(function: int) → bool

Check if a TN3270E function is supported.

get_error_rate() → float

Get the current error rate.

is_healthy() → bool

Check if the session is healthy.

class pure3270.P3270Client(hostName: str | None = None, hostPort: int = 23, ssl: bool = False, *args: Any, **kwargs: Any)

Bases: object

disconnect() → None

Disconnect from TN3270 host and clean up session.

isConnected() → bool

Check if client is connected to TN3270 host.

getScreen() → str

Get the current screen content as text (always ASCII/Unicode).

connect() → None

Establish connection to TN3270 host using pure3270.Session. If no hostName is configured, perform a no-op to match legacy client behavior in test harnesses that call connect() without specifying a host.

numOfInstances = 0

__init__(hostName: str | None = None, hostPort: int = 23, ssl: bool = False, *args: Any, **kwargs: Any) → None

Initialize P3270Client with hostName and hostPort, matching p3270 API.

hostName: str | None

hostPort: int

ssl: bool

printScreen() → str

Print current screen content (returns screen text).

saveScreen(filename: str) → bool

Save current screen to file.

sendText(text: str, asterisks: bool = False) → None

Send text to the current cursor position.

Parameters:

• text – Text to send

• asterisks – If True, mask input (for passwords)

sendEnter() → None

Send Enter key.

sendTab() → None

Send Tab key.

sendBackTab() → None

Send BackTab key.

sendBackSpace() → None

Send BackSpace key.

sendHome() → None

Send Home key.

clearScreen() → None

Clear the screen.

sendPF(pfNum: int) → None

Send PF key (PF1-PF24).

sendPA(paNum: int) → None

Send PA key (PA1-PA3).

sendKeys(keys: str) → None

Send arbitrary key sequence.

moveTo(row: int, col: int) → None

Move cursor to specified position.

moveCursorUp() → None

Move cursor up one row.

moveCursorDown() → None

Move cursor down one row.

moveCursorLeft() → None

Move cursor left one column.

moveCursorRight() → None

Move cursor right one column.

moveToFirstInputField() → None

Move cursor to first input field.

delChar() → None

Delete character at cursor position.

delField() → None

Delete current field.

delWord() → None

Delete current word.

eraseChar() → None

Erase character (same as delete).

endSession() → None

Alias for disconnect (legacy p3270 API).

makeArgs(*args: Any) → List[Any]

Return a list constructed from provided args (legacy helper).

Mirrors p3270.P3270Client.makeArgs behavior by simply returning a list of the arguments passed, used by some calling code that expects an args list for command assembly.

readTextAtPosition(row: int, col: int, length: int) → str

Read text at specified position.

Parameters:

• row – Row position (0-based)

• col – Column position (0-based)

• length – Number of characters to read

Returns:

Text at the specified position

readTextArea(startRow: int, startCol: int, endRow: int, endCol: int) → str

Read text from a rectangular area.

Parameters:

• startRow – Starting row (0-based)

• startCol – Starting column (0-based)

• endRow – Ending row (0-based)

• endCol – Ending column (0-based)

Returns:

Text from the specified area

foundTextAtPosition(row: int, col: int, text: str) → bool

Check if specific text is found at position.

Parameters:

• row – Row position (0-based)

• col – Column position (0-based)

• text – Text to search for

Returns:

True if text is found at position

trySendTextToField(text: str, row: int | None = None, col: int | None = None) → bool

Try to send text to a specific field.

Parameters:

• text – Text to send

• row – Row position (optional)

• col – Column position (optional)

Returns:

True if successful

waitForCursorAt(row: int, col: int, timeout: float = 5.0) → bool

Wait for cursor to be at specific position.

waitForCursorAtOffset(offset: int, timeout: float = 5.0) → bool

Wait for cursor at specific offset from start of screen.

waitForStringAt(row: int, col: int, text: str, timeout: float = 5.0) → bool

Wait for specific text to appear at position.

waitForStringAtOffset(offset: int, text: str, timeout: float = 5.0) → bool

Wait for specific text to appear at offset.

waitForField(timeout: float = 5.0) → bool

Wait for an input field to be available.

waitForFieldAt(row: int, col: int, timeout: float = 5.0) → bool

Wait for input field at specific position.

waitForFieldAtOffset(offset: int, timeout: float = 5.0) → bool

Wait for input field at specific offset.

waitForOutput(timeout: float = 5.0) → bool

Wait for output to be ready.

waitFor3270Mode(timeout: float = 5.0) → bool

Wait for 3270 mode to be active.

waitForNVTMode(timeout: float = 5.0) → bool

Wait for NVT mode to be active.

waitForDisconnect(timeout: float = 5.0) → bool

Wait for disconnection.

waitForUnlock(timeout: float = 5.0) → bool

Wait for keyboard unlock.

waitForTimeout(timeout: float) → bool

Wait for specified timeout period.

close() → None

Alias for disconnect.

send(command: str) → None

Send command (for backwards compatibility).

read() → str

Read screen (for backwards compatibility).

pure3270.setup_logging(level: str = 'WARNING') → None

Setup basic logging configuration.

Parameters:

level – Logging level (DEBUG, INFO, WARNING, ERROR, CRITICAL)

Session Module

Session management for pure3270, handling synchronous and asynchronous 3270 connections.

exception pure3270.session.SessionError(message: str, context: Dict[str, Any] | None = None)

Bases: Exception

Base exception for session-related errors.

When context is provided, include key details in the string representation so tests can assert on them without reaching into attributes.

__init__(message: str, context: Dict[str, Any] | None = None)

exception pure3270.session.ConnectionError(message: str, context: Dict[str, Any] | None = None)

Bases: SessionError

Raised when connection fails.

class pure3270.session.Session(host: str | None = None, port: int = 23, ssl_context: Any | None = None, force_mode: str | None = None, allow_fallback: bool = True, enable_trace: bool = False, terminal_type: str = 'IBM-3278-2')

Bases: object

Synchronous wrapper for AsyncSession.

This class provides a synchronous interface to the asynchronous 3270 session. All methods use asyncio.run() to execute async operations.

__init__(host: str | None = None, port: int = 23, ssl_context: Any | None = None, force_mode: str | None = None, allow_fallback: bool = True, enable_trace: bool = False, terminal_type: str = 'IBM-3278-2') → None

Initialize a synchronous session with a dedicated thread and event loop.

connect(host: str | None = None, port: int | None = None, ssl_context: Any | None = None) → None

Connect to the 3270 host synchronously.

Parameters:

• host – Optional host override.

• port – Optional port override (default 23).

• ssl_context – Optional SSL context override.

Raises:

ConnectionError – If connection fails.

send(data: bytes) → None

Send data to the session.

Parameters:

data – Bytes to send.

Raises:

SessionError – If send fails.

read(timeout: float = 5.0) → bytes

Read data from the session.

Parameters:

timeout – Read timeout in seconds.

Returns:

Received bytes.

Raises:

SessionError – If read fails.

get_aid() → int | None

Get AID synchronously (last known AID value).

close() → None

Close the session synchronously.

property connected: bool

Check if session is connected.

get_trace_events() → List[Any]

open(host: str, port: int = 23) → None

Open connection synchronously (s3270 Open() action).

property screen_buffer: ScreenBuffer

Expose screen buffer for sync Session tests.

property input_inhibited: bool

Get Input Inhibited state (keyboard locked).

property tn3270e_mode: bool

Check if session is in TN3270E mode.

Preference order: 1) Async session’s tn3270 mode flag when True 2) Handler’s negotiated_tn3270e flag as a fallback

close_script() → None

Close script synchronously (s3270 CloseScript() action).

ascii(data: bytes) → str

Convert EBCDIC data to ASCII text (s3270 Ascii() action).

Parameters:

data – EBCDIC bytes to convert.

Returns:

ASCII string representation.

ebcdic(text: str) → bytes

Convert ASCII text to EBCDIC data (s3270 Ebcdic() action).

Parameters:

text – ASCII text to convert.

Returns:

EBCDIC bytes representation.

ascii1(byte_val: int) → str

Convert a single EBCDIC byte to ASCII character.

This is a pure conversion utility and does not require a live session.

ebcdic1(char: str) → int

Convert a single ASCII character to EBCDIC byte.

This is a pure conversion utility and does not require a live session.

ascii_field(field_index: int) → str

Convert field content to ASCII text (s3270 AsciiField() action).

Parameters:

field_index – Index of field to convert.

Returns:

ASCII string representation of field content.

cursor_select() → None

Select field at cursor (s3270 CursorSelect() action).

Raises:

SessionError – If not connected.

delete_field() → None

Delete field at cursor (s3270 DeleteField() action).

Raises:

SessionError – If not connected.

string(text: str) → None

Send string to the session (s3270 String() action).

circum_not() → None

Toggle circumvention of field protection (s3270 CircumNot() action).

Raises:

SessionError – If not connected.

script(commands: str) → None

Execute script (s3270 Script() action).

Parameters:

commands – Script commands.

Raises:

SessionError – If not connected.

execute(command: str) → str

Execute external command synchronously (s3270 Execute() action).

info() → str

Get session information synchronously.

query(query_type: str = 'All') → str

Query session information synchronously.

Supported query types (s3270 protocol parity):

• All: Summary information

• CodePage: Host code page

• ConnectionState: Connection state

• Cursor: Cursor position

• Model: 3270 model information

• ScreenCurSize: Current screen dimensions

• Tn3270eOptions: Active TN3270E functions

• TelnetHostOptions: Host TELNET options

• TelnetMyOptions: Client TELNET options

• And 30+ more query types…

move_cursor1(row: int, col: int) → None

Move cursor to specified position (s3270 MoveCursor1 action).

hex_string(hex_data: str) → None

Send raw hex bytes as input (s3270 HexString() action).

set(option: str, value: str) → None

Set option synchronously.

print_text(text: str) → None

Print text synchronously.

snap() → None

Take snapshot synchronously.

show() → None

Show screen synchronously.

trace(on: bool = True) → None

Enable/disable tracing synchronously.

wait(seconds: float = 1.0) → None

Wait synchronously.

sleep(seconds: float = 1.0) → None

Sleep synchronously.

transfer(file: str) → None

Transfer file synchronously.

source(file: str) → None

Source file synchronously.

expect(pattern: str, timeout: float = 10.0) → bool

Expect pattern synchronously.

fail(message: str) → None

Fail with message synchronously.

compose(text: str) → None

Compose text synchronously.

set_field_attribute(field_index: int, attr: str, value: int) → None

Set a screen field attribute via the underlying async session.

This is a synchronous passthrough used by compatibility tests; the underlying implementation is synchronous as well.

cookie(cookie_string: str) → None

Set cookie synchronously.

interrupt() → None

Send interrupt synchronously (s3270 Interrupt() action).

key(keyname: str) → None

Send key synchronously (s3270 Key() action).

submit(aid: int) → None

Submit with AID synchronously.

home() → None

Move cursor to home position.

up() → None

Move cursor up.

down() → None

Move cursor down.

tab() → None

Move cursor to next tab stop.

backtab() → None

Move cursor to previous tab stop.

backspace() → None

Send backspace key.

enter() → None

Send Enter key.

pf(n: str) → None

Send PF key.

pa(n: str) → None

Send PA key.

erase() → None

Erase entire screen.

clear() → None

Clear entire screen (alias for erase).

newline() → None

Move cursor to start of next line.

erase_eof() → None

Erase to end of field.

erase_input() → None

Erase all input fields.

field_end() → None

Move cursor to end of field.

field_mark() → None

Set field mark.

dup() → None

Duplicate character.

pause(seconds: float = 1.0) → None

Pause session.

bell() → None

Ring bell.

left2() → None

Move cursor left by 2.

right() → None

Move cursor right.

right2() → None

Move cursor right by 2.

reset() → None

Reset session.

field_exit() → None

Exit field.

sysreq() → None

Send SysReq key.

attn() → None

Send Attention key.

test() → None

Send Test key.

left() → None

Move cursor left.

class pure3270.session.AsyncSession(host: str | None = None, port: int = 23, ssl_context: Any | None = None, force_mode: str | None = None, allow_fallback: bool = True, enable_trace: bool = False, terminal_type: str = 'IBM-3278-2', is_printer_session: bool = False)

Bases: object

Asynchronous 3270 session handler.

This class provides an asynchronous interface to the 3270 session. All operations are async and use asyncio for non-blocking I/O.

AID_MAP: Dict[str, int] = {'Attn': 241, 'BackSpace': 248, 'CLEAR': 109, 'Dup': 245, 'Enter': 125, 'PA(1)': 108, 'PA(2)': 110, 'PA(3)': 107, 'PF(1)': 241, 'PF(10)': 122, 'PF(11)': 123, 'PF(12)': 124, 'PF(13)': 193, 'PF(14)': 194, 'PF(15)': 195, 'PF(16)': 196, 'PF(17)': 197, 'PF(18)': 198, 'PF(19)': 199, 'PF(2)': 242, 'PF(20)': 200, 'PF(21)': 201, 'PF(22)': 74, 'PF(23)': 75, 'PF(24)': 76, 'PF(3)': 243, 'PF(4)': 244, 'PF(5)': 245, 'PF(6)': 246, 'PF(7)': 247, 'PF(8)': 248, 'PF(9)': 249, 'RESET': 106, 'SysReq': 240, 'TEST': 17, 'Test': 17}

__init__(host: str | None = None, port: int = 23, ssl_context: Any | None = None, force_mode: str | None = None, allow_fallback: bool = True, enable_trace: bool = False, terminal_type: str = 'IBM-3278-2', is_printer_session: bool = False) → None

Initialize the AsyncSession.

Parameters:

• host – Host to connect to.

• port – Port to connect to.

• ssl_context – SSL context for secure connections.

• force_mode – Force specific TN3270 mode.

• allow_fallback – Allow fallback to TN3270 if TN3270E fails.

• enable_trace – Enable tracing.

• terminal_type – Terminal model type.

• is_printer_session – True if this is a printer session.

resources: Dict[str, str]

property host: str | None

Public host accessor for tests.

property port: int

Public port accessor for tests.

property model: str

property screen_buffer: ScreenBuffer

Get the screen buffer.

property screen: ScreenBuffer

Get the screen buffer (alias for screen_buffer).

property input_inhibited: bool

Get Input Inhibited state (keyboard locked).

property handler: TN3270Handler | None

Get the underlying TN3270 handler (may be None if not connected).

property tn3270e_mode: bool

Check if session is in TN3270E mode.

property connected: bool

Check if session is connected.

async connect(host: str | None = None, port: int | None = None, ssl_context: Any | None = None) → None

Connect to the 3270 host asynchronously.

async send(data: bytes) → None

Send data to the session with retry logic.

async send_data(data: bytes) → None

Send data to the session (alias for send).

async read(timeout: float = 5.0) → bytes

Read data from the session with retry logic.

async receive_data(timeout: float = 5.0) → bytes

Receive data from the session (alias for read).

get_aid() → int | None

Get the last AID value.

async close() → None

Close the session.

async close_script() → None

Close script (s3270 CloseScript() action).

get_trace_events() → List[Any]

Get trace events.

managed() → AsyncIterator[AsyncSession]

Provide a managed async context that guarantees close() on exit.

ascii(data: bytes) → str

Convert EBCDIC to ASCII.

ebcdic(text: str) → bytes

Convert ASCII to EBCDIC.

ascii1(byte_val: int) → str

Convert single EBCDIC byte to ASCII.

ebcdic1(char: str) → int

Convert single ASCII char to EBCDIC.

ascii_field(field_index: int) → str

Convert field to ASCII.

async cursor_select() → None

Select field at cursor.

async delete_field() → None

Delete field at cursor.

async insert_text(text: str) → None

Insert text at cursor.

async string(text: str) → None

Send string to the session.

async circum_not() → None

Toggle circumvention.

async script(commands: str) → None

Execute script command(s).

Supports both simple method names and function-style commands like: - Simple: “connect”, “disconnect” - With args: “String(test)”, “MoveCursor(5,10)”

async execute(command: str) → str

Execute external command.

async interrupt() → None

Send interrupt.

async key(keyname: str) → None

Send key synchronously (s3270 Key() action).

capabilities() → str

Get capabilities.

async set_option(option: str, value: str) → None

Set option (placeholder for future functionality).

async query(query_type: str = 'All') → str

Query screen.

async set(option: str, value: str) → None

Set option (alias).

async exit() → None

Exit session.

async bell() → None

Ring bell.

async pause(seconds: float = 1.0) → None

Pause.

async ansi_text(data: bytes) → str

Convert EBCDIC to ANSI.

async hex_string(hex_str: str) → bytes

Convert hex to bytes.

async show() → None

Show screen.

async snap() → None

Save snapshot.

async newline() → None

Move cursor to start of next line.

async page_down() → None

Page down: wrap to top row for tests.

async page_up() → None

Page up: move to top row for tests.

async paste_string(text: str) → None

async end() → None

Move cursor to end of current line.

async move_cursor(row: int, col: int) → None

async move_cursor1(row1: int, col1: int) → None

async next_word() → None

async previous_word() → None

async toggle_insert() → None

async flip() → None

async insert() → None

async delete() → None

Delete character at cursor by shifting remainder left and clearing last.

async disconnect() → None

async left() → None

Move cursor left.

async right() → None

Move cursor right.

async left2() → None

Move cursor left by 2.

async right2() → None

Move cursor right by 2.

async mono_case() → None

Toggle monocase.

async nvt_text(text: str) → None

Send NVT text.

async print_text(text: str) → None

Print text.

async prompt(message: str) → str

Prompt for input.

async read_buffer() → bytes

Read buffer.

async reconnect() → None

Reconnect.

async info() → str

Get session information.

async quit() → None

Quit the session.

async home() → None

Move cursor to home position.

async up() → None

Move cursor up.

async down() → None

Move cursor down.

async tab() → None

Send Tab key.

async backtab() → None

Move cursor to previous tab stop.

async backspace() → None

Send backspace key.

async enter() → None

Send Enter key.

async clear() → None

Clear screen locally.

async pf(n: str) → None

Send PF key.

async pa(n: str) → None

Send PA key.

async macro(commands: List[str]) → None

Execute a sequence of commands.

async erase() → None

Erase entire screen (local).

async erase_eof() → None

Erase to end of field.

async erase_input() → None

Erase all input fields.

async field_end() → None

Move cursor to end of field.

async field_mark() → None

Set field mark.

async dup() → None

Duplicate character.

async field_exit() → None

Exit field.

async sysreq() → None

Send SysReq key.

async attn() → None

Send Attention key.

async test() → None

Send Test key.

async screen_trace() → None

Screen trace.

async source(file: str) → None

Source script.

async subject_names() → None

Subject names.

async sys_req(command: str | None = None) → None

Send SysReq.

async send_break() → None

Send Telnet BREAK.

async send_soh_message(status_code: int) → None

Send SOH message.

async toggle_option(option: str) → None

Toggle option.

async trace(on: bool) → None

Enable/disable tracing.

async transfer(file: str) → None

Transfer file.

async send_file(local_path: str, remote_name: str) → None

Send a file to the host using IND$FILE protocol.

async receive_file(remote_name: str, local_path: str) → None

Receive a file from the host using IND$FILE protocol.

async wait_condition(condition: str) → None

Wait for condition.

async compose(text: str) → None

Compose text.

async cookie(cookie_string: str) → None

Set a cookie.

This sets the cookie immediately and returns None. Tests can call this without awaiting since the side-effect happens immediately.

async expect(pattern: str, timeout: float = 10.0) → bool

Wait for pattern.

async fail(message: str) → None

Fail with message.

async select_light_pen(row: int, col: int) → None

Select light pen.

async start_lu_lu_session(lu_name: str) → None

Start LU-LU session.

property lu_lu_session: Any

Get the current LU-LU session if active.

async load_resource_definitions(file_path: str) → None

Load resources.

async apply_resources() → None

Apply resources.

set_field_attribute(field_index: int, attr: str, value: int) → None

Set field attribute.

async submit(aid: int) → None

Submit with AID.

Patching Module

Minimal patching stub to satisfy tests.

The test suite sometimes patches pure3270.patching.enable_replacement; provide a no-op implementation here to avoid AttributeError during tests that import it.

pure3270.patching.patching.enable_replacement() → None

No-op replacement enabler for compatibility with p3270 patching tests.

Real patching is out of scope for these unit tests; this function simply exists so unittest.mock.patch can reference it without raising errors.

Emulation Modules

Screen Buffer

Screen buffer management for 3270 emulation.

class pure3270.emulation.screen_buffer.Field(start: Tuple[int, int], end: Tuple[int, int], protected: bool = False, numeric: bool = False, modified: bool = False, selected: bool = False, intensity: int = 0, color: int = 0, background: int = 0, validation: int = 0, outlining: int = 0, character_set: int = 0, sfe_highlight: int = 0, light_pen: int = 0, content: bytes | None = None, codepage: str = 'cp037')

Bases: object

A simple Field object that mirrors the previous namedtuple API with convenient defaults and helper methods used by tests.

It intentionally implements get_content, set_content, and _replace to be compatible with test expectations. content is stored as raw EBCDIC bytes; get_content/set_content use the EBCDICCodec by default but will fall back to EmulationEncoder when the codec isn’t present.

__init__(start: Tuple[int, int], end: Tuple[int, int], protected: bool = False, numeric: bool = False, modified: bool = False, selected: bool = False, intensity: int = 0, color: int = 0, background: int = 0, validation: int = 0, outlining: int = 0, character_set: int = 0, sfe_highlight: int = 0, light_pen: int = 0, content: bytes | None = None, codepage: str = 'cp037') → None

get_content(codec: EBCDICCodec | None = None) → str

Return decoded content as a Unicode string.

Accepts an optional codec for decoding; if not provided we use EBCDICCodec() when available, otherwise EmulationEncoder.

set_content(text: str, codec: EBCDICCodec | None = None) → None

Set the field content from a Unicode string and mark modified.

Uses optional codec for encoding; falls back to EmulationEncoder.

class pure3270.emulation.screen_buffer.ScreenBuffer(rows: int = 24, cols: int = 80, init_value: int = 64, codepage: str = 'cp037')

Bases: BufferWriter

property ascii_buffer: str

Return the entire screen buffer as a multi-line ASCII string.

Tests expect exactly rows lines separated by newlines, regardless of trailing blank content. This method converts each row of the EBCDIC buffer to printable ASCII, masking non-printables to spaces, and joins all rows with newlines without trimming.

In ASCII mode, returns the buffer content directly as ASCII text.

__init__(rows: int = 24, cols: int = 80, init_value: int = 64, codepage: str = 'cp037')

Initialize the ScreenBuffer.

Parameters:

• rows – Number of rows (default 24).

• cols – Number of columns (default 80).

• init_value – Initial value for buffer (default 0x40 for space).

set_ascii_mode(ascii_mode: bool) → None

Set ASCII mode for the screen buffer.

In ASCII mode, characters are stored as ASCII bytes directly. In EBCDIC mode (default), characters are converted to EBCDIC.

Parameters:

ascii_mode – True for ASCII mode, False for EBCDIC mode

get_ascii_mode() → bool

Get current ASCII mode setting.

property input_inhibited: bool

Get Input Inhibited state (keyboard locked).

get_field_content(field_index: int) → str

Return the content of the field at the given index as a decoded string.

read_modified_fields() → list[tuple[tuple[int, int], str]]

Return a list of tuples (position, content) for modified fields.

begin_bulk_update() → None

Suspend field detection until end_bulk_update is called.

end_bulk_update() → None

Resume field detection and run a single detection pass.

clear() → None

Clear the screen buffer and reset fields.

is_ascii_mode() → bool

Return True if screen buffer is in ASCII mode.

set_position(row: int, col: int, wrap: bool = False, strict: bool = False) → None

Set cursor position with bounds checking or wrapping.

Parameters:

• row – Target row position

• col – Target column position

• wrap – If True, allow wrapping to next line; if False, check bounds

• strict – If True, raise IndexError on out of bounds; if False, clamp values

Raises:

IndexError – When strict=True and position is out of bounds

write_char(ebcdic_byte: int, row: int | None = None, col: int | None = None, protected: bool = False, circumvent_protection: bool = False, mark_field_start: bool = False) → None

Write an EBCDIC character to the buffer at position.

Parameters:

• ebcdic_byte – EBCDIC byte value.

• row – Row position.

• col – Column position.

• protected – Protection attribute to set.

• circumvent_protection – If True, write even to protected fields.

• mark_field_start – If True, mark this position as a field start.

update_from_stream(data: bytes) → None

Update buffer from a 3270 data stream using proper DataStreamParser for accurate 3270 order handling.

For backward compatibility with tests expecting simple byte sequences, this method detects whether the data appears to be raw bytes (no 3270 orders) vs. actual 3270 data streams.

Parameters:

data – Raw 3270 data stream bytes.

to_text() → str

Convert screen buffer to Unicode text string.

Returns:

Multi-line string representation.

get_content() → str

Retrieve the buffer content as a string.

get_field_at_position(row: int, col: int) → Field | None

Get the field containing the given position, if any (optimized with caching).

move_cursor_to_first_input_field() → None

Moves the cursor to the beginning of the first unprotected field with performance optimization.

move_cursor_to_next_input_field() → None

Moves the cursor to the beginning of the next unprotected, non-skipped, non-autoskip field. Wraps around to the first field if no next field is found.

In ASCII mode, this also considers ASCII fields detected by VT100 parser.

set_attribute(attr: int, row: int | None = None, col: int | None = None) → None

Set field attribute at specified or current position.

repeat_attribute(attr: int, repeat: int) → None

Repeat attribute a specified number of times.

graphic_ellipsis(count: int) → None

Insert graphic ellipsis characters.

insert_cursor() → None

Insert cursor at current position.

program_tab() → None

Program tab - move to next tab stop.

set_extended_attribute_sfe(attr_type: int, attr_value: int) → None

Accumulate extended field attributes for the field at the current buffer address.

Important semantics: - Extended attributes (from SFE or SA) do NOT consume a display byte and

must NOT create a field-start marker on their own.

• Only the base field attribute (set via SF or SFE type 0xC0) occupies a buffer position and should be recorded in _field_starts.

This method therefore ONLY updates the accumulated extended attributes and does not modify the display buffer or _field_starts.

set_extended_attribute(row: int, col: int, attr_type: str, value: int) → None

Set an extended attribute at a specific position (used by tests and APIs).

This also records a field start at that position to align with field-start-driven detection.

get_field_at(row: int, col: int) → Field | None

Alias for get_field_at_position for compatibility.

get_extended_attributes_at(row: int, col: int) → ExtendedAttributeSet | None

Get extended attributes for a specific position.

Parameters:

• row – Row position

• col – Column position

Returns:

ExtendedAttributeSet if attributes exist, None otherwise

set_field_validation(field: Field, validation_rules: ValidationAttribute) → None

Set validation rules for a field.

Parameters:

• field – The field to set validation for

• validation_rules – Validation attribute with rules

get_field_validation(field: Field) → ValidationAttribute | None

Get validation rules for a field.

Parameters:

field – The field to get validation for

Returns:

ValidationAttribute if set, None otherwise

validate_field_input(field: Field, input_text: str) → Tuple[bool, str | None]

Validate input against field’s validation rules.

Parameters:

• field – The field to validate input for

• input_text – The input text to validate

Returns:

Tuple of (is_valid, error_message)

select_light_pen(row: int, col: int) → int | None

Simulate a light pen selection at the given row and col.

If the field at the given position is light-pen selectable, this method will mark the field as selected, store the selection position, and return the light pen AID. Otherwise, it returns None.

reset_mdt_flags() → None

Reset Modified Data Tag flags for all input fields per RFC 1576.

set_keyboard_lock(locked: bool) → None

Lock or unlock keyboard for user input.

sound_alarm() → None

Trigger terminal alarm/bell.

terminal_reset() → None

Perform general terminal reset.

property keyboard_locked: bool

Check if keyboard is currently locked.

EBCDIC

EBCDIC to ASCII translation utilities for 3270 emulation. Based on IBM Code Page 037.

class pure3270.emulation.ebcdic.EmulationEncoder

Bases: object

Utility class for EBCDIC encoding/decoding in 3270 emulation.

The class exposes encode(text) -> bytes and decode(data) -> str which use IBM Code Page 037 (CP037) for conversions.

static decode(data: bytes) → str

Decode EBCDIC bytes to a printable ASCII string.

Uses CP037 decoding followed by mapping/normalization to ensure the returned string contains only printable ASCII characters (plus newline/carriage/tab). This hides raw EBCDIC/box-graphics from the caller and provides readable approximations for line-drawing chars.

static encode(text: str) → bytes

Encode a Unicode string to EBCDIC CP037 bytes.

pure3270.emulation.ebcdic.translate_ebcdic_to_ascii(data: bytes, codepage: str = 'cp037') → str

Translate raw EBCDIC bytes to ASCII string matching p3270 behavior.

Decode using specified EBCDIC codepage and return raw decoded string like p3270 does, without heavy filtering or normalization.

pure3270.emulation.ebcdic.translate_ascii_to_ebcdic(text: str, codepage: str = 'cp037') → bytes

Translate an ASCII/Unicode string to EBCDIC bytes using specified codepage.

pure3270.emulation.ebcdic.get_p3270_version() → str | None

Get p3270 version for patching.

Returns the installed p3270 package version string or None.

pure3270.emulation.ebcdic.encode_field_attribute(attr: int) → int

Encode 3270 field attribute to EBCDIC.

Current implementation treats attributes as direct values and returns the input unchanged. This is a placeholder for future mapping logic.

class pure3270.emulation.ebcdic.EBCDICCodec(codepage: str = 'cp037', compat: str | None = None)

Bases: object

Backwards-compatible EBCDIC codec used by tests.

This class provides a simple mapping-based encode/decode API that matches historical project tests: methods return tuples (value, length) and unknown characters are encoded as 0x7A and decoded as ‘z’.

The implementation uses configurable EBCDIC codec for deriving a base mapping where possible but intentionally restricts to reverse mapping to a conservative set (uppercase letters, digits and space) so that punctuation and other characters fall back to as ‘unknown’ mapping.

__init__(codepage: str = 'cp037', compat: str | None = None) → None

decode(data: bytes) → tuple[str, int]

Optimized decode with pre-computed character processing.

debug_decode_byte(byte: int) → str

Debug method to see how a single EBCDIC byte is decoded.

encode(text: str | bytes) → Tuple[bytes, int]

Encode text to EBCDIC bytes, returning (bytes, length).

Accepts either str or bytes. If bytes are provided they are interpreted as latin-1 and decoded before encoding so callers that pass raw bytes won’t raise. Prefers full CP037 mapping, falling back to conservative mapping only on failure. Characters not present in the conservative reverse mapping are encoded as 0x7A.

encode_to_unicode_table(text: str) → bytes

Encode without returning length (compat helper used by tests).

pure3270.emulation.ebcdic.detect_codepage_from_trace(trace_file: str) → str

Detect the appropriate EBCDIC codepage from a trace file.

Parameters:

trace_file – Path to the trace file

Returns:

The detected codepage (e.g., ‘cp037’, ‘cp937’, ‘cp933’)

Return type:

str

Protocol Modules

TN3270 Handler

TN3270 protocol handler for pure3270. Handles negotiation, data sending/receiving, and protocol specifics.

class pure3270.protocol.tn3270_handler.HandlerState

Bases: object

TN3270Handler state constants.

DISCONNECTED = 'DISCONNECTED'

CONNECTING = 'CONNECTING'

NEGOTIATING = 'NEGOTIATING'

CONNECTED = 'CONNECTED'

ASCII_MODE = 'ASCII_MODE'

TN3270_MODE = 'TN3270_MODE'

ERROR = 'ERROR'

RECOVERING = 'RECOVERING'

CLOSING = 'CLOSING'

exception pure3270.protocol.tn3270_handler.StateTransitionError

Bases: Exception

Raised when an invalid state transition is attempted.

exception pure3270.protocol.tn3270_handler.StateValidationError

Bases: Exception

Raised when state validation fails.

class pure3270.protocol.tn3270_handler.TN3270Handler(reader: StreamReader | None, writer: StreamWriter | None, screen_buffer: ScreenBuffer | None = None, ssl_context: SSLContext | None = None, host: str = 'localhost', port: int = 23, is_printer_session: bool = False, force_mode: str | None = None, allow_fallback: bool = True, recorder: TraceRecorder | None = None, terminal_type: str = 'IBM-3278-2')

Bases: object

Handler for TN3270 protocol over Telnet. Manages stream I/O, negotiation, and data parsing for 3270 emulation.

__init__(reader: StreamReader | None, writer: StreamWriter | None, screen_buffer: ScreenBuffer | None = None, ssl_context: SSLContext | None = None, host: str = 'localhost', port: int = 23, is_printer_session: bool = False, force_mode: str | None = None, allow_fallback: bool = True, recorder: TraceRecorder | None = None, terminal_type: str = 'IBM-3278-2')

reader: Any | None

writer: Any | None

ssl_context: SSLContext | None

ssl: bool

host: str

port: int

screen_buffer: ScreenBuffer

printer_buffer: PrinterBuffer | None

negotiator: Negotiator

parser: DataStreamParser

recorder: TraceRecorder | None

get_sequence_number_info() → Dict[str, Any]

Get comprehensive sequence number information.

enable_sequence_sync(enable: bool = True) → None

Enable or disable sequence number synchronization.

set_sequence_window(window_size: int) → None

Set the sequence number validation window size with enhanced bounds.

reset_sequence_numbers() → None

Reset sequence numbers to initial state.

get_negotiation_status() → Dict[str, Any]

Get comprehensive negotiation status information.

async negotiate_addressing_mode() → None

Perform addressing mode negotiation during TN3270E session establishment.

This method coordinates with the addressing negotiator to determine the appropriate addressing mode based on client capabilities and server responses.

get_negotiated_addressing_mode() → AddressingMode | None

Get the currently negotiated addressing mode.

Returns:

The negotiated addressing mode, or None if not yet negotiated

async handle_bind_image(bind_image_data: bytes) → None

Handle BIND-IMAGE structured field and update addressing mode negotiation.

Parameters:

bind_image_data – Raw BIND-IMAGE structured field data

async validate_addressing_mode_transition(from_mode: AddressingMode | None, to_mode: AddressingMode) → bool

Validate if an addressing mode transition is allowed.

Parameters:

• from_mode – Current addressing mode

• to_mode – Proposed new addressing mode

Returns:

True if transition is valid, False otherwise

async transition_addressing_mode(new_mode: AddressingMode, reason: str = 'mode transition') → None

Perform a thread-safe addressing mode transition.

Parameters:

• new_mode – The new addressing mode to transition to

• reason – Reason for the transition

Raises:

ValueError – If transition is not allowed or fails

get_addressing_negotiation_summary() → Dict[str, Any]

Get a summary of the addressing mode negotiation process.

Returns:

Dictionary containing negotiation details

get_state_info() → Dict[str, Any]

Get comprehensive state information (thread-safe).

enable_state_validation(enable: bool = True) → None

Enable or disable state validation.

set_max_state_history(max_history: int) → None

Set maximum state history size.

configure_timing_profile(profile: str = 'standard') → None

Configure x3270-compatible timing profile for negotiation.

get_timing_metrics() → Dict[str, Any]

Get timing metrics from the negotiator.

get_current_timing_profile() → str

Get the current timing profile.

enable_timing_monitoring(enable: bool = True) → None

Enable or disable timing monitoring.

enable_step_delays(enable: bool = True) → None

Enable or disable step-by-step delays.

add_state_change_callback(state: str, callback: Callable[[str, str, str], Awaitable[None]]) → None

Add a callback for state changes (expects 3 arguments).

Accepts regular callables, coroutine functions, MagicMocks, and objects with __call__. Signature is validated best-effort; if introspection is unavailable, the callback is accepted.

remove_state_change_callback(state: str, callback: Callable[[str, str, str], Awaitable[None]]) → None

Remove a state change callback (expects 3 arguments).

add_state_entry_callback(state: str, callback: Callable[[str], Awaitable[None]]) → None

Add a callback for when entering a state (expects 1 argument).

Accepts regular callables, coroutine functions, MagicMocks, and objects with __call__. Signature validation is best-effort; non-introspectable callables are accepted.

add_state_exit_callback(state: str, callback: Callable[[str], Awaitable[None]]) → None

Add a callback for when exiting a state (expects 1 argument).

Accepts regular callables, coroutine functions, MagicMocks, and objects with __call__. Signature validation is best-effort; non-introspectable callables are accepted.

wait_for_state(state: str, timeout: float = 30.0) → Event

Get an event that will be set when entering the specified state.

enable_event_signaling(enable: bool = True) → None

Enable or disable event signaling.

get_state_change_event(state: str) → Event

Get the event for a specific state change.

async connect(host: str | None = None, port: int | None = None, ssl_context: SSLContext | None = None) → None

Connect the handler with enhanced state management and x3270 timing.

async negotiate() → None

Perform initial Telnet negotiation.

Delegates to negotiator.

set_ascii_mode() → None

Set the handler to ASCII mode fallback.

Disables EBCDIC processing and sets screen buffer to ASCII mode.

async send_data(data: bytes) → None

Send data over the connection.

Parameters:

data – Bytes to send.

Raises:

ProtocolError – If writer is None or send fails.

async receive_data(timeout: float = 5.0) → bytes

Receive data with timeout.

Parameters:

timeout – Receive timeout in seconds.

Returns:

Received bytes.

Raises:

• asyncio.TimeoutError – If timeout exceeded.

• ProtocolError – If reader is None.

async send_scs_data(scs_data: bytes) → None

Send SCS character data for printer sessions.

Parameters:

scs_data – SCS character data to send

Raises:

ProtocolError – If not connected or not a printer session

async send_printer_status_sf(status_code: int) → None

Send a Printer Status Structured Field to the host.

Parameters:

status_code – The status code to send (e.g., DEVICE_END, INTERVENTION_REQUIRED).

Raises:

ProtocolError – If not connected or writer is None.

async send_sysreq_command(command_code: int) → None

Send a SYSREQ command to the host.

Per RFC 2355 10.5.2: - When in suspended mode, second SYSREQ key press exits suspended mode - SYSREQ with ATTN code while suspended sends LUSTAT and exits suspended mode

Parameters:

command_code – The byte code representing the SYSREQ command.

async send_break() → None

Send a Telnet BREAK command (IAC BRK) to the host.

Raises:

ProtocolError – If not connected or writer is None.

async send_soh_message(status_code: int) → None

Send an SOH (Start of Header) message for printer status to the host.

Parameters:

status_code – The status code to send (e.g., SOH_SUCCESS, SOH_DEVICE_END).

Raises:

ProtocolError – If not connected or writer is None.

async send_print_eoj() → None

Send PRINT-EOJ (End of Job) command for printer sessions.

Raises:

ProtocolError – If not connected or not a printer session

async close() → None

Close the connection with enhanced state management.

is_connected() → bool

Check if the handler is connected.

property negotiated_tn3270e: bool

set_negotiated_tn3270e(value: bool, propagate: bool = True) → None

Set negotiated_tn3270e flag on handler and optionally propagate to negotiator.

Parameters:

• value – New negotiated flag value

• propagate – When True, update negotiator as well (default True). When False, the call is assumed to originate from the negotiator and should not re-propagate.

property lu_name: str | None

Get the LU name.

property screen_rows: int

Get screen rows.

property screen_cols: int

Get screen columns.

property is_printer_session: bool

Get printer session status.

property printer_status: int | None

Get the current printer status.

property sna_session_state: str

Get the current SNA session state.

property connected: bool

Check if handler is connected.

validate_negotiation_completion() → bool

Validate that negotiation is complete before processing screen data.

This method checks if TN3270E negotiation has completed successfully and is ready for screen data processing. Integrates with the negotiator’s validation method and adds additional handler-level checks.

For connected-3270 traces, we are more lenient since the server may not perform full TN3270E negotiation.

Returns:

True if negotiation is complete and ready for screen processing, False otherwise.

validate_negotiation_completion_with_details() → Dict[str, Any]

Get detailed validation information for troubleshooting.

Returns:

Dictionary with validation status and details.

Data Stream

Data stream parser and sender for 3270 protocol.

class pure3270.protocol.data_stream.DataStreamParser(screen_buffer: ScreenBuffer, printer_buffer: PrinterBuffer | None = None, negotiator: Negotiator | None = None, addressing_mode: AddressingMode = AddressingMode.MODE_12_BIT)

Bases: object

parse_light_pen_aid(data: bytes) → None

Minimal parsing support for light-pen AID sequences used by tests. Expected sequence for light-pen AID is:

[LIGHT_PEN_AID, high6bits, low6bits]

where the two following bytes contain 6-bit high/low parts of the 12-bit screen address: address = (high & 0x3F) << 6 | (low & 0x3F) and address maps to row,col using screen.columns.

__init__(screen_buffer: ScreenBuffer, printer_buffer: PrinterBuffer | None = None, negotiator: Negotiator | None = None, addressing_mode: AddressingMode = AddressingMode.MODE_12_BIT) → None

Initialize the DataStreamParser with enhanced validation and buffer protection.

Parameters:

• screen_buffer – ScreenBuffer to update.

• printer_buffer – PrinterBuffer to update for printer sessions.

• negotiator – Negotiator instance for communicating dimension updates.

• addressing_mode – Addressing mode for parsing operations.

screen: ScreenBuffer

printer: PrinterBuffer | None

negotiator: Negotiator | None

addressing_mode: AddressingMode

parser: BaseParser | None

wcc: int | None

aid: int | None

sf_validator: StructuredFieldValidator

ind_file_handler: Any | None

get_validation_errors() → List[str]

Get current validation errors.

clear_validation_errors() → None

Clear validation errors.

get_parser_stats() → Dict[str, Any]

Get parser statistics.

get_aid() → int | None

Get the current AID value.

parse(data: bytes, data_type: int = 0) → Any

Parse 3270 data stream or other data types with enhanced validation and buffer protection.

build_query_reply_sf(query_type: int, data: bytes = b'') → bytes

Build a Query Reply structured field (type byte + optional data).

build_device_type_query_reply() → bytes

Build a basic Device Type Query Reply SF (IBM-3278-2).

build_characteristics_query_reply() → bytes

Build a basic Characteristics Query Reply SF.

class pure3270.protocol.data_stream.DataStreamSender

Bases: object

Data stream sender for building 3270 protocol data streams.

__init__() → None

sf_handlers: Dict[int, Callable[[bytes], None]]

sf_validator: StructuredFieldValidator | None

build_read_modified_all() → bytes

Build a read modified all command.

build_read_modified_fields() → bytes

Build a read modified fields command.

build_key_press(aid: int) → bytes

Build a key press command.

build_write(data: bytes) → bytes

Build a write command.

build_input_stream(modified_fields: List[Tuple[int, bytes]], aid: int, cols: int) → bytes

Build input stream from modified fields.

build_sba(row: int, col: int) → bytes

Build Set Buffer Address command.

build_scs_ctl_codes(code: int) → bytes

Build SCS control codes.

build_data_stream_ctl(code: int) → bytes

Build data stream control.

build_query_sf(query_type: int) → bytes

Build query structured field.

build_printer_status_sf(status_code: int) → bytes

Build printer status structured field for SNA SOH compliance.

This method constructs a Structured Field (SF) for printer status reporting. For SNA SOH integration, the SF can be wrapped in SOH (0x01) when sent in SCS data streams.

Parameters:

status_code – The printer status code (e.g., 0x00 for success, 0x40 for device end, 0x80 for intervention required).

Returns:

Bytes representing the SF for printer status.

Note

Common status codes: - 0x00: Success/Ready - 0x40: Device End - 0x80: Intervention Required - 0x81: Power Off/On - 0x82: Not Ready - 0x83: Intervention Required (specific)

get_structured_field_info() → Dict[str, Any]

Get information about structured field processing capabilities.

validate_current_structured_field() → bool

Validate the current structured field being processed.

get_validation_errors() → List[str]

Get current validation errors.

get_validation_warnings() → List[str]

Get current validation warnings.

clear_validation_state() → None

Clear validation state.

build_soh_message(status_code: int) → bytes

Build SOH (Start of Header) message.

class pure3270.protocol.data_stream.SnaResponse(response_type: int, flags: int | None = None, sense_code: int | None = None, data: object | None = None)

Bases: object

Represents a parsed SNA response.

__init__(response_type: int, flags: int | None = None, sense_code: int | None = None, data: object | None = None)

is_positive() → bool

Check if the response is positive.

is_negative() → bool

Check if the response is negative.

get_sense_code_name() → str

Get a human-readable name for the sense code.

get_response_type_name() → str

Get a human-readable name for the response type.

get_flags_name() → str

Get a human-readable name for the flags.

class pure3270.protocol.data_stream.BindImage(rows: int | None = None, cols: int | None = None, query_reply_ids: List[int] | None = None, model: int | None = None, flags: int | None = None, session_parameters: Dict[str, Any] | None = None)

Bases: object

Represents a parsed BIND-IMAGE Structured Field with enhanced validation.

__init__(rows: int | None = None, cols: int | None = None, query_reply_ids: List[int] | None = None, model: int | None = None, flags: int | None = None, session_parameters: Dict[str, Any] | None = None)

Negotiator

Negotiator for TN3270 protocol specifics. Handles Telnet negotiation and TN3270E subnegotiation.

class pure3270.protocol.negotiator.SnaSessionState(value)

Bases: Enum

Represents the current state of the SNA session.

NORMAL = 'NORMAL'

ERROR = 'ERROR'

PENDING_RECOVERY = 'PENDING_RECOVERY'

SESSION_DOWN = 'SESSION_DOWN'

LU_BUSY = 'LU_BUSY'

INVALID_SEQUENCE = 'INVALID_SEQUENCE'

STATE_ERROR = 'STATE_ERROR'

exception pure3270.protocol.negotiator.SnaStateTransitionError

Bases: Exception

Raised when an invalid SNA state transition is attempted.

class pure3270.protocol.negotiator.Negotiator(writer: Any | None, parser: DataStreamParser | None = None, screen_buffer: ScreenBuffer | None = None, handler: TN3270Handler | None = None, is_printer_session: bool = False, force_mode: str | None = None, allow_fallback: bool = True, recorder: TraceRecorder | None = None, terminal_type: str = 'IBM-3278-2', negotiation_mode: str = 'strict')

Bases: object

__init__(writer: Any | None, parser: DataStreamParser | None = None, screen_buffer: ScreenBuffer | None = None, handler: TN3270Handler | None = None, is_printer_session: bool = False, force_mode: str | None = None, allow_fallback: bool = True, recorder: TraceRecorder | None = None, terminal_type: str = 'IBM-3278-2', negotiation_mode: str = 'strict')

Initialize the Negotiator.

_timing_metrics: Dict[str, Any] def __init__(

writer: StreamWriter for sending commands. parser: DataStreamParser for parsing responses. screen_buffer: ScreenBuffer to update during negotiation. handler: TN3270Handler instance for accessing reader methods. is_printer_session: True if this is a printer session.

property negotiated_tn3270e: bool

Backwards-compatible negotiated flag property.

set_negotiated_tn3270e(value: bool) → None

Public API: set negotiated_tn3270e flag and propagate state changes.

validate_negotiation_completion() → bool

Validate that the negotiation has completed sufficiently to process TN3270/TN3270E screen data.

This is a conservative check used by the TN3270 handler to decide whether it is safe to treat incoming data as 3270 data rather than early ASCII/VT100 streams. It must be synchronous to integrate with the handler’s fast-path calls.

Returns:

True if negotiation is complete, False otherwise.

validate_negotiation_completion_with_details() → Dict[str, Any]

Return a dictionary with negotiation validation details for diagnostics.

update_printer_status(status_code: int) → None

Update cached printer status and signal any waiters.

This mirrors legacy negotiator behavior that exposed a simple status update hook used by printer session paths.

infer_tn3270e_from_trace(trace: bytes) → bool

Infer TN3270E negotiation success from raw Telnet negotiation bytes.

This mirrors the temporary heuristic previously implemented in the handler. We keep it here so that test fixtures can rely on a single canonical implementation and the handler stays slim.

Rules:

1. If IAC WONT TN3270E (FF FC 24) or IAC DONT TN3270E (FF FE 24) appears => failure (False).

2. Else if IAC WILL EOR (FF FB 19) appears => success (True).

3. Otherwise => False.

The heuristic is intentionally conservative; explicit refusal always wins over implied success.

async negotiate() → None

Perform initial Telnet negotiation with x3270-compatible timing.

Sends WILL TERMINAL-TYPE and DO TERMINAL-TYPE during initial negotiation, then waits for responses with precise timing that matches x3270’s negotiation patterns.

Raises:

NegotiationError – If negotiation fails after all retries.

set_ascii_mode() → None

Set to ASCII mode fallback, matching s3270 behavior.

Disables EBCDIC processing and enables ASCII/VT100 terminal emulation.

async handle_iac_command(command: int, option: int) → None

Handle Telnet IAC (Interpret As Command) commands.

Parameters:

• command – The IAC command (DO, DONT, WILL, WONT).

• option – The Telnet option number.

is_printer_session_active() → bool

Check if this is a printer session.

Returns:

True if printer session.

Return type:

bool

property lu_name: str | None

Get the LU name.

property is_data_stream_ctl_active: bool

Check if DATA-STREAM-CTL function is active.

property is_bind_image_active: bool

Return True when BIND-IMAGE function bit is active, or override for tests.

async handle_subnegotiation(option: int, sub_payload: bytes) → None

Handle Telnet subnegotiation for non-TN3270E options.

Parameters:

• option – The Telnet option number.

• sub_payload – The subnegotiation payload.

property current_sna_session_state: SnaSessionState

Get the current SNA session state.

handle_bind_image(bind_image: BindImage) → None

Handle BIND-IMAGE structured field.

Parameters:

bind_image – The BIND-IMAGE to handle.

async negotiate_addressing_mode() → None

Perform addressing mode negotiation during TN3270E session establishment.

This method coordinates with the addressing negotiator to determine the appropriate addressing mode based on client capabilities and server responses.

get_negotiated_addressing_mode() → AddressingMode | None

Get the currently negotiated addressing mode.

Returns:

The negotiated addressing mode, or None if not yet negotiated

async handle_bind_image_addressing(bind_image_data: bytes) → None

Handle BIND-IMAGE structured field and update addressing mode negotiation.

Parameters:

bind_image_data – Raw BIND-IMAGE structured field data

async validate_addressing_mode_transition(from_mode: AddressingMode | None, to_mode: AddressingMode) → bool

Validate if an addressing mode transition is allowed.

Parameters:

• from_mode – Current addressing mode (None if not set)

• to_mode – Proposed new addressing mode

Returns:

True if transition is valid, False otherwise

async transition_addressing_mode(new_mode: AddressingMode, reason: str = 'mode transition') → None

Perform a thread-safe addressing mode transition.

Parameters:

• new_mode – The new addressing mode to transition to

• reason – Reason for the transition

Raises:

ValueError – If transition is not allowed or fails

get_addressing_negotiation_summary() → Dict[str, str]

Get a summary of the addressing mode negotiation process.

Returns:

Dictionary containing negotiation details

SSL Wrapper

SSL/TLS wrapper for secure TN3270 connections using stdlib ssl module.

exception pure3270.protocol.ssl_wrapper.SSLError

Bases: Exception

Error during SSL operations.

class pure3270.protocol.ssl_wrapper.SSLWrapper(verify: bool = True, cafile: str | None = None, capath: str | None = None)

Bases: object

Layers SSL/TLS on top of asyncio connections using Python’s ssl module.

__init__(verify: bool = True, cafile: str | None = None, capath: str | None = None)

Initialize the SSLWrapper.

Parameters:

• verify – Whether to verify the server’s certificate.

• cafile – Path to CA certificate file.

• capath – Path to CA certificates directory.

create_context() → SSLContext

Create an SSLContext for secure connections.

This implementation matches the expectations of our tests: - Build context with ssl.SSLContext(ssl.PROTOCOL_TLS_CLIENT) - When verify=True: check_hostname=True and verify_mode=CERT_REQUIRED - Enforce minimum TLS 1.2 - Configure cipher suite to “HIGH:!aNULL:!MD5”

Returns:

Configured SSLContext.

Raises:

SSLError – If context creation fails.

wrap_connection(telnet_connection: Any) → Any

Wrap an existing telnet connection with SSL with x3270-compatible fallback mechanisms.

Parameters:

telnet_connection – The telnet connection object (e.g., from asyncio.open_connection).

Returns:

Wrapped connection.

Note: This is a stub; asyncio.open_connection handles SSL natively via ssl parameter.

create_fallback_context() → SSLContext

Create a fallback SSL context for legacy server compatibility.

Returns:

Fallback SSLContext with relaxed security for compatibility.

Raises:

SSLError – If context creation fails.

test_ssl_compatibility(hostname: str, port: int) → Dict[str, Any]

Test SSL compatibility with a target server.

Parameters:

• hostname – Target hostname

• port – Target port

Returns:

Compatibility test results

get_context() → SSLContext

Get the SSLContext (create if not exists).

decrypt(encrypted_data: bytes) → bytes

Stub for decrypting data (for testing).

Printer Sessions

High-level printer session classes for TN3270E printer LU support.

Session management for pure3270, handling synchronous and asynchronous 3270 connections.

exception pure3270.session.SessionError(message: str, context: Dict[str, Any] | None = None)

Bases: Exception

Base exception for session-related errors.

When context is provided, include key details in the string representation so tests can assert on them without reaching into attributes.

__init__(message: str, context: Dict[str, Any] | None = None)

exception pure3270.session.ConnectionError(message: str, context: Dict[str, Any] | None = None)

Bases: SessionError

Raised when connection fails.

class pure3270.session.Session(host: str | None = None, port: int = 23, ssl_context: Any | None = None, force_mode: str | None = None, allow_fallback: bool = True, enable_trace: bool = False, terminal_type: str = 'IBM-3278-2')

Bases: object

Synchronous wrapper for AsyncSession.

This class provides a synchronous interface to the asynchronous 3270 session. All methods use asyncio.run() to execute async operations.

__init__(host: str | None = None, port: int = 23, ssl_context: Any | None = None, force_mode: str | None = None, allow_fallback: bool = True, enable_trace: bool = False, terminal_type: str = 'IBM-3278-2') → None

Initialize a synchronous session with a dedicated thread and event loop.

connect(host: str | None = None, port: int | None = None, ssl_context: Any | None = None) → None

Connect to the 3270 host synchronously.

Parameters:

• host – Optional host override.

• port – Optional port override (default 23).

• ssl_context – Optional SSL context override.

Raises:

ConnectionError – If connection fails.

send(data: bytes) → None

Send data to the session.

Parameters:

data – Bytes to send.

Raises:

SessionError – If send fails.

read(timeout: float = 5.0) → bytes

Read data from the session.

Parameters:

timeout – Read timeout in seconds.

Returns:

Received bytes.

Raises:

SessionError – If read fails.

get_aid() → int | None

Get AID synchronously (last known AID value).

close() → None

Close the session synchronously.

property connected: bool

Check if session is connected.

get_trace_events() → List[Any]

open(host: str, port: int = 23) → None

Open connection synchronously (s3270 Open() action).

property screen_buffer: ScreenBuffer

Expose screen buffer for sync Session tests.

property input_inhibited: bool

Get Input Inhibited state (keyboard locked).

property tn3270e_mode: bool

Check if session is in TN3270E mode.

Preference order: 1) Async session’s tn3270 mode flag when True 2) Handler’s negotiated_tn3270e flag as a fallback

close_script() → None

Close script synchronously (s3270 CloseScript() action).

ascii(data: bytes) → str

Convert EBCDIC data to ASCII text (s3270 Ascii() action).

Parameters:

data – EBCDIC bytes to convert.

Returns:

ASCII string representation.

ebcdic(text: str) → bytes

Convert ASCII text to EBCDIC data (s3270 Ebcdic() action).

Parameters:

text – ASCII text to convert.

Returns:

EBCDIC bytes representation.

ascii1(byte_val: int) → str

Convert a single EBCDIC byte to ASCII character.

This is a pure conversion utility and does not require a live session.

ebcdic1(char: str) → int

Convert a single ASCII character to EBCDIC byte.

This is a pure conversion utility and does not require a live session.

ascii_field(field_index: int) → str

Convert field content to ASCII text (s3270 AsciiField() action).

Parameters:

field_index – Index of field to convert.

Returns:

ASCII string representation of field content.

cursor_select() → None

Select field at cursor (s3270 CursorSelect() action).

Raises:

SessionError – If not connected.

delete_field() → None

Delete field at cursor (s3270 DeleteField() action).

Raises:

SessionError – If not connected.

string(text: str) → None

Send string to the session (s3270 String() action).

circum_not() → None

Toggle circumvention of field protection (s3270 CircumNot() action).

Raises:

SessionError – If not connected.

script(commands: str) → None

Execute script (s3270 Script() action).

Parameters:

commands – Script commands.

Raises:

SessionError – If not connected.

execute(command: str) → str

Execute external command synchronously (s3270 Execute() action).

info() → str

Get session information synchronously.

query(query_type: str = 'All') → str

Query session information synchronously.

Supported query types (s3270 protocol parity):

• All: Summary information

• CodePage: Host code page

• ConnectionState: Connection state

• Cursor: Cursor position

• Model: 3270 model information

• ScreenCurSize: Current screen dimensions

• Tn3270eOptions: Active TN3270E functions

• TelnetHostOptions: Host TELNET options

• TelnetMyOptions: Client TELNET options

• And 30+ more query types…

move_cursor1(row: int, col: int) → None

Move cursor to specified position (s3270 MoveCursor1 action).

hex_string(hex_data: str) → None

Send raw hex bytes as input (s3270 HexString() action).

set(option: str, value: str) → None

Set option synchronously.

print_text(text: str) → None

Print text synchronously.

snap() → None

Take snapshot synchronously.

show() → None

Show screen synchronously.

trace(on: bool = True) → None

Enable/disable tracing synchronously.

wait(seconds: float = 1.0) → None

Wait synchronously.

sleep(seconds: float = 1.0) → None

Sleep synchronously.

transfer(file: str) → None

Transfer file synchronously.

source(file: str) → None

Source file synchronously.

expect(pattern: str, timeout: float = 10.0) → bool

Expect pattern synchronously.

fail(message: str) → None

Fail with message synchronously.

compose(text: str) → None

Compose text synchronously.

set_field_attribute(field_index: int, attr: str, value: int) → None

Set a screen field attribute via the underlying async session.

This is a synchronous passthrough used by compatibility tests; the underlying implementation is synchronous as well.

cookie(cookie_string: str) → None

Set cookie synchronously.

interrupt() → None

Send interrupt synchronously (s3270 Interrupt() action).

key(keyname: str) → None

Send key synchronously (s3270 Key() action).

submit(aid: int) → None

Submit with AID synchronously.

home() → None

Move cursor to home position.

up() → None

Move cursor up.

down() → None

Move cursor down.

tab() → None

Move cursor to next tab stop.

backtab() → None

Move cursor to previous tab stop.

backspace() → None

Send backspace key.

enter() → None

Send Enter key.

pf(n: str) → None

Send PF key.

pa(n: str) → None

Send PA key.

erase() → None

Erase entire screen.

clear() → None

Clear entire screen (alias for erase).

newline() → None

Move cursor to start of next line.

erase_eof() → None

Erase to end of field.

erase_input() → None

Erase all input fields.

field_end() → None

Move cursor to end of field.

field_mark() → None

Set field mark.

dup() → None

Duplicate character.

pause(seconds: float = 1.0) → None

Pause session.

bell() → None

Ring bell.

left2() → None

Move cursor left by 2.

right() → None

Move cursor right.

right2() → None

Move cursor right by 2.

reset() → None

Reset session.

field_exit() → None

Exit field.

sysreq() → None

Send SysReq key.

attn() → None

Send Attention key.

test() → None

Send Test key.

left() → None

Move cursor left.

class pure3270.session.AsyncSession(host: str | None = None, port: int = 23, ssl_context: Any | None = None, force_mode: str | None = None, allow_fallback: bool = True, enable_trace: bool = False, terminal_type: str = 'IBM-3278-2', is_printer_session: bool = False)

Bases: object

Asynchronous 3270 session handler.

This class provides an asynchronous interface to the 3270 session. All operations are async and use asyncio for non-blocking I/O.

AID_MAP: Dict[str, int] = {'Attn': 241, 'BackSpace': 248, 'CLEAR': 109, 'Dup': 245, 'Enter': 125, 'PA(1)': 108, 'PA(2)': 110, 'PA(3)': 107, 'PF(1)': 241, 'PF(10)': 122, 'PF(11)': 123, 'PF(12)': 124, 'PF(13)': 193, 'PF(14)': 194, 'PF(15)': 195, 'PF(16)': 196, 'PF(17)': 197, 'PF(18)': 198, 'PF(19)': 199, 'PF(2)': 242, 'PF(20)': 200, 'PF(21)': 201, 'PF(22)': 74, 'PF(23)': 75, 'PF(24)': 76, 'PF(3)': 243, 'PF(4)': 244, 'PF(5)': 245, 'PF(6)': 246, 'PF(7)': 247, 'PF(8)': 248, 'PF(9)': 249, 'RESET': 106, 'SysReq': 240, 'TEST': 17, 'Test': 17}

__init__(host: str | None = None, port: int = 23, ssl_context: Any | None = None, force_mode: str | None = None, allow_fallback: bool = True, enable_trace: bool = False, terminal_type: str = 'IBM-3278-2', is_printer_session: bool = False) → None

Initialize the AsyncSession.

Parameters:

• host – Host to connect to.

• port – Port to connect to.

• ssl_context – SSL context for secure connections.

• force_mode – Force specific TN3270 mode.

• allow_fallback – Allow fallback to TN3270 if TN3270E fails.

• enable_trace – Enable tracing.

• terminal_type – Terminal model type.

• is_printer_session – True if this is a printer session.

resources: Dict[str, str]

property host: str | None

Public host accessor for tests.

property port: int

Public port accessor for tests.

property model: str

property screen_buffer: ScreenBuffer

Get the screen buffer.

property screen: ScreenBuffer

Get the screen buffer (alias for screen_buffer).

property input_inhibited: bool

Get Input Inhibited state (keyboard locked).

property handler: TN3270Handler | None

Get the underlying TN3270 handler (may be None if not connected).

property tn3270e_mode: bool

Check if session is in TN3270E mode.

property connected: bool

Check if session is connected.

async connect(host: str | None = None, port: int | None = None, ssl_context: Any | None = None) → None

Connect to the 3270 host asynchronously.

async send(data: bytes) → None

Send data to the session with retry logic.

async send_data(data: bytes) → None

Send data to the session (alias for send).

async read(timeout: float = 5.0) → bytes

Read data from the session with retry logic.

async receive_data(timeout: float = 5.0) → bytes

Receive data from the session (alias for read).

get_aid() → int | None

Get the last AID value.

async close() → None

Close the session.

async close_script() → None

Close script (s3270 CloseScript() action).

get_trace_events() → List[Any]

Get trace events.

managed() → AsyncIterator[AsyncSession]

Provide a managed async context that guarantees close() on exit.

ascii(data: bytes) → str

Convert EBCDIC to ASCII.

ebcdic(text: str) → bytes

Convert ASCII to EBCDIC.

ascii1(byte_val: int) → str

Convert single EBCDIC byte to ASCII.

ebcdic1(char: str) → int

Convert single ASCII char to EBCDIC.

ascii_field(field_index: int) → str

Convert field to ASCII.

async cursor_select() → None

Select field at cursor.

async delete_field() → None

Delete field at cursor.

async insert_text(text: str) → None

Insert text at cursor.

async string(text: str) → None

Send string to the session.

async circum_not() → None

Toggle circumvention.

async script(commands: str) → None

Execute script command(s).

Supports both simple method names and function-style commands like: - Simple: “connect”, “disconnect” - With args: “String(test)”, “MoveCursor(5,10)”

async execute(command: str) → str

Execute external command.

async interrupt() → None

Send interrupt.

async key(keyname: str) → None

Send key synchronously (s3270 Key() action).

capabilities() → str

Get capabilities.

async set_option(option: str, value: str) → None

Set option (placeholder for future functionality).

async query(query_type: str = 'All') → str

Query screen.

async set(option: str, value: str) → None

Set option (alias).

async exit() → None

Exit session.

async bell() → None

Ring bell.

async pause(seconds: float = 1.0) → None

Pause.

async ansi_text(data: bytes) → str

Convert EBCDIC to ANSI.

async hex_string(hex_str: str) → bytes

Convert hex to bytes.

async show() → None

Show screen.

async snap() → None

Save snapshot.

async newline() → None

Move cursor to start of next line.

async page_down() → None

Page down: wrap to top row for tests.

async page_up() → None

Page up: move to top row for tests.

async paste_string(text: str) → None

async end() → None

Move cursor to end of current line.

async move_cursor(row: int, col: int) → None

async move_cursor1(row1: int, col1: int) → None

async next_word() → None

async previous_word() → None

async toggle_insert() → None

async flip() → None

async insert() → None

async delete() → None

Delete character at cursor by shifting remainder left and clearing last.

async disconnect() → None

async left() → None

Move cursor left.

async right() → None

Move cursor right.

async left2() → None

Move cursor left by 2.

async right2() → None

Move cursor right by 2.

async mono_case() → None

Toggle monocase.

async nvt_text(text: str) → None

Send NVT text.

async print_text(text: str) → None

Print text.

async prompt(message: str) → str

Prompt for input.

async read_buffer() → bytes

Read buffer.

async reconnect() → None

Reconnect.

async info() → str

Get session information.

async quit() → None

Quit the session.

async home() → None

Move cursor to home position.

async up() → None

Move cursor up.

async down() → None

Move cursor down.

async tab() → None

Send Tab key.

async backtab() → None

Move cursor to previous tab stop.

async backspace() → None

Send backspace key.

async enter() → None

Send Enter key.

async clear() → None

Clear screen locally.

async pf(n: str) → None

Send PF key.

async pa(n: str) → None

Send PA key.

async macro(commands: List[str]) → None

Execute a sequence of commands.

async erase() → None

Erase entire screen (local).

async erase_eof() → None

Erase to end of field.

async erase_input() → None

Erase all input fields.

async field_end() → None

Move cursor to end of field.

async field_mark() → None

Set field mark.

async dup() → None

Duplicate character.

async field_exit() → None

Exit field.

async sysreq() → None

Send SysReq key.

async attn() → None

Send Attention key.

async test() → None

Send Test key.

async screen_trace() → None

Screen trace.

async source(file: str) → None

Source script.

async subject_names() → None

Subject names.

async sys_req(command: str | None = None) → None

Send SysReq.

async send_break() → None

Send Telnet BREAK.

async send_soh_message(status_code: int) → None

Send SOH message.

async toggle_option(option: str) → None

Toggle option.

async trace(on: bool) → None

Enable/disable tracing.

async transfer(file: str) → None

Transfer file.

async send_file(local_path: str, remote_name: str) → None

Send a file to the host using IND$FILE protocol.

async receive_file(remote_name: str, local_path: str) → None

Receive a file from the host using IND$FILE protocol.

async wait_condition(condition: str) → None

Wait for condition.

async compose(text: str) → None

Compose text.

async cookie(cookie_string: str) → None

Set a cookie.

This sets the cookie immediately and returns None. Tests can call this without awaiting since the side-effect happens immediately.

async expect(pattern: str, timeout: float = 10.0) → bool

Wait for pattern.

async fail(message: str) → None

Fail with message.

async select_light_pen(row: int, col: int) → None

Select light pen.

async start_lu_lu_session(lu_name: str) → None

Start LU-LU session.

property lu_lu_session: Any

Get the current LU-LU session if active.

async load_resource_definitions(file_path: str) → None

Load resources.

async apply_resources() → None

Apply resources.

set_field_attribute(field_index: int, attr: str, value: int) → None

Set field attribute.

async submit(aid: int) → None

Submit with AID.

Printer Protocol

Low-level printer protocol implementation.

Advanced printer session support for TN3270E protocol with comprehensive SCS control codes.

class pure3270.protocol.printer.PrinterJob(job_id: str = '', max_data_size: int = 1048576)

Bases: object

Represents a printer job in a TN3270E printer session with advanced features.

__init__(job_id: str = '', max_data_size: int = 1048576)

Initialize a printer job with thread-safe operations.

add_data(data: bytes) → None

Add SCS character data to the job with thread safety.

add_scs_control_code(scs_code: int) → None

Add SCS control code to the job.

increment_line_count() → None

Increment line count for the job.

increment_page_count() → None

Increment page count for the job.

complete_job() → None

Mark the job as completed with thread safety.

set_error(error_msg: str) → None

Mark the job as having an error with thread safety.

pause_job() → None

Pause the job with thread safety.

resume_job() → None

Resume the job with thread safety.

get_duration() → float

Get the job duration in seconds.

get_page_count() → int

Get the number of pages in the job.

get_data_size() → int

Get the size of the job data in bytes.

get_scs_control_codes() → List[int]

Get the list of SCS control codes.

get_error_message() → str | None

Get the error message if any.

is_thread_safe() → bool

Check if the job supports thread-safe operations.

class pure3270.protocol.printer.PrinterSession

Bases: object

Advanced TN3270E printer session handler with comprehensive SCS support.

__init__() → None

Initialize the printer session with thread safety.

activate() → None

Activate the printer session with thread safety.

deactivate() → None

Deactivate the printer session with thread safety.

start_new_job(job_id: str = '') → PrinterJob

Start a new printer job.

add_scs_data(data: bytes, holding_lock: bool = False) → None

Add SCS character data to the current job.

Parameters:

• data – SCS data to add

• holding_lock – True if caller already holds self.lock (internal use only)

handle_print_eoj() → None

Handle PRINT-EOJ (End of Job) command.

prune() → None

Prune completed jobs to the last max_jobs.

get_current_job() → PrinterJob | None

Get the current active job.

get_completed_jobs() → List[PrinterJob]

Get the list of completed jobs.

get_job_statistics() → Dict[str, Any]

Get printer job statistics.

clear_completed_jobs() → None

Clear the list of completed jobs.

close() → None

Close the printer session and clear all jobs.

handle_scs_control_code(scs_code: int) → None

Handle SCS control codes with comprehensive support.

process_tn3270e_message(header: TN3270EHeader, data: bytes) → None

Process a TN3270E message for printer session with comprehensive support.

get_session_info() → Dict[str, Any]

Get comprehensive session information.

set_device_type(device_type: str) → None

Set the device type for the printer session.

add_tn3270e_function(function: int) → None

Add a supported TN3270E function.

supports_tn3270e_function(function: int) → bool

Check if a TN3270E function is supported.

get_error_rate() → float

Get the current error rate.

is_healthy() → bool

Check if the session is healthy.

TN3270E Header

TN3270E message header structure for pure3270.

class pure3270.protocol.tn3270e_header.TN3270EHeader(data_type: int = 0, request_flag: int = 0, response_flag: int = 0, seq_number: int = 0)

Bases: object

TN3270E Message Header Structure (RFC 2355 Section 3).

The TN3270E header is a 5-byte structure with the following fields:

Byte 0: DATA-TYPE Byte 1: REQUEST-FLAG Byte 2: RESPONSE-FLAG Byte 3-4: SEQ-NUMBER (16-bit big-endian)

Extended for addressing mode negotiation: - REQUEST-FLAG bit 7: Extended addressing capability (14-bit support) - RESPONSE-FLAG bit 7: Extended addressing negotiated

REQUEST_EXTENDED_ADDRESSING = 128

RESPONSE_EXTENDED_ADDRESSING = 128

__init__(data_type: int = 0, request_flag: int = 0, response_flag: int = 0, seq_number: int = 0)

Initialize a TN3270E header.

Parameters:

• data_type – Type of data (TN3270_DATA, SCS_DATA, etc.)

• request_flag – Request flags

• response_flag – Response flags

• seq_number – Sequence number (0-65535)

classmethod from_bytes(data: bytes) → TN3270EHeader | None

Parse a TN3270E header from bytes with enhanced malformed data handling.

Parameters:

data – 5 bytes containing the TN3270E header

Returns:

TN3270EHeader instance or None if parsing fails

to_bytes() → bytes

Convert the header to bytes.

Returns:

5 bytes representing the TN3270E header

is_tn3270_data() → bool

Check if this is TN3270_DATA type.

is_scs_data() → bool

Check if this is SCS_DATA type.

is_response() → bool

Check if this is RESPONSE type.

is_no_response() → bool

Check if this is a no-response flag (0x00).

Per RFC 2355, NO-RESPONSE (0x00) means the sender does not expect the receiver to respond either positively or negatively.

is_error_response() → bool

Check if this is an error response.

is_negative_response() → bool

Check if this is a negative response (error response per RFC 2355).

Note: RFC 2355 doesn’t have a separate ‘negative response’ flag. ERROR-RESPONSE (0x01) means ‘respond only if error’. Negative responses are indicated by ERROR-RESPONSE with sense data.

is_always_response() → bool

Check if this is an always response.

has_extended_addressing_request() → bool

Check if this header requests extended addressing capability.

has_extended_addressing_response() → bool

Check if this header indicates extended addressing was negotiated.

set_extended_addressing_request(enabled: bool = True) → None

Set or clear the extended addressing request flag.

set_extended_addressing_response(enabled: bool = True) → None

Set or clear the extended addressing response flag.

get_data_type_name() → str

Get human-readable name for data type.

get_response_flag_name() → str

Get human-readable name for response flag.

handle_negative_response(data: bytes) → None

Handle negative response by parsing the code and raising ProtocolError with details.

Parameters:

data – Bytes following the header, starting with the negative code byte.

Raises:

ProtocolError – With details of the negative response code and sense if applicable.

class pure3270.protocol.tn3270e_header.MalformedDataHandler

Bases: object

__init__() → None

malformed_data_log: list[dict[str, object]]

Handles malformed TN3270E data and provides recovery mechanisms.

handle_malformed_header(data: bytes, context: str = '') → TN3270EHeader | None

Attempt to recover from malformed TN3270E headers.

Parameters:

• data – Raw bytes that may contain a malformed header

• context – Context information for logging

Returns:

TN3270EHeader instance if recovery successful, None otherwise

get_malformed_data_stats() → Dict[str, Any]

Get statistics about malformed data encountered.

clear_malformed_data_log() → None

Clear the malformed data log.

pure3270.protocol.tn3270e_header.test_malformed_data_handling() → None

Test the malformed data handling functionality.

Utils

pure3270.protocol.utils.TN3270E_QUERY = 15

Utility functions for TN3270/TN3270E protocol handling.

Typing notes: - Writer parameters are annotated as Optional[asyncio.StreamWriter] to reflect possibility of absent writer during teardown. - _schedule_if_awaitable centralizes best-effort handling of AsyncMock.write returning an awaitable to avoid repeated inline inspection logic.

class pure3270.protocol.utils.TerminalCapabilities(screen_size: Tuple[int, int], color_support: bool, extended_attributes: bool, programmed_symbols: bool, extended_highlighting: bool, light_pen_support: bool, magnetic_slot_reader: bool, operator_information_area: bool, max_alternate_screen_size: Tuple[int, int] | None = None, character_sets: List[str] | None = None)

Bases: object

Defines the capabilities and characteristics of a 3270 terminal model.

Based on IBM 3270 Data Stream Programmer’s Reference and GA23-0059 documentation.

screen_size: Tuple[int, int]

color_support: bool

extended_attributes: bool

programmed_symbols: bool

extended_highlighting: bool

light_pen_support: bool

magnetic_slot_reader: bool

operator_information_area: bool

max_alternate_screen_size: Tuple[int, int] | None = None

character_sets: List[str] | None = None

__init__(screen_size: Tuple[int, int], color_support: bool, extended_attributes: bool, programmed_symbols: bool, extended_highlighting: bool, light_pen_support: bool, magnetic_slot_reader: bool, operator_information_area: bool, max_alternate_screen_size: Tuple[int, int] | None = None, character_sets: List[str] | None = None) → None

pure3270.protocol.utils.get_supported_terminal_models() → List[str]

Get list of all supported terminal model names.

pure3270.protocol.utils.is_valid_terminal_model(model: str) → bool

Check if a terminal model is supported.

pure3270.protocol.utils.get_terminal_capabilities(model: str) → TerminalCapabilities | None

Get capabilities for a specific terminal model.

pure3270.protocol.utils.get_screen_size(model: str) → Tuple[int, int]

Get screen dimensions for a terminal model.

pure3270.protocol.utils.send_iac(writer: StreamWriter | None, command: bytes) → None

Send an IAC command to the writer.

Ensures the IAC (0xFF) prefix is present exactly once. If the provided command already begins with IAC, it will not be duplicated.

pure3270.protocol.utils.send_subnegotiation(writer: StreamWriter | None, opt: bytes, data: bytes) → None

Send subnegotiation.

Parameters:

• writer – StreamWriter.

• opt – Option byte.

• data – Subnegotiation data.

pure3270.protocol.utils.strip_telnet_iac(data: bytes, handle_eor_ga: bool = False, enable_logging: bool = False) → bytes

Strip Telnet IAC sequences from data.

Parameters:

• data – Raw bytes containing potential IAC sequences.

• handle_eor_ga – If True, specifically handle EOR (0x19) and GA (0xf9) commands.

• enable_logging – If True, log EOR/GA stripping.

Returns:

Cleaned bytes without IAC sequences.

class pure3270.protocol.utils.BaseStringParser(text: str)

Bases: object

__init__(text: str)

has_more() → bool

remaining() → int

peek_char() → str

read_char() → str

advance(n: int = 1) → None

VT100 Parser

VT100 terminal emulator for pure3270 ASCII mode support.

This module provides VT100 escape sequence parsing to support ASCII terminal emulation as a fallback when TN3270 negotiation fails, matching s3270 behavior.

class pure3270.protocol.vt100_parser.VT100ParserState

Bases: object

Represents the recoverable state of a VT100 parser.

__init__() → None

save_from_parser(parser: VT100Parser) → None

Save current state from parser instance.

restore_to_parser(parser: VT100Parser) → None

Restore state to parser instance.

class pure3270.protocol.vt100_parser.VT100Parser(screen_buffer: ScreenBuffer)

Bases: object

VT100 escape sequence parser for ASCII terminal emulation.

__init__(screen_buffer: ScreenBuffer) → None

Initialize VT100 parser.

Parameters:

screen_buffer – ScreenBuffer instance to update with parsed content

enable_error_recovery() → None

Enable error recovery mechanism.

disable_error_recovery() → None

Disable error recovery mechanism for debugging.

is_error_recovery_enabled() → bool

Check if error recovery is enabled.

parse(data: bytes) → None

Parse VT100 escape sequences and update screen buffer.

Parameters:

data – Raw ASCII data with VT100 escape sequences

Emulation Modules

Screen Buffer

Screen buffer management for 3270 emulation.

class pure3270.emulation.screen_buffer.Field(start: Tuple[int, int], end: Tuple[int, int], protected: bool = False, numeric: bool = False, modified: bool = False, selected: bool = False, intensity: int = 0, color: int = 0, background: int = 0, validation: int = 0, outlining: int = 0, character_set: int = 0, sfe_highlight: int = 0, light_pen: int = 0, content: bytes | None = None, codepage: str = 'cp037')

Bases: object

A simple Field object that mirrors the previous namedtuple API with convenient defaults and helper methods used by tests.

It intentionally implements get_content, set_content, and _replace to be compatible with test expectations. content is stored as raw EBCDIC bytes; get_content/set_content use the EBCDICCodec by default but will fall back to EmulationEncoder when the codec isn’t present.

__init__(start: Tuple[int, int], end: Tuple[int, int], protected: bool = False, numeric: bool = False, modified: bool = False, selected: bool = False, intensity: int = 0, color: int = 0, background: int = 0, validation: int = 0, outlining: int = 0, character_set: int = 0, sfe_highlight: int = 0, light_pen: int = 0, content: bytes | None = None, codepage: str = 'cp037') → None

get_content(codec: EBCDICCodec | None = None) → str

Return decoded content as a Unicode string.

Accepts an optional codec for decoding; if not provided we use EBCDICCodec() when available, otherwise EmulationEncoder.

set_content(text: str, codec: EBCDICCodec | None = None) → None

Set the field content from a Unicode string and mark modified.

Uses optional codec for encoding; falls back to EmulationEncoder.

class pure3270.emulation.screen_buffer.ScreenBuffer(rows: int = 24, cols: int = 80, init_value: int = 64, codepage: str = 'cp037')

Bases: BufferWriter

property ascii_buffer: str

Return the entire screen buffer as a multi-line ASCII string.

Tests expect exactly rows lines separated by newlines, regardless of trailing blank content. This method converts each row of the EBCDIC buffer to printable ASCII, masking non-printables to spaces, and joins all rows with newlines without trimming.

In ASCII mode, returns the buffer content directly as ASCII text.

__init__(rows: int = 24, cols: int = 80, init_value: int = 64, codepage: str = 'cp037')

Initialize the ScreenBuffer.

Parameters:

• rows – Number of rows (default 24).

• cols – Number of columns (default 80).

• init_value – Initial value for buffer (default 0x40 for space).

set_ascii_mode(ascii_mode: bool) → None

Set ASCII mode for the screen buffer.

In ASCII mode, characters are stored as ASCII bytes directly. In EBCDIC mode (default), characters are converted to EBCDIC.

Parameters:

ascii_mode – True for ASCII mode, False for EBCDIC mode

get_ascii_mode() → bool

Get current ASCII mode setting.

property input_inhibited: bool

Get Input Inhibited state (keyboard locked).

get_field_content(field_index: int) → str

Return the content of the field at the given index as a decoded string.

read_modified_fields() → list[tuple[tuple[int, int], str]]

Return a list of tuples (position, content) for modified fields.

begin_bulk_update() → None

Suspend field detection until end_bulk_update is called.

end_bulk_update() → None

Resume field detection and run a single detection pass.

clear() → None

Clear the screen buffer and reset fields.

is_ascii_mode() → bool

Return True if screen buffer is in ASCII mode.

set_position(row: int, col: int, wrap: bool = False, strict: bool = False) → None

Set cursor position with bounds checking or wrapping.

Parameters:

• row – Target row position

• col – Target column position

• wrap – If True, allow wrapping to next line; if False, check bounds

• strict – If True, raise IndexError on out of bounds; if False, clamp values

Raises:

IndexError – When strict=True and position is out of bounds

write_char(ebcdic_byte: int, row: int | None = None, col: int | None = None, protected: bool = False, circumvent_protection: bool = False, mark_field_start: bool = False) → None

Write an EBCDIC character to the buffer at position.

Parameters:

• ebcdic_byte – EBCDIC byte value.

• row – Row position.

• col – Column position.

• protected – Protection attribute to set.

• circumvent_protection – If True, write even to protected fields.

• mark_field_start – If True, mark this position as a field start.

update_from_stream(data: bytes) → None

Update buffer from a 3270 data stream using proper DataStreamParser for accurate 3270 order handling.

For backward compatibility with tests expecting simple byte sequences, this method detects whether the data appears to be raw bytes (no 3270 orders) vs. actual 3270 data streams.

Parameters:

data – Raw 3270 data stream bytes.

to_text() → str

Convert screen buffer to Unicode text string.

Returns:

Multi-line string representation.

get_content() → str

Retrieve the buffer content as a string.

get_field_at_position(row: int, col: int) → Field | None

Get the field containing the given position, if any (optimized with caching).

move_cursor_to_first_input_field() → None

Moves the cursor to the beginning of the first unprotected field with performance optimization.

move_cursor_to_next_input_field() → None

Moves the cursor to the beginning of the next unprotected, non-skipped, non-autoskip field. Wraps around to the first field if no next field is found.

In ASCII mode, this also considers ASCII fields detected by VT100 parser.

set_attribute(attr: int, row: int | None = None, col: int | None = None) → None

Set field attribute at specified or current position.

repeat_attribute(attr: int, repeat: int) → None

Repeat attribute a specified number of times.

graphic_ellipsis(count: int) → None

Insert graphic ellipsis characters.

insert_cursor() → None

Insert cursor at current position.

program_tab() → None

Program tab - move to next tab stop.

set_extended_attribute_sfe(attr_type: int, attr_value: int) → None

Accumulate extended field attributes for the field at the current buffer address.

Important semantics: - Extended attributes (from SFE or SA) do NOT consume a display byte and

must NOT create a field-start marker on their own.

• Only the base field attribute (set via SF or SFE type 0xC0) occupies a buffer position and should be recorded in _field_starts.

This method therefore ONLY updates the accumulated extended attributes and does not modify the display buffer or _field_starts.

set_extended_attribute(row: int, col: int, attr_type: str, value: int) → None

Set an extended attribute at a specific position (used by tests and APIs).

This also records a field start at that position to align with field-start-driven detection.

get_field_at(row: int, col: int) → Field | None

Alias for get_field_at_position for compatibility.

get_extended_attributes_at(row: int, col: int) → ExtendedAttributeSet | None

Get extended attributes for a specific position.

Parameters:

• row – Row position

• col – Column position

Returns:

ExtendedAttributeSet if attributes exist, None otherwise

set_field_validation(field: Field, validation_rules: ValidationAttribute) → None

Set validation rules for a field.

Parameters:

• field – The field to set validation for

• validation_rules – Validation attribute with rules

get_field_validation(field: Field) → ValidationAttribute | None

Get validation rules for a field.

Parameters:

field – The field to get validation for

Returns:

ValidationAttribute if set, None otherwise

validate_field_input(field: Field, input_text: str) → Tuple[bool, str | None]

Validate input against field’s validation rules.

Parameters:

• field – The field to validate input for

• input_text – The input text to validate

Returns:

Tuple of (is_valid, error_message)

select_light_pen(row: int, col: int) → int | None

Simulate a light pen selection at the given row and col.

If the field at the given position is light-pen selectable, this method will mark the field as selected, store the selection position, and return the light pen AID. Otherwise, it returns None.

reset_mdt_flags() → None

Reset Modified Data Tag flags for all input fields per RFC 1576.

set_keyboard_lock(locked: bool) → None

Lock or unlock keyboard for user input.

sound_alarm() → None

Trigger terminal alarm/bell.

terminal_reset() → None

Perform general terminal reset.

property keyboard_locked: bool

Check if keyboard is currently locked.

EBCDIC

EBCDIC to ASCII translation utilities for 3270 emulation. Based on IBM Code Page 037.

class pure3270.emulation.ebcdic.EmulationEncoder

Bases: object

Utility class for EBCDIC encoding/decoding in 3270 emulation.

The class exposes encode(text) -> bytes and decode(data) -> str which use IBM Code Page 037 (CP037) for conversions.

static decode(data: bytes) → str

Decode EBCDIC bytes to a printable ASCII string.

Uses CP037 decoding followed by mapping/normalization to ensure the returned string contains only printable ASCII characters (plus newline/carriage/tab). This hides raw EBCDIC/box-graphics from the caller and provides readable approximations for line-drawing chars.

static encode(text: str) → bytes

Encode a Unicode string to EBCDIC CP037 bytes.

pure3270.emulation.ebcdic.translate_ebcdic_to_ascii(data: bytes, codepage: str = 'cp037') → str

Translate raw EBCDIC bytes to ASCII string matching p3270 behavior.

Decode using specified EBCDIC codepage and return raw decoded string like p3270 does, without heavy filtering or normalization.

pure3270.emulation.ebcdic.translate_ascii_to_ebcdic(text: str, codepage: str = 'cp037') → bytes

Translate an ASCII/Unicode string to EBCDIC bytes using specified codepage.

pure3270.emulation.ebcdic.get_p3270_version() → str | None

Get p3270 version for patching.

Returns the installed p3270 package version string or None.

pure3270.emulation.ebcdic.encode_field_attribute(attr: int) → int

Encode 3270 field attribute to EBCDIC.

Current implementation treats attributes as direct values and returns the input unchanged. This is a placeholder for future mapping logic.

class pure3270.emulation.ebcdic.EBCDICCodec(codepage: str = 'cp037', compat: str | None = None)

Bases: object

Backwards-compatible EBCDIC codec used by tests.

This class provides a simple mapping-based encode/decode API that matches historical project tests: methods return tuples (value, length) and unknown characters are encoded as 0x7A and decoded as ‘z’.

The implementation uses configurable EBCDIC codec for deriving a base mapping where possible but intentionally restricts to reverse mapping to a conservative set (uppercase letters, digits and space) so that punctuation and other characters fall back to as ‘unknown’ mapping.

__init__(codepage: str = 'cp037', compat: str | None = None) → None

decode(data: bytes) → tuple[str, int]

Optimized decode with pre-computed character processing.

debug_decode_byte(byte: int) → str

Debug method to see how a single EBCDIC byte is decoded.

encode(text: str | bytes) → Tuple[bytes, int]

Encode text to EBCDIC bytes, returning (bytes, length).

Accepts either str or bytes. If bytes are provided they are interpreted as latin-1 and decoded before encoding so callers that pass raw bytes won’t raise. Prefers full CP037 mapping, falling back to conservative mapping only on failure. Characters not present in the conservative reverse mapping are encoded as 0x7A.

encode_to_unicode_table(text: str) → bytes

Encode without returning length (compat helper used by tests).

pure3270.emulation.ebcdic.detect_codepage_from_trace(trace_file: str) → str

Detect the appropriate EBCDIC codepage from a trace file.

Parameters:

trace_file – Path to the trace file

Returns:

The detected codepage (e.g., ‘cp037’, ‘cp937’, ‘cp933’)

Return type:

str

Printer Buffer

Printer buffer and rendering logic for 3287 printer emulation.

class pure3270.emulation.printer_buffer.PrinterBuffer(max_lines: int = 10000, auto_reset: bool = False)

Bases: BufferWriter

__init__(max_lines: int = 10000, auto_reset: bool = False) → None

reset() → None

Resets the printer buffer.

write_char(ebcdic_byte: int, row: int | None = None, col: int | None = None, protected: bool = False, circumvent_protection: bool = False) → None

Write an EBCDIC byte to the printer buffer.

The incoming value is an EBCDIC code point (0-255). Decode it to a Unicode character using the project’s EBCDICCodec rather than doing a direct chr() which treats the value as a Unicode codepoint and yields incorrect characters (e.g. 0xD9 -> ‘Ù’). Preserve simple positioning semantics used elsewhere in the codebase.

set_attribute(attr: int, row: int | None = None, col: int | None = None) → None

Set attribute at position. Printer buffer does not support attributes.

get_content() → str

Retrieve the buffer content as a cleaned, ASCII-safe string.

This method conservatively filters the rendered output to preserve printable ASCII and a few allowed control characters that are meaningful for printer output: newline, form-feed, and tab. It does NOT perform Unicode transliteration to avoid mangling decoded EBCDIC text.

write_scs_data(data: bytes, parser_pos: int | None = None) → None

Processes incoming SCS data, parsing SCS commands and robustly translating EBCDIC to ASCII.

Notes: - Consume SOH (0x01) and the following status byte (when present) to avoid misalignment. - Decode contiguous runs of non-control bytes in bulk using EBCDICCodec.decode

instead of decoding one byte at a time to preserve any multi-byte mappings.

• Preserve explicit handling for SCS control bytes: LF, CR, FF, HT.

• Suppress screen-like content such as trace metadata, screen buffer dumps, and field commands.

get_rendered_output() → str

Returns the current rendered output as a string.

get_buffer_content() → List[str]

Returns the raw buffer content (list of lines).

update_status(status_code: int) → None

Updates the printer’s internal status with the given status code.

Parameters:

status_code – The status code (e.g., 0x00 success, 0x40 device end).

This method updates the internal status state and can trigger events or updates for status SF handling in TN3270E printer sessions.

get_status() → int

Get the current printer status code.

Returns:

The current status code, or 0x00 if not set.

end_job() → None

Ends the current print job. Appends a literal ‘PRINT-EOJ’ marker to the buffer for display parity.

Field Attributes

Advanced field attribute classes for 3270 emulation.

class pure3270.emulation.field_attributes.AttributeType(value)

Bases: Enum

Enumeration of supported extended attribute types.

COLOR = 'color'

HIGHLIGHT = 'highlight'

VALIDATION = 'validation'

OUTLINING = 'outlining'

LIGHT_PEN = 'light_pen'

BACKGROUND = 'background'

CHARACTER_SET = 'character_set'

class pure3270.emulation.field_attributes.ExtendedAttribute(value: int | str = 0)

Bases: ABC

Abstract base class for extended field attributes.

Provides common functionality for all extended attribute types including validation, serialization, and thread-safe operations.

__init__(value: int | str = 0) → None

Initialize extended attribute with value.

Parameters:

value – The attribute value (int or string representation)

Raises:

ValueError – If the value is invalid for this attribute type

property value: int

Get the current attribute value.

abstract to_dict() → Dict[str, Any]

Convert attribute to dictionary representation.

Returns:

Dictionary containing attribute type and value

abstract from_dict(data: Dict[str, Any]) → None

Load attribute from dictionary representation.

Parameters:

data – Dictionary containing attribute data

Raises:

ValueError – If data is invalid

class pure3270.emulation.field_attributes.ExtendedAttributeSet

Bases: object

Container for managing multiple extended attributes for a field position.

Provides thread-safe operations for setting, getting, and managing extended attributes with proper validation and error handling.

get(attr_type: str, default: ExtendedAttribute | int | None = None) → ExtendedAttribute | int | None

Get an extended attribute by type, with optional default (test compatibility). Returns the raw value if the attribute is an object with a .value property.

__init__() → None

Initialize empty attribute set.

set_attribute(attr_type: str, attribute: Any) → None

Set an extended attribute.

Parameters:

• attr_type – The attribute type key

• attribute – The attribute instance

Raises:

TypeError – If attribute is not an ExtendedAttribute instance

get_attribute(attr_type: str) → Any

Get an extended attribute by type.

Parameters:

attr_type – The attribute type key

Returns:

The attribute instance or None if not found

remove_attribute(attr_type: str) → bool

Remove an extended attribute.

Parameters:

attr_type – The attribute type key

Returns:

True if attribute was removed, False if not found

has_attribute(attr_type: str) → bool

Check if an attribute type is present.

Parameters:

attr_type – The attribute type key

Returns:

True if attribute exists

clear() → None

Clear all attributes.

get_all_attributes() → Dict[str, ExtendedAttribute]

Get a copy of all attributes.

Returns:

Dictionary of all attributes

to_dict() → Dict[str, Dict[str, Any]]

Convert all attributes to dictionary representation.

Returns:

Dictionary mapping attribute types to their data

from_dict(data: Dict[str, Dict[str, Any]]) → None

Load attributes from dictionary representation.

Parameters:

data – Dictionary containing attribute data

Raises:

ValueError – If data contains invalid attributes

class pure3270.emulation.field_attributes.ColorAttribute(value: int | str = 0)

Bases: ExtendedAttribute

Color attribute supporting base 16 and extended 256 color support.

Supports both standard 3270 colors (0-15) and extended colors (16-255) as defined in TN3270E specifications.

BASE_COLORS = {0: 'neutral_black', 1: 'blue', 2: 'red', 3: 'pink', 4: 'green', 5: 'turquoise', 6: 'yellow', 7: 'neutral_white', 8: 'black', 9: 'deep_blue', 10: 'orange', 11: 'purple', 12: 'pale_green', 13: 'pale_turquoise', 14: 'grey', 15: 'white'}

__init__(value: int | str = 0) → None

Initialize color attribute.

Parameters:

value – Color value (0-255) or color name string

Raises:

ValueError – If value is invalid

get_color_name() → str | None

Get the color name for base colors.

Returns:

Color name if it’s a base color (0-15), None otherwise

is_base_color() → bool

Check if this is a base 16 color.

Returns:

True if color is in base 16 range (0-15)

is_extended_color() → bool

Check if this is an extended color.

Returns:

True if color is in extended range (16-255)

to_dict() → Dict[str, Any]

Convert color attribute to dictionary.

Returns:

Dictionary with color data

from_dict(data: Dict[str, Any]) → None

Load color attribute from dictionary.

Parameters:

data – Dictionary containing color data

Raises:

ValueError – If data is invalid

class pure3270.emulation.field_attributes.HighlightAttribute(value: int | str = 0)

Bases: ExtendedAttribute

Highlight attribute for visual highlighting effects.

Supports various highlighting effects like blink, reverse video, underscore, and intensity levels.

NORMAL = 0

BLINK = 1

REVERSE = 2

UNDERSCORE = 4

INTENSITY_HIGH = 8

INTENSITY_LOW = 16

EFFECT_NAMES = {0: 'normal', 1: 'blink', 2: 'reverse', 4: 'underscore', 8: 'high_intensity', 16: 'low_intensity'}

__init__(value: int | str = 0) → None

Initialize highlight attribute.

Parameters:

value – Highlight value or effect name

Raises:

ValueError – If value is invalid

get_effect_name() → str | None

Get the effect name for known effects.

Returns:

Effect name if recognized, None otherwise

has_effect(effect: int) → bool

Check if a specific effect is enabled.

Parameters:

effect – Effect constant to check

Returns:

True if effect is enabled

set_effect(effect: int, enabled: bool = True) → None

Set or clear a specific effect.

Parameters:

• effect – Effect constant

• enabled – Whether to enable or disable the effect

to_dict() → Dict[str, Any]

Convert highlight attribute to dictionary.

Returns:

Dictionary with highlight data

from_dict(data: Dict[str, Any]) → None

Load highlight attribute from dictionary.

Parameters:

data – Dictionary containing highlight data

Raises:

ValueError – If data is invalid

class pure3270.emulation.field_attributes.ValidationAttribute(value: int | str = 0, min_length: int | None = None, max_length: int | None = None, custom_rule: str | None = None)

Bases: ExtendedAttribute

Validation attribute for field validation rules.

Supports various validation constraints like mandatory fields, numeric-only input, length limits, and custom validation rules.

NONE = 0

MANDATORY = 1

NUMERIC = 2

ALPHABETIC = 4

ALPHANUMERIC = 8

LENGTH_MIN = 16

LENGTH_MAX = 32

CUSTOM = 64

VALIDATION_NAMES = {0: 'none', 1: 'mandatory', 2: 'numeric', 4: 'alphabetic', 8: 'alphanumeric', 16: 'min_length', 32: 'max_length', 64: 'custom'}

__init__(value: int | str = 0, min_length: int | None = None, max_length: int | None = None, custom_rule: str | None = None) → None

Initialize validation attribute.

Parameters:

• value – Validation type value or name

• min_length – Minimum field length (if LENGTH_MIN set)

• max_length – Maximum field length (if LENGTH_MAX set)

• custom_rule – Custom validation rule string (if CUSTOM set)

Raises:

ValueError – If parameters are invalid

get_validation_name() → str | None

Get the validation type name.

Returns:

Validation type name if recognized, None otherwise

has_validation(validation_type: int) → bool

Check if a specific validation type is enabled.

Parameters:

validation_type – Validation type constant

Returns:

True if validation type is enabled

set_validation(validation_type: int, enabled: bool = True) → None

Set or clear a specific validation type.

Parameters:

• validation_type – Validation type constant

• enabled – Whether to enable or disable the validation

validate_input(input_text: str) → Tuple[bool, str | None]

Validate input text against validation rules.

Parameters:

input_text – The input text to validate

Returns:

Tuple of (is_valid, error_message)

to_dict() → Dict[str, Any]

Convert validation attribute to dictionary.

Returns:

Dictionary with validation data

from_dict(data: Dict[str, Any]) → None

Load validation attribute from dictionary.

Parameters:

data – Dictionary containing validation data

Raises:

ValueError – If data is invalid

class pure3270.emulation.field_attributes.OutliningAttribute(value: int | str = 0)

Bases: ExtendedAttribute

Outlining attribute for field borders and outlines.

Supports various outlining styles like boxes, underlines, overlines, and custom border patterns.

NONE = 0

UNDERLINE = 1

OVERLINE = 2

LEFT_LINE = 4

RIGHT_LINE = 8

BOX = 15

DOUBLE_LINE = 16

DASHED = 32

DOTTED = 64

STYLE_NAMES = {0: 'none', 1: 'underline', 2: 'overline', 4: 'left_line', 8: 'right_line', 15: 'box', 16: 'double_line', 32: 'dashed', 64: 'dotted'}

__init__(value: int | str = 0) → None

Initialize outlining attribute.

Parameters:

value – Outlining style value or name

Raises:

ValueError – If value is invalid

get_style_name() → str | None

Get the outlining style name.

Returns:

Style name if recognized, None otherwise

has_style(style: int) → bool

Check if a specific outlining style is enabled.

Parameters:

style – Style constant to check

Returns:

True if style is enabled

set_style(style: int, enabled: bool = True) → None

Set or clear a specific outlining style.

Parameters:

• style – Style constant

• enabled – Whether to enable or disable the style

is_box() → bool

Check if this is a complete box outline.

Returns:

True if all four sides are outlined

to_dict() → Dict[str, Any]

Convert outlining attribute to dictionary.

Returns:

Dictionary with outlining data

from_dict(data: Dict[str, Any]) → None

Load outlining attribute from dictionary.

Parameters:

data – Dictionary containing outlining data

Raises:

ValueError – If data is invalid

class pure3270.emulation.field_attributes.LightPenAttribute(value: int | str = 0)

Bases: ExtendedAttribute

Light pen attribute for designating a field as light-pen selectable.

__init__(value: int | str = 0) → None

Initialize light pen attribute.

to_dict() → Dict[str, Any]

Convert light pen attribute to dictionary.

from_dict(data: Dict[str, Any]) → None

Load light pen attribute from dictionary.

class pure3270.emulation.field_attributes.BackgroundAttribute(value: int | str = 0)

Bases: ExtendedAttribute

Background color attribute supporting base 16 and extended 256 color support.

Supports both standard 3270 background colors (0-15) and extended colors (16-255) as defined in TN3270E specifications.

BASE_COLORS = {0: 'neutral_black', 1: 'blue', 2: 'red', 3: 'pink', 4: 'green', 5: 'turquoise', 6: 'yellow', 7: 'neutral_white', 8: 'black', 9: 'deep_blue', 10: 'orange', 11: 'purple', 12: 'pale_green', 13: 'pale_turquoise', 14: 'grey', 15: 'white'}

__init__(value: int | str = 0) → None

Initialize background color attribute.

Parameters:

value – Background color value (0-255) or color name string

Raises:

ValueError – If value is invalid

get_color_name() → str | None

Get the background color name for base colors.

Returns:

Color name if it’s a base color (0-15), None otherwise

is_base_color() → bool

Check if this is a base 16 background color.

Returns:

True if color is in base 16 range (0-15)

is_extended_color() → bool

Check if this is an extended background color.

Returns:

True if color is in extended range (16-255)

to_dict() → Dict[str, Any]

Convert background color attribute to dictionary.

Returns:

Dictionary with background color data

from_dict(data: Dict[str, Any]) → None

Load background color attribute from dictionary.

Parameters:

data – Dictionary containing background color data

Raises:

ValueError – If data is invalid

class pure3270.emulation.field_attributes.CharacterSetAttribute(value: int | str = 0)

Bases: ExtendedAttribute

Character set attribute for field-specific character set selection.

Supports various character sets as defined in 3270 and TN3270E specifications including APL, Katakana, and user-defined character sets.

CHARACTER_SETS = {0: 'default', 1: 'apl', 2: 'katakana', 3: 'user1', 4: 'user2', 5: 'user3', 6: 'user4', 65: 'dbcs_katakana', 66: 'dbcs_hiragana', 67: 'dbcs_kanji', 68: 'dbcs_mixed', 69: 'dbcs_chinese', 70: 'dbcs_korean', 241: 'dbcs_chinese_host', 242: 'dbcs_japanese_host', 243: 'dbcs_korean_host', 244: 'dbcs_thai_host', 245: 'dbcs_arabic_host', 246: 'dbcs_hebrew_host', 247: 'dbcs_latin2_host', 248: 'dbcs_latin3_host', 249: 'dbcs_latin4_host', 250: 'dbcs_greek_host', 251: 'dbcs_cyrillic_host', 252: 'dbcs_turkish_host', 253: 'dbcs_latin5_host', 254: 'dbcs_japanese_extended'}

__init__(value: int | str = 0) → None

Initialize character set attribute.

get_character_set_name() → str | None

Get the name of the character set.

is_default() → bool

Check if this is the default character set.

is_apl() → bool

Check if this is the APL character set.

is_katakana() → bool

Check if this is the Katakana character set.

is_user_defined() → bool

Check if this is a user-defined character set.

is_dbcs() → bool

Check if this is a Double-Byte Character Set (DBCS).

DBCS character sets require special handling for cursor advancement and character display, advancing the cursor by 2 positions instead of 1.

Returns:

True if this character set is DBCS, False otherwise

to_dict() → Dict[str, Any]

Convert character set attribute to dictionary.

from_dict(data: Dict[str, Any]) → None

Load character set attribute from dictionary.

Extended Position

Extended position class with 14-bit addressing support for large screen emulation.

class pure3270.emulation.extended_position.ExtendedPosition(row: int = 0, col: int = 0, cols: int = 80, addressing_mode: AddressingMode = AddressingMode.MODE_12_BIT)

Bases: object

Represents a position in a 3270 screen buffer with support for extended 14-bit addressing.

This class provides coordinate-based positioning with automatic conversion to/from linear addresses, supporting both traditional 12-bit and extended 14-bit addressing modes.

__init__(row: int = 0, col: int = 0, cols: int = 80, addressing_mode: AddressingMode = AddressingMode.MODE_12_BIT) → None

Initialize an ExtendedPosition.

Parameters:

• row – Row coordinate (0-based)

• col – Column coordinate (0-based)

• cols – Number of columns in the screen (default 80)

• addressing_mode – Addressing mode for validation and conversion

Raises:

ValueError – If coordinates are invalid for the addressing mode

property row: int

Get the row coordinate.

property col: int

Get the column coordinate.

property cols: int

Get the number of columns.

property addressing_mode: AddressingMode

Get the addressing mode.

property linear_address: int

Get the linear address for this position.

property coordinates: Tuple[int, int]

Get the (row, col) coordinates as a tuple.

set_coordinates(row: int, col: int) → None

Set new coordinates for this position.

Parameters:

• row – New row coordinate

• col – New column coordinate

Raises:

ValueError – If new coordinates are invalid

set_from_address(address: int) → None

Set coordinates from a linear address.

Parameters:

address – Linear address to convert from

Raises:

ValueError – If address is invalid for the addressing mode

convert_addressing_mode(new_mode: AddressingMode) → ExtendedPosition

Create a new ExtendedPosition with the same coordinates but different addressing mode.

Parameters:

new_mode – New addressing mode

Returns:

New ExtendedPosition instance with converted addressing mode

Raises:

ValueError – If coordinates are invalid in the new mode

is_valid_for_mode(mode: AddressingMode) → bool

Check if this position is valid for a given addressing mode.

Parameters:

mode – Addressing mode to check against

Returns:

True if position is valid for the mode, False otherwise

move_relative(row_delta: int = 0, col_delta: int = 0) → None

Move the position by relative offsets.

Parameters:

• row_delta – Rows to move (positive for down, negative for up)

• col_delta – Columns to move (positive for right, negative for left)

Raises:

ValueError – If new position is invalid

move_to_next_line() → None

Move to the beginning of the next line.

move_to_previous_line() → None

Move to the beginning of the previous line.

clamp_to_bounds(max_row: int, max_col: int) → None

Clamp coordinates to specified maximum bounds.

Parameters:

• max_row – Maximum row (exclusive, 0-based)

• max_col – Maximum column (exclusive, 0-based)

Extended Screen Buffer

Extended screen buffer with 14-bit addressing support for large screen emulation.

class pure3270.emulation.extended_screen_buffer.ExtendedScreenBuffer(rows: int = 24, cols: int = 80, init_value: int = 64, addressing_mode: AddressingMode = AddressingMode.MODE_12_BIT)

Bases: ScreenBuffer

Extended screen buffer supporting 14-bit addressing for large screens.

This class extends the traditional ScreenBuffer to support larger screen sizes through 14-bit addressing while maintaining full backward compatibility with existing ScreenBuffer API.

__init__(rows: int = 24, cols: int = 80, init_value: int = 64, addressing_mode: AddressingMode = AddressingMode.MODE_12_BIT)

Initialize the ExtendedScreenBuffer.

Parameters:

• rows – Number of rows (default 24)

• cols – Number of columns (default 80)

• init_value – Initial value for buffer positions (default 0x40 for EBCDIC space)

• addressing_mode – Addressing mode for the buffer

Raises:

ValueError – If dimensions exceed addressing mode limits

property addressing_mode: AddressingMode

Get the addressing mode for this buffer.

property extended_cursor: ExtendedPosition

Get the extended cursor position.

set_position(row: int, col: int, wrap: bool = False, strict: bool = False) → None

Set cursor position with extended addressing validation.

Parameters:

• row – Row coordinate

• col – Column coordinate

• wrap – Whether to wrap coordinates to valid range

• strict – If True, raise IndexError on out of bounds; if False, clamp values

Raises:

• ValueError – If position is invalid for the addressing mode

• IndexError – When strict=True and position is out of bounds

set_position_from_address(address: int) → None

Set cursor position from a linear address.

Parameters:

address – Linear address to set cursor to

Raises:

ValueError – If address is invalid for the addressing mode

get_position_address() → int

Get the current cursor position as a linear address.

Returns:

Linear address of current cursor position

write_char_at_address(ebcdic_byte: int, address: int, protected: bool = False, circumvent_protection: bool = False) → None

Write a character at a specific linear address.

Parameters:

• ebcdic_byte – EBCDIC byte value to write

• address – Linear address to write to

• protected – Whether to set protection attribute

• circumvent_protection – Whether to write even to protected fields

Raises:

ValueError – If address is invalid for the addressing mode

read_char_at_address(address: int) → int | None

Read a character from a specific linear address.

Parameters:

address – Linear address to read from

Returns:

EBCDIC byte value at the address, or None if address is invalid

is_address_valid(address: int) → bool

Check if an address is valid for this buffer’s addressing mode.

Parameters:

address – Address to validate

Returns:

True if address is valid, False otherwise

get_address_range() → Tuple[int, int]

Get the valid address range for this buffer.

Returns:

Tuple of (min_address, max_address) inclusive

convert_addressing_mode(new_mode: AddressingMode) → ExtendedScreenBuffer | None

Create a new ExtendedScreenBuffer with the same content but different addressing mode.

Parameters:

new_mode – New addressing mode

Returns:

New ExtendedScreenBuffer instance, or None if conversion fails

Raises:

ValueError – If current buffer size exceeds new mode limits

get_field_at_address(address: int) → Field | None

Get the field containing a specific linear address.

Parameters:

address – Linear address to check

Returns:

Field containing the address, or None if no field or invalid address

move_cursor_to_address(address: int) → None

Move cursor to a specific linear address.

Parameters:

address – Linear address to move cursor to

Raises:

ValueError – If address is invalid

Addressing

Addressing mode enumeration and utilities for 3270 emulation.

class pure3270.emulation.addressing.AddressingMode(value)

Bases: Enum

Enumeration of supported 3270 addressing modes.

MODE_12_BIT = '12-bit'

Traditional 12-bit addressing mode supporting up to 4096 positions (24x80, 32x80, etc.).

MODE_14_BIT = '14-bit'

Extended 14-bit addressing mode supporting up to 16384 positions for large screens.

class pure3270.emulation.addressing.AddressCalculator

Bases: object

Utility class for address calculations and conversions between addressing modes.

static validate_address(address: int, mode: AddressingMode) → bool

Validate if an address is valid for the given addressing mode.

Parameters:

• address – The address to validate (0-based linear position)

• mode – The addressing mode to validate against

Returns:

True if address is valid for the mode, False otherwise

static address_to_coords(address: int, cols: int, mode: AddressingMode) → Tuple[int, int] | None

Convert a linear address to row/column coordinates.

Parameters:

• address – Linear address (0-based)

• cols – Number of columns in the screen

• mode – Addressing mode

Returns:

Tuple of (row, col) coordinates, or None if address is invalid

static coords_to_address(row: int, col: int, cols: int, mode: AddressingMode) → int | None

Convert row/column coordinates to a linear address.

Parameters:

• row – Row coordinate (0-based)

• col – Column coordinate (0-based)

• cols – Number of columns in the screen

• mode – Addressing mode

Returns:

Linear address, or None if coordinates would exceed mode limits

static get_max_positions(mode: AddressingMode) → int

Get the maximum number of positions supported by the addressing mode.

Parameters:

mode – The addressing mode

Returns:

Maximum number of positions (1-based, so max address + 1)

static convert_address_mode(address: int, from_mode: AddressingMode, to_mode: AddressingMode, cols: int) → int | None

Convert an address from one addressing mode to another.

This is primarily useful for converting between 12-bit and 14-bit addressing when the same row/column coordinates are used but different linear addressing.

Parameters:

• address – The address to convert

• from_mode – Source addressing mode

• to_mode – Target addressing mode

• cols – Number of columns (needed for coordinate conversion)

Returns:

Converted address, or None if conversion is not possible

Buffer Writer

Common base class for buffer writers in pure3270 emulation.

class pure3270.emulation.buffer_writer.BufferWriter

Bases: ABC

Abstract base class consolidating write operations, cursor management, and content retrieval for screen, printer, and session buffers.

__init__() → None

set_position(row: int, col: int) → None

Set cursor position.

get_position() → Tuple[int, int]

Get current cursor position.

abstract write_char(ebcdic_byte: int, row: int | None = None, col: int | None = None, protected: bool = False, circumvent_protection: bool = False) → None

Write an EBCDIC character to the buffer at specified or current position. Handles insertion, protection checks, and overflow where applicable.

abstract set_attribute(attr: int, row: int | None = None, col: int | None = None) → None

Set attribute (e.g., protection, intensity) at specified or current position.

abstract get_content() → str

Retrieve the buffer content as a string.

Snapshot

Screen buffer snapshot and comparison system for regression testing.

class pure3270.emulation.snapshot.ScreenSnapshot(screen_buffer: ScreenBuffer)

Bases: object

Represents a snapshot of a ScreenBuffer state for regression testing.

__init__(screen_buffer: ScreenBuffer)

Create a snapshot from a ScreenBuffer instance.

to_dict() → Dict[str, Any]

Convert snapshot to dictionary for JSON serialization.

to_json() → str

Convert snapshot to JSON string.

classmethod from_dict(data: Dict[str, Any]) → ScreenSnapshot

Create snapshot from dictionary data.

classmethod from_json(json_str: str) → ScreenSnapshot

Create snapshot from JSON string.

to_screen_buffer() → ScreenBuffer

Reconstruct a ScreenBuffer from this snapshot.

save_to_file(filepath: str) → None

Save snapshot to a JSON file.

classmethod load_from_file(filepath: str) → ScreenSnapshot

Load snapshot from a JSON file.

class pure3270.emulation.snapshot.SnapshotComparison(snapshot1: ScreenSnapshot, snapshot2: ScreenSnapshot)

Bases: object

Represents the result of comparing two snapshots.

__init__(snapshot1: ScreenSnapshot, snapshot2: ScreenSnapshot)

Compare two snapshots and identify differences.

has_differences() → bool

Return True if there are any differences between snapshots.

get_summary() → Dict[str, Any]

Get a summary of the comparison results.

print_report() → None

Print a human-readable comparison report.

pure3270.emulation.snapshot.take_snapshot(screen_buffer: ScreenBuffer) → ScreenSnapshot

Convenience function to take a snapshot of a ScreenBuffer.

pure3270.emulation.snapshot.create_ascii_mode_snapshot(screen_buffer: ScreenBuffer) → ScreenSnapshot

Create a snapshot with the screen buffer in ASCII mode.

pure3270.emulation.snapshot.create_ebcdic_mode_snapshot(screen_buffer: ScreenBuffer) → ScreenSnapshot

Create a snapshot with the screen buffer in EBCDIC mode.

pure3270.emulation.snapshot.compare_snapshots(snapshot1: ScreenSnapshot, snapshot2: ScreenSnapshot) → SnapshotComparison

Convenience function to compare two snapshots.

class pure3270.emulation.snapshot.SnapshotDiffer(snapshot1: ScreenSnapshot, snapshot2: ScreenSnapshot)

Bases: object

Advanced snapshot comparison with detailed diff analysis.

__init__(snapshot1: ScreenSnapshot, snapshot2: ScreenSnapshot)

Initialize differ with two snapshots.

get_buffer_diff_positions() → List[Tuple[int, int, int, int]]

Get positions where buffer bytes differ.

get_attribute_diff_positions() → List[Tuple[int, int, int, int]]

Get positions where attributes differ.

get_field_differences() → Dict[str, Any]

Get detailed field differences.

Regression Harness

Exceptions

Enhanced exceptions for pure3270 with contextual information.

exception pure3270.exceptions.Pure3270Error(message: str, context: Dict[str, Any] | None = None, original_exception: Exception | None = None)

Bases: Exception

Base error for pure3270 with contextual information.

__init__(message: str, context: Dict[str, Any] | None = None, original_exception: Exception | None = None)

Initialize a pure3270 error.

Parameters:

• message – Error message

• context – Optional context information (host, port, operation, stream_pos, order_byte, trace, etc.)

• original_exception – Original exception that caused this error

add_context(key: str, value: Any) → None

Add context information to the exception.

Parameters:

• key – Context key

• value – Context value

get_context(key: str, default: Any | None = None) → Any

Get context information from the exception.

Parameters:

• key – Context key

• default – Default value if key not found

Returns:

Context value or default

exception pure3270.exceptions.ConnectionError(message: str, context: Dict[str, Any] | None = None, original_exception: Exception | None = None)

Bases: Pure3270Error

Connection error with connection-specific context.

exception pure3270.exceptions.NegotiationError(message: str, context: Dict[str, Any] | None = None, original_exception: Exception | None = None)

Bases: Pure3270Error

Negotiation error with protocol-specific context.

exception pure3270.exceptions.ParseError(message: str, context: Dict[str, Any] | None = None, original_exception: Exception | None = None)

Bases: Pure3270Error

Parsing error with data-specific context.

exception pure3270.exceptions.ProtocolError(message: str, context: Dict[str, Any] | None = None, original_exception: Exception | None = None)

Bases: Pure3270Error

Protocol error with protocol-specific context.

exception pure3270.exceptions.TimeoutError(message: str, context: Dict[str, Any] | None = None, original_exception: Exception | None = None)

Bases: Pure3270Error

Timeout error with operation-specific context.

exception pure3270.exceptions.NotConnectedError(message: str, context: Dict[str, Any] | None = None, original_exception: Exception | None = None)

Bases: Pure3270Error

Error raised when operation is attempted on a not connected session.

exception pure3270.exceptions.EnhancedSessionError(message: str, context: Dict[str, Any] | None = None, original_exception: Exception | None = None)

Bases: Pure3270Error

Enhanced session error with session-specific context.