Add support for TN3270 alarm and VT100 bell

oec/controller.py

@@ -5,7 +5,7 @@ oec.controller
 
 import time
 import logging
-from coax import poll, poll_ack, KeystrokePollResponse, ReceiveTimeout, \
+from coax import poll, poll_ack, PollAction, KeystrokePollResponse, ReceiveTimeout, \
                  ReceiveError, ProtocolError
 
 from .terminal import Terminal, read_terminal_ids
@@ -166,7 +166,9 @@ class Controller:
 
         self.last_poll_time = time.perf_counter()
 
-        poll_response = poll(self.interface, timeout=1)
+        poll_action = self.terminal.get_poll_action() if self.terminal else PollAction.NONE
+
+        poll_response = poll(self.interface, poll_action, timeout=1)
 
         if poll_response:
             try:

oec/terminal.py

@@ -5,7 +5,8 @@ oec.terminal
 
 import time
 import logging
-from coax import read_terminal_id, read_extended_id, ReceiveError, ProtocolError
+from coax import read_terminal_id, read_extended_id, PollAction, ReceiveError, \
+                 ProtocolError
 
 from .display import Dimensions, Display 
 from .keyboard import Keyboard
@@ -67,3 +68,16 @@ class Terminal:
 
         self.display = Display(interface, dimensions)
         self.keyboard = Keyboard(keymap)
+
+        self.alarm = False
+
+    def sound_alarm(self):
+        self.alarm = True
+
+    def get_poll_action(self):
+        if self.alarm:
+            self.alarm = False
+
+            return PollAction.ALARM
+
+        return PollAction.NONE

oec/tn3270.py

@@ -71,6 +71,8 @@ class TN3270Session(Session):
 
         self.emulator = Emulator(self.telnet, rows, columns)
 
+        self.emulator.alarm = lambda: self.terminal.sound_alarm()
+
     def terminate(self):
         if self.telnet:
             self._disconnect_host()

oec/vt100.py

@@ -86,6 +86,14 @@ class VT100Session(Session):
 
         self.vt100_screen.write_process_input = lambda data: self.host_process.write(data.encode())
 
+        # Unfortunately multiple VT100 bells will be replaced with a single 3270 terminal
+        # alarm - also because the alarm is only sounded on terminal POLL the alarm sound
+        # may appear out of sync with the terminal.
+        #
+        # A better approach may be to perform a flush when the bell is encountered but
+        # that does not appear possible with the standard pyte ByteStream.
+        self.vt100_screen.bell = lambda: self.terminal.sound_alarm()
+
         self.vt100_stream = pyte.ByteStream(self.vt100_screen)
 
     def start(self):

requirements.txt

@@ -1,8 +1,8 @@
 ptyprocess==0.6.0
-pycoax==0.1.2
+pycoax==0.2.0
 pyserial==3.4
 pyte==0.8.0
-pytn3270==0.3.0
+pytn3270==0.4.0
 sliplib==0.3.0
 sortedcontainers==2.1.0
 wcwidth==0.1.7

tests/test_controller.py

@@ -1,6 +1,6 @@
 import unittest
 from unittest.mock import Mock, PropertyMock, patch
-from coax import PowerOnResetCompletePollResponse, KeystrokePollResponse, ReceiveTimeout
+from coax import PollAction, PowerOnResetCompletePollResponse, KeystrokePollResponse, ReceiveTimeout
 from coax.protocol import TerminalId
 
 import context
@@ -97,6 +97,23 @@ class RunLoopTestCase(unittest.TestCase):
 
         self.assertEqual(self.create_session_mock.call_count, 2)
 
+    def test_alarm(self):
+        # Arrange
+        self._assert_run_loop(0, PowerOnResetCompletePollResponse(0xa), 0, True)
+        self._assert_run_loop(0, None, 0, False)
+
+        self.assertIsNotNone(self.controller.terminal)
+
+        # Act
+        self.controller.terminal.sound_alarm()
+
+        # Assert
+        self._assert_run_loop(0.5, None, 0.5, False)
+
+        self.assertEqual(self.poll_mock.call_args[0][1], PollAction.ALARM)
+
+        self.assertFalse(self.controller.terminal.alarm)
+
     def _assert_run_loop(self, poll_time, poll_response, expected_delay, expected_poll_ack):
         # Arrange
         self.poll_mock.side_effect = [poll_response]

tests/test_terminal.py

@@ -0,0 +1,11 @@
+import unittest
+
+import context
+
+from oec.keymap_3278_2 import KEYMAP as KEYMAP_3278_2
+
+class TerminalGetPollActionTestCase(unittest.TestCase):
+    def setUp(self):
+        self.interface = Mock()
+
+        self.terminal = Terminal(self.interface, TerminalId(0b11110100), 'c1348300', KEYMAP_3278_2)

tests/test_vt100.py

@@ -8,34 +8,50 @@ from oec.keyboard import Key, KeyboardModifiers
 from oec.vt100 import VT100Session, select
 
 class SessionHandleHostTestCase(unittest.TestCase):
-    @patch('oec.vt100.select')
-    def test(self, select_mock):
+    def setUp(self):
+        self.terminal = Mock()
+
+        self.terminal.display.dimensions = Dimensions(24, 80)
+
+        self.session = VT100Session(self.terminal, None)
+
+        self.session.host_process = Mock()
+
+        patcher = patch('oec.vt100.select')
+
+        select_mock = patcher.start()
+
+        select_mock.return_value = [[self.session.host_process]]
+
+        self.addCleanup(patch.stopall)
+
+    def test(self):
         # Arrange
-        terminal = Mock()
-
-        terminal.display.dimensions = Dimensions(24, 80)
-
-        session = VT100Session(terminal, None)
-
-        session.host_process = Mock()
-
-        session.host_process.read = Mock(return_value=b'abc')
-
-        select_mock.return_value = [[session.host_process]]
+        self.session.host_process.read = Mock(return_value=b'abc')
 
         # Act
-        session.handle_host()
+        self.session.handle_host()
 
         # Assert
-        terminal.display.buffered_write.assert_any_call(0x80, row=0, column=0)
-        terminal.display.buffered_write.assert_any_call(0x81, row=0, column=1)
-        terminal.display.buffered_write.assert_any_call(0x82, row=0, column=2)
+        self.terminal.display.buffered_write.assert_any_call(0x80, row=0, column=0)
+        self.terminal.display.buffered_write.assert_any_call(0x81, row=0, column=1)
+        self.terminal.display.buffered_write.assert_any_call(0x82, row=0, column=2)
 
-        terminal.display.flush.assert_called()
+        self.terminal.display.flush.assert_called()
 
-        terminal.display.move_cursor.assert_called_with(row=0, column=3)
+        self.terminal.display.move_cursor.assert_called_with(row=0, column=3)
 
-        self.assertFalse(session.vt100_screen.dirty)
+        self.assertFalse(self.session.vt100_screen.dirty)
+
+    def test_bell(self):
+        # Arrange
+        self.session.host_process.read = Mock(return_value=b'\a')
+
+        # Act
+        self.session.handle_host()
+
+        # Assert
+        self.terminal.sound_alarm.assert_called()
 
 class SessionHandleKeyTestCase(unittest.TestCase):
     def setUp(self):