Semi-automatic test module & fixes for colors, commands, disk, fs, gps, keys

Author: neumond

computercraft/server.py

@@ -54,7 +54,7 @@ def __init__(self, nid, program, cleanup_callback):
         self._event_to_tids = {}
         self._tid_to_event = {}
 
-        self.colors = ColorsAPI
+        self.colors = ColorsAPI(self)
         self.commands = CommandsAPI(self)
         self.disk = DiskAPI(self)
         self.fs = FSAPI(self)
@@ -96,7 +96,7 @@ async def prog_wrap():
 
         self._task = asyncio.create_task(prog_wrap())
 
-    def _new_task_id(self):
+    def _new_task_id(self) -> str:
         task_id = base36(self._task_autoid)
         self._task_autoid += 1
         return task_id
@@ -150,7 +150,7 @@ class CCApplication(web.Application):
     async def _sender(ws, api):
         while not ws.closed:
             cmd = await api._cmd.get()
-            print(f'_sender: {cmd}')
+            # print(f'_sender: {cmd}')
             if not ws.closed:
                 await ws.send_json(cmd)
             if cmd['action'] == 'close':
@@ -159,7 +159,7 @@ async def _sender(ws, api):
     @staticmethod
     async def _json_messages(ws):
         async for msg in ws:
-            print('ws received', msg)
+            # print('ws received', msg)
             if msg.type != WSMsgType.TEXT:
                 continue
             # print('ws received', msg.data)

computercraft/subapis/base.py

@@ -85,10 +85,12 @@ def op_fn(v):
     return op_fn
 
 
-def make_single_value_return(cond):
+def make_single_value_return(validator, converter=None):
     def fn(v):
         assert len(v) == 1
-        assert cond(v[0])
+        if converter is not None:
+            v[0] = converter(v[0])
+        assert validator(v[0])
         return v[0]
     return fn
 
@@ -97,7 +99,11 @@ def fn(v):
 int_return = make_single_value_return(lambda v: isinstance(v, int) and not isinstance(v, bool))
 number_return = make_single_value_return(lambda v: isinstance(v, (int, float)) and not isinstance(v, bool))
 str_return = make_single_value_return(lambda v: isinstance(v, str))
-list_return = make_single_value_return(lambda v: isinstance(v, list))
+str_bool_return = make_single_value_return(lambda v: isinstance(v, (str, bool)))
+list_return = make_single_value_return(
+    lambda v: isinstance(v, list),
+    lambda v: [] if v == {} else v,
+)
 dict_return = make_single_value_return(lambda v: isinstance(v, dict))
 
 
@@ -110,6 +116,7 @@ def nil_return(v):
 opt_int_return = make_optional(int_return)
 opt_number_return = make_optional(number_return)
 opt_str_return = make_optional(str_return)
+opt_str_bool_return = make_optional(str_bool_return)
 opt_list_return = make_optional(list_return)
 opt_dict_return = make_optional(dict_return)
 

computercraft/subapis/colors.py

@@ -1,4 +1,11 @@
-class ColorsAPI:
+from typing import Tuple
+
+from .base import BaseSubAPI, bool_return, int_return
+
+
+class ColorsAPI(BaseSubAPI):
+    _API = 'colors'
+
     white = 0x1
     orange = 0x2
     magenta = 0x4
@@ -34,3 +41,25 @@ class ColorsAPI:
         'e': red,
         'f': black,
     }
+
+    async def combine(self, *colors: int) -> int:
+        return int_return(await self._send('combine', *colors))
+
+    async def subtract(self, color_set: int, *colors: int) -> int:
+        return int_return(await self._send('subtract', color_set, *colors))
+
+    async def test(self, colors: int, color: int) -> bool:
+        return bool_return(await self._send('test', colors, color))
+
+    async def packRGB(self, r: float, g: float, b: float) -> int:
+        return int_return(await self._send('packRGB', r, g, b))
+
+    async def unpackRGB(self, rgb: int) -> Tuple[float, float, float]:
+        ret = await self._send('unpackRGB', rgb)
+        assert isinstance(ret, list)
+        assert len(ret) == 3
+        return tuple(ret)
+
+    async def rgb8(self, *args):
+        r = await self._send('rgb8', *args)
+        return r[0] if len(r) == 1 else r

computercraft/subapis/commands.py

@@ -1,16 +1,29 @@
-from typing import Tuple, List
+from typing import Tuple, List, Optional
 from .base import BaseSubAPI, int_return, list_return, dict_return
+from ..errors import CommandException
 
 
 class CommandsAPI(BaseSubAPI):
     _API = 'commands'
 
-    async def exec(self, command: str):
+    async def exec(self, command: str) -> Tuple[str, Optional[int]]:
         r = await self._send('exec', command)
-        assert len(r) == 2
+        assert len(r) == 3
         assert isinstance(r[0], bool)
+
+        if r[1] == {}:
+            r[1] = []
         assert isinstance(r[1], list)
-        return r[0], r[1]
+        r[1] = '\n'.join(r[1])
+
+        if r[2] is None:
+            r[2] = 0
+        assert isinstance(r[2], int)
+
+        if r[0] is False:
+            raise CommandException(r[1])
+
+        return r[1], r[2]
 
     async def execAsync(self, command: str) -> int:
         return int_return(await self._send('execAsync', command))
@@ -19,7 +32,10 @@ async def list(self) -> List[str]:
         return list_return(await self._send('list'))
 
     async def getBlockPosition(self) -> Tuple[int, int, int]:
-        return tuple(await self._send('getBlockPosition'))
+        ret = await self._send('getBlockPosition')
+        assert isinstance(ret, list)
+        assert len(ret) == 3
+        return tuple(ret)
 
     async def getBlockInfo(self, x: int, y: int, z: int) -> dict:
         return dict_return(await self._send('getBlockInfo', x, y, z))

computercraft/subapis/disk.py

@@ -1,5 +1,5 @@
-from typing import Optional
-from .base import BaseSubAPI, nil_return, bool_return, opt_str_return, opt_int_return
+from typing import Optional, Union
+from .base import BaseSubAPI, nil_return, bool_return, opt_str_return, opt_int_return, opt_str_bool_return
 
 
 class DiskAPI(BaseSubAPI):
@@ -26,8 +26,8 @@ async def getID(self, side: str) -> Optional[int]:
     async def hasAudio(self, side: str) -> bool:
         return bool_return(await self._send('hasAudio', side))
 
-    async def getAudioTitle(self, side: str) -> Optional[str]:
-        return opt_str_return(await self._send('getAudioTitle', side))
+    async def getAudioTitle(self, side: str) -> Optional[Union[bool, str]]:
+        return opt_str_bool_return(await self._send('getAudioTitle', side))
 
     async def playAudio(self, side: str):
         return nil_return(await self._send('playAudio', side))

computercraft/subapis/fs.py

@@ -1,46 +1,29 @@
-from typing import Optional, Union, List
+from contextlib import asynccontextmanager
+from typing import Optional, List
+
 from .base import (
     BaseSubAPI, lua_string,
-    nil_return, opt_str_return, str_return, bool_return, int_return, list_return, opt_int_return,
+    nil_return, opt_str_return, str_return, bool_return,
+    int_return, list_return, dict_return,
 )
-from uuid import uuid4
 
 
-class CCFile(BaseSubAPI):
-    def __init__(self, cc, path, mode):
+class BaseHandle(BaseSubAPI):
+    def __init__(self, cc, var):
         super().__init__(cc)
-        self._path = path
-        self._mode = mode
-
-    async def __aenter__(self):
-        self._id = str(uuid4())
-        self._API = 'temp[{}]'.format(lua_string(self._id))
-        await self._cc._send_cmd('{} = fs.open({}, {})'.format(self._API, *map(lua_string, [
-            self._path, self._mode
-        ])))
-        return self
+        self._API = var
 
-    async def __aexit__(self, exc_type, exc, tb):
-        await self._cc._send_cmd('{}.close(); {} = nil'.format(self._API, self._API))
 
-    async def read(self) -> Optional[int]:
-        return opt_int_return(await self._send('read'))
+class ReadHandle(BaseHandle):
+    async def read(self, count: int) -> Optional[str]:
+        return opt_str_return(await self._send('read', count))
 
     async def readLine(self) -> Optional[str]:
         return opt_str_return(await self._send('readLine'))
 
     async def readAll(self) -> str:
         return str_return(await self._send('readAll'))
 
-    async def write(self, data: Union[str, int]):
-        return nil_return(await self._send('write', data))
-
-    async def writeLine(self, data: str):
-        return nil_return(await self._send('writeLine', data))
-
-    async def flush(self):
-        return nil_return(await self._send('flush'))
-
     def __aiter__(self):
         return self
 
@@ -51,6 +34,17 @@ async def __anext__(self):
         return line
 
 
+class WriteHandle(BaseHandle):
+    async def write(self, text: str):
+        return nil_return(await self._send('write', text))
+
+    async def writeLine(self, text: str):
+        return nil_return(await self._send('writeLine', text))
+
+    async def flush(self):
+        return nil_return(await self._send('flush'))
+
+
 class FSAPI(BaseSubAPI):
     _API = 'fs'
 
@@ -66,9 +60,6 @@ async def isDir(self, path: str) -> bool:
     async def isReadOnly(self, path: str) -> bool:
         return bool_return(await self._send('isReadOnly', path))
 
-    async def getName(self, path: str) -> str:
-        return str_return(await self._send('getName', path))
-
     async def getDrive(self, path: str) -> Optional[str]:
         return opt_str_return(await self._send('getDrive', path))
 
@@ -78,6 +69,9 @@ async def getSize(self, path: str) -> int:
     async def getFreeSpace(self, path: str) -> int:
         return int_return(await self._send('getFreeSpace', path))
 
+    async def getCapacity(self, path: str) -> int:
+        return int_return(await self._send('getCapacity', path))
+
     async def makeDir(self, path: str):
         return nil_return(await self._send('makeDir', path))
 
@@ -93,7 +87,8 @@ async def delete(self, path: str):
     async def combine(self, basePath: str, localPath: str) -> str:
         return str_return(await self._send('combine', basePath, localPath))
 
-    def open(self, path: str, mode: str) -> CCFile:
+    @asynccontextmanager
+    async def open(self, path: str, mode: str):
         '''
         Usage:
 
@@ -104,15 +99,32 @@ def open(self, path: str, mode: str) -> CCFile:
             async for line in f:
                 ...
         '''
-        return CCFile(self._cc, path, mode)
+        fid = self._cc._new_task_id()
+        var = 'temp[{}]'.format(lua_string(fid))
+        await self._cc._send_cmd('{} = fs.open({}, {})'.format(
+            var, *map(lua_string, [path, mode])))
+        try:
+            yield (ReadHandle if mode == 'r' else WriteHandle)(self._cc, var)
+        finally:
+            await self._cc._send_cmd('{}.close(); {} = nil'.format(var, var))
 
     async def find(self, wildcard: str) -> List[str]:
         return list_return(await self._send('find', wildcard))
 
     async def getDir(self, path: str) -> str:
         return str_return(await self._send('getDir', path))
 
+    async def getName(self, path: str) -> str:
+        return str_return(await self._send('getName', path))
+
+    async def isDriveRoot(self, path: str) -> bool:
+        return bool_return(await self._send('isDriveRoot', path))
+
     async def complete(
-        self, partialName: str, path: str, includeFiles: bool = None, includeSlashes: bool = None,
+        self, partialName: str, path: str, includeFiles: bool = None, includeDirs: bool = None,
     ) -> List[str]:
-        return list_return(await self._send('complete', partialName, path, includeFiles, includeSlashes))
+        return list_return(await self._send(
+            'complete', partialName, path, includeFiles, includeDirs, omit_nulls=False))
+
+    async def attributes(self, path: str) -> dict:
+        return dict_return(await self._send('attributes', path))

computercraft/subapis/gps.py

@@ -5,8 +5,12 @@
 class GpsAPI(BaseSubAPI):
     _API = 'gps'
 
+    CHANNEL_GPS = 65534
+
     async def locate(self, timeout: LuaNum = None, debug: bool = None) -> Optional[Tuple[int, int, int]]:
-        r = await self._send('locate', timeout, debug)
-        if r == [None]:
+        r = await self._send('locate', timeout, debug, omit_nulls=False)
+        if r == []:
             return None
+        assert isinstance(r, list)
+        assert len(r) == 3
         return tuple(r)

computercraft/subapis/keys.py

@@ -1,8 +1,10 @@
-from .base import BaseSubAPI, str_return
+from typing import Optional
+
+from .base import BaseSubAPI, opt_str_return
 
 
 class KeysAPI(BaseSubAPI):
     _API = 'keys'
 
-    async def getName(self, code: int) -> str:
-        return str_return(await self._send('getName', code))
+    async def getName(self, code: int) -> Optional[str]:
+        return opt_str_return(await self._send('getName', code))