Window API; terminal redirections

Author: neumond

computercraft/rproc.py

@@ -116,6 +116,7 @@ def proc(result):
 tuple3_number = fact_tuple(number, number, number)
 tuple2_integer = fact_tuple(integer, integer)
 tuple3_integer = fact_tuple(integer, integer, integer)
+tuple3_string = fact_tuple(string, string, string)
 array_integer = fact_array(integer)
 array_string = fact_array(string)
 option_integer = fact_option(integer)

computercraft/server.py

@@ -2,12 +2,14 @@
 import json
 import string
 from aiohttp import web, WSMsgType
+from contextlib import asynccontextmanager
 from traceback import print_exc
 from os.path import join, dirname, abspath
 from importlib import import_module
 import argparse
 
 from .subapis.root import RootAPIMixin
+from .lua import lua_string
 from . import rproc
 
 from .subapis.colors import ColorsAPI
@@ -148,6 +150,18 @@ async def _stop_queue(self, task_id):
                 'event': event,
             })
 
+    @asynccontextmanager
+    async def _create_temp_object(self, create_expr: str, finalizer_template: str = ''):
+        fid = self._new_task_id()
+        var = 'temp[{}]'.format(lua_string(fid))
+        await self.eval_coro('{} = {}'.format(var, create_expr))
+        try:
+            yield var
+        finally:
+            finalizer_template += '; {e} = nil'
+            finalizer_template = finalizer_template.lstrip(' ;')
+            await self.eval_coro(finalizer_template.format(e=var))
+
 
 class CCApplication(web.Application):
     @staticmethod

computercraft/subapis/fs.py

@@ -2,7 +2,7 @@
 from typing import Optional, List
 
 from .base import BaseSubAPI
-from ..lua import lua_string
+from ..lua import lua_args
 from ..rproc import boolean, string, integer, nil, array_string, option_string, fact_scheme_dict
 
 
@@ -97,14 +97,13 @@ async def open(self, path: str, mode: str):
             async for line in f:
                 ...
         '''
-        fid = self._cc._new_task_id()
-        var = 'temp[{}]'.format(lua_string(fid))
-        await self._cc.eval_coro('{} = fs.open({}, {})'.format(
-            var, lua_string(path), lua_string(mode)))
-        try:
+        create_expr = '{}.open({})'.format(
+            self.get_expr_code(),
+            lua_args(path, mode),
+        )
+        fin_tpl = '{e}.close()'
+        async with self._cc._create_temp_object(create_expr, fin_tpl) as var:
             yield (ReadHandle if 'r' in mode else WriteHandle)(self._cc, var)
-        finally:
-            await self._cc.eval_coro('{}.close(); {} = nil'.format(var, var))
 
     async def find(self, wildcard: str) -> List[str]:
         return array_string(await self._send('find', wildcard))

computercraft/subapis/mixins.py

@@ -1,5 +1,6 @@
 from typing import Tuple
 
+from ..lua import LuaExpr
 from ..rproc import boolean, nil, integer, tuple3_number, tuple2_integer
 
 
@@ -54,3 +55,11 @@ async def getPaletteColor(self, colorID: int) -> Tuple[float, float, float]:
 
     async def setPaletteColor(self, colorID: int, r: float, g: float, b: float):
         return nil(await self._send('setPaletteColor', colorID, r, g, b))
+
+
+class TermTarget(LuaExpr):
+    def __init__(self, code):
+        self._code = code
+
+    def get_expr_code(self):
+        return self._code

computercraft/subapis/os.py

@@ -2,7 +2,7 @@
 from typing import Optional, List
 
 from .base import BaseSubAPI
-from ..lua import LuaTable, LuaNum
+from ..lua import LuaNum
 from ..rproc import nil, string, option_string, number, integer, boolean
 
 
@@ -36,7 +36,7 @@ async def getComputerLabel(self) -> Optional[str]:
     async def setComputerLabel(self, label: Optional[str]):
         return nil(await self._send('setComputerLabel', label))
 
-    async def run(self, environment: LuaTable, programPath: str, *args: List[str]):
+    async def run(self, environment: dict, programPath: str, *args: List[str]):
         return boolean(await self._send('run', environment, programPath, *args))
 
     @asynccontextmanager

computercraft/subapis/peripheral.py

@@ -3,7 +3,7 @@
 from typing import Optional, List, Tuple, Any, Union
 
 from .base import BaseSubAPI
-from .mixins import TermMixin
+from .mixins import TermMixin, TermTarget
 from ..lua import LuaNum, lua_args
 from ..rproc import (
     boolean, nil, integer, string, option_integer, option_string,
@@ -118,6 +118,10 @@ async def transmit(self, channel: int, replyChannel: int, message: Any):
     async def isWireless(self) -> bool:
         return boolean(await self._send('isWireless'))
 
+    @property
+    def _side(self):
+        return self._prepend_params[0]
+
     def _mk_recv_filter(self, channel):
         def filter(msg):
             if msg[0] != self._side:
@@ -159,6 +163,7 @@ async def isPresentRemote(self, peripheralName: str) -> bool:
 
     async def wrapRemote(self, peripheralName: str) -> Optional[BasePeripheral]:
         # use instead getMethodsRemote and callRemote
+        # NOTE: you can also use peripheral.wrap(peripheralName)
 
         ptype = await self.getTypeRemote(peripheralName)
         if ptype is None:
@@ -170,6 +175,8 @@ async def wrapRemote(self, peripheralName: str) -> Optional[BasePeripheral]:
             'callRemote', peripheralName,
         )
 
+    # NOTE: for TermTarget use peripheral.get_term_target(peripheralName)
+
 
 class CCPrinter(BasePeripheral):
     async def newPage(self) -> bool:
@@ -276,3 +283,9 @@ async def wrap(self, side: str) -> Optional[BasePeripheral]:
                 return CCWiredModem(self._cc, m, side)
         else:
             return TYPE_MAP[ptype](self._cc, m, side)
+
+    def get_term_target(self, side: str) -> TermTarget:
+        return TermTarget('{}.wrap({})'.format(
+            self.get_expr_code(),
+            lua_args(side),
+        ))

computercraft/subapis/term.py

@@ -1,36 +1,30 @@
+from contextlib import asynccontextmanager
 from typing import Tuple
 
 from .base import BaseSubAPI
-from .mixins import TermMixin
-from ..rproc import nil, tuple2_integer, tuple3_number
-
-
-class CCWindow(BaseSubAPI, TermMixin):
-    # TODO
-    def __init__(self, cc):
-        super().__init__(cc)
-
-    async def setVisible(self, visibility: bool):
-        return nil(await self._send('setVisible', visibility))
-
-    async def redraw(self):
-        return nil(await self._send('redraw'))
-
-    async def restoreCursor(self):
-        return nil(await self._send('restoreCursor'))
-
-    async def getPosition(self) -> Tuple[int, int]:
-        return tuple2_integer(await self._send('getPosition'))
-
-    async def reposition(self, x: int, y: int, width: int = None, height: int = None):
-        return nil(await self._send('reposition', x, y, width, height))
+from .mixins import TermMixin, TermTarget
+from ..lua import lua_args
+from ..rproc import tuple3_number
 
 
 class TermAPI(BaseSubAPI, TermMixin):
     async def nativePaletteColor(self, colorID: int) -> Tuple[float, float, float]:
         return tuple3_number(await self._send('nativePaletteColor', colorID))
 
-    # TODO
-    # term.redirect(target) 	table previous terminal object
-    # term.current() 	table terminal object
-    # term.native() 	table terminal object
+    @asynccontextmanager
+    async def redirect(self, target: TermTarget):
+        create_expr = '{}.redirect({})'.format(
+            self.get_expr_code(),
+            lua_args(target),
+        )
+        fin_tpl = '{}.redirect({{e}})'.format(
+            self.get_expr_code(),
+        )
+        async with self._cc._create_temp_object(create_expr, fin_tpl):
+            yield
+
+    def get_current_target(self) -> TermTarget:
+        return TermTarget('{}.current()'.format(self.get_expr_code()))
+
+    def get_native_target(self) -> TermTarget:
+        return TermTarget('{}.native()'.format(self.get_expr_code()))

computercraft/subapis/window.py

@@ -1,10 +1,43 @@
+from contextlib import asynccontextmanager
+from typing import Tuple
+
+from ..lua import lua_args
+from ..rproc import nil, tuple2_integer, tuple3_string
 from .base import BaseSubAPI
-from ..lua import LuaTable
+from .mixins import TermMixin, TermTarget
+
+
+class CCWindow(BaseSubAPI, TermMixin):
+    async def setVisible(self, visibility: bool):
+        return nil(await self._send('setVisible', visibility))
+
+    async def redraw(self):
+        return nil(await self._send('redraw'))
+
+    async def restoreCursor(self):
+        return nil(await self._send('restoreCursor'))
+
+    async def getPosition(self) -> Tuple[int, int]:
+        return tuple2_integer(await self._send('getPosition'))
+
+    async def reposition(self, x: int, y: int, width: int = None, height: int = None, parent: TermTarget = None):
+        return nil(await self._send('reposition', x, y, width, height, parent))
+
+    async def getLine(self, y: int) -> Tuple[str, str, str]:
+        return tuple3_string(await self._send('getLine', y))
+
+    def get_term_target(self) -> TermTarget:
+        return TermTarget(self.get_expr_code())
 
 
 class WindowAPI(BaseSubAPI):
+    @asynccontextmanager
     async def create(
-        self, parentTerm: LuaTable, x: int, y: int, width: int, height: int, visible: bool = None,
-    ) -> LuaTable:
-        # TODO
-        return await self._send('create', parentTerm, x, y, width, height, visible)
+        self, parentTerm: TermTarget, x: int, y: int, width: int, height: int, visible: bool = None,
+    ) -> CCWindow:
+        create_expr = '{}.create({})'.format(
+            self.get_expr_code(),
+            lua_args(parentTerm, x, y, width, height, visible),
+        )
+        async with self._cc._create_temp_object(create_expr) as var:
+            yield CCWindow(self._cc, var)

testmod.py

@@ -1,6 +1,6 @@
 import asyncio
 import random
-from contextlib import contextmanager
+from contextlib import contextmanager, AsyncExitStack
 from time import monotonic
 from types import FunctionType
 
@@ -1632,4 +1632,106 @@ async def test_shell(api):
     await api.print('Test finished successfully')
 
 
+async def test_window(api):
+    async with api.window.create(
+        api.term.get_current_target(),
+        15, 5, 5, 5, False,
+    ) as win:
+        assert await win.getPosition() == (15, 5)
+        assert await win.getSize() == (5, 5)
+
+        await win.setBackgroundColor(api.colors.red)
+        await win.clear()
+        await win.setVisible(True)
+
+        await asyncio.sleep(1)
+
+        await win.setVisible(False)
+        await win.setCursorPos(1, 1)
+        await win.setTextColor(api.colors.yellow)
+        await win.write('*********')
+        await win.setVisible(True)
+
+        await asyncio.sleep(1)
+
+        await api.term.clear()
+
+        await asyncio.sleep(1)
+
+        await win.redraw()
+        assert await win.getLine(1) == ('*****', '44444', 'eeeee')
+
+        # draws immediately
+        await win.reposition(21, 5)
+        await win.reposition(27, 5)
+
+    await api.print('Test finished successfully')
+
+
+async def test_redirect_to_window(api):
+    w, h = await api.term.getSize()
+    async with AsyncExitStack() as stack:
+        left = await stack.enter_async_context(api.window.create(
+            api.term.get_current_target(),
+            1, 1, w // 2, h, True,
+        ))
+        right = await stack.enter_async_context(api.window.create(
+            api.term.get_current_target(),
+            w // 2 + 1, 1, w // 2, h, True,
+        ))
+        async with api.term.redirect(left.get_term_target()):
+            await api.term.setBackgroundColor(api.colors.green)
+            await api.term.setTextColor(api.colors.white)
+            await api.term.clear()
+            await api.term.setCursorPos(1, h // 2)
+            await api.print('Left part')
+        async with api.term.redirect(right.get_term_target()):
+            await api.term.setBackgroundColor(api.colors.red)
+            await api.term.setTextColor(api.colors.yellow)
+            await api.term.clear()
+            await api.term.setCursorPos(1, h // 2)
+            await api.print('Right part')
+        await api.print('Default terminal restored')
+
+    await api.print('Test finished successfully')
+
+
+async def test_redirect_to_local_monitor(api):
+    side = 'left'
+    await step(api, f'Attach 3x3 color monitor to {side} side of computer')
+
+    async with api.term.redirect(api.peripheral.get_term_target(side)):
+        await api.term.setBackgroundColor(api.colors.green)
+        await api.term.setTextColor(api.colors.white)
+        await api.term.clear()
+        await api.term.setCursorPos(1, 1)
+        await api.print('Redirected to monitor')
+
+    await api.print('Test finished successfully')
+
+
+async def test_redirect_to_remote_monitor(api):
+    side = 'back'
+    await step(api, f'Attach wired modem to {side} side of computer')
+
+    mod = await api.peripheral.wrap(side)
+
+    await step(api, 'Connect remote monitor using wires, activate its modem')
+
+    for name in await mod.getNamesRemote():
+        if await mod.getTypeRemote(name) == 'monitor':
+            break
+    else:
+        assert False
+
+    async with api.term.redirect(api.peripheral.get_term_target(name)):
+        await api.term.setBackgroundColor(api.colors.blue)
+        await api.term.setTextColor(api.colors.white)
+        await api.term.clear()
+        await api.term.setCursorPos(1, 1)
+        await api.print('Redirected to monitor')
+
+    await api.print('Test finished successfully')
+
+
 # vector won't be implemented, use python equivalent