Import reader/writer from amiga-brain-api

.github/workflows/ci_python.DISABLEDyml → .github/workflows/ci_python.yml

@@ -22,8 +22,10 @@ jobs:
       with:
         python-version: ${{ matrix.python-version }}
     - name: Install dependencies
-      working-directory: python
+      working-directory: py
       run: pip3 install -e .[dev]
     - name: Run Tests
-      working-directory: python
-      run: pytest -v tests/ --mypy
+      working-directory: py
+      # TODO: enable mypy later
+      # run: pytest -v tests/ --mypy
+      run: pytest -v tests/

.vscode/settings.json

@@ -0,0 +1,9 @@
+{
+    "python.testing.pytestArgs": [
+        "py"
+    ],
+    "python.testing.unittestEnabled": false,
+    "python.testing.pytestEnabled": true,
+    "python.linting.enabled": true,
+	"python.linting.pylintEnabled": true
+}

protos/farm_ng/core/event.proto

@@ -16,3 +16,8 @@ message Event {
   repeated Timestamp timestamps = 2;
   int64 payload_length = 3;
 }
+
+// TODO(edgar): remove after transition to new setup
+message EventHeader {
+  string robot_name = 1;
+}

py/README.md

@@ -0,0 +1,46 @@
+# farm-ng-core [python]
+
+## Install
+
+Install `pip3` & `virtualenv`:
+
+```bash
+sudo apt-get install python3-pip
+sudo pip3 install virtualenv
+```
+
+Clone the project:
+
+```bash
+git clone https://github.com/farm-ng/farm-ng-core.git
+```
+
+Move to the python code directory:
+
+```bash
+cd farm-ng-core/py
+```
+
+Start a virtual environment:
+
+```bash
+# assuming you're already in the amiga-brain-api/ directory
+python3 -m venv venv
+source venv/bin/activate
+```
+
+Create and install the `farm_ng.core` package:
+
+```bash
+# install to system
+pip3 install .
+
+# or for development mode
+pip3 install -e .[dev]
+```
+
+Verify that you have installed the package
+
+```bash
+python3 -c "import farm_ng; print(farm_ng.__version__)"
+```

py/farm_ng/core/events_file_reader.py

@@ -2,46 +2,59 @@
 import sys
 from collections import defaultdict
 from pathlib import Path
-from typing import Any
-from typing import BinaryIO
-from typing import Dict
+from typing import Any, Tuple
+from typing import cast
+from typing import DefaultDict
+from typing import IO
 from typing import List
 from typing import Optional
 
 from farm_ng.core import event_pb2
 from farm_ng.core.uri import Uri
+from farm_ng.core import uri_pb2
 
 
 class EventsFileReader:
     def __init__(self, file_name: Path) -> None:
         self._file_name = file_name.absolute()
         assert Path(self._file_name.parents[0]).is_dir()
 
-        self._file_stream: Optional[BinaryIO] = None
+        self._file_stream: Optional[IO] = None
         # assert self.open()
         # store the bytes offset in a dictionary where:
         # - the key is the string representation of the Uri of the message
         # - the value is a list of offsets to messages with that Uri
-        self.offsets: Optional[Dict[str, List[int]]] = None
+        self.offsets: DefaultDict[str, List[int]] = defaultdict(list)
         # self.compute_offsets()
 
     def __repr__(self) -> str:
-        return f"file_name: {str(self.file_name)}\nfile_stream: {self._file_stream}\nis_open: {self.is_open}"
+        return (
+            f"file_name: {str(self.file_name)}\n"
+            f"file_stream: {self._file_stream}\nis_open: {self.is_open}"
+        )
 
     def reset_offsets(self) -> None:
         self.offsets = defaultdict(list)
 
     def uris(self) -> List[Uri]:
+        if self.offsets is None:
+            return []
         return [Uri.from_string(key) for key in self.offsets.keys()]
 
     def has_uri(self, uri) -> bool:
+        if self.offsets is None:
+            return False
         return uri.string() in self.offsets.keys()
 
-    # TODO: discuss the api signature
-    def compute_offsets(self) -> Dict[Uri, List[int]]:
+    def compute_offsets(self) -> None:
         if not self.is_open:
             raise Exception("Reader not open. Please, use reader.open()")
 
+        maybe_file_stream = self._file_stream
+        if maybe_file_stream is None:
+            return None
+        file_stream = cast(IO, maybe_file_stream)
+
         # clear, if any the previous offsets
         self.reset_offsets()
 
@@ -54,16 +67,17 @@ def compute_offsets(self) -> Dict[Uri, List[int]]:
         assert header is not None
 
         current_offset += 1 + header.ByteSize()
+        header_offset = current_offset
 
         while True:
 
-            event_len = int.from_bytes(self._file_stream.read(1), sys.byteorder)
+            event_len = int.from_bytes(file_stream.read(1), sys.byteorder)
             # end file condition
             if event_len == 0:
-                self._file_stream.seek(0)
+                file_stream.seek(header_offset)
                 break
 
-            event: bytes = self._file_stream.read(event_len)
+            event: bytes = file_stream.read(event_len)
 
             event_proto = event_pb2.Event()
             event_proto.ParseFromString(event)
@@ -77,7 +91,7 @@ def compute_offsets(self) -> Dict[Uri, List[int]]:
 
             current_offset += 1 + event_len + skip_bytes
 
-            self._file_stream.seek(current_offset)
+            file_stream.seek(current_offset)
 
     @property
     def file_name(self) -> Optional[Path]:
@@ -95,6 +109,8 @@ def is_closed(self) -> bool:
 
     def open(self) -> bool:
         self._file_stream = open(self._file_name, "rb")
+        # can't always compute offsets?
+        # self.compute_offsets()
         return self.is_open()
 
     def close(self) -> bool:
@@ -111,43 +127,56 @@ def num_frames(self, uri: Uri) -> int:
     def seek(self, uri: Uri, frame_id: int) -> None:
         assert uri.string() in self.offsets.keys()
         assert frame_id < self.num_frames(uri)
-        self._file_stream.seek(self.offsets[uri.string()][frame_id])
+        maybe_file_stream = self._file_stream
+        if maybe_file_stream is None:
+            return None
+        file_stream = cast(IO, maybe_file_stream)
+        file_stream.seek(self.offsets[uri.string()][frame_id])
 
     def seek_and_read(self, uri: Uri, frame_idx: int) -> Optional[Any]:
         self.seek(uri, frame_idx)
         return self.read()
 
-    def read(self) -> Optional[Any]:
+    def read(self) -> Optional[Tuple[Any, uri_pb2.Uri]]:
         # Read the message's length as bytes and convert it to an integer
-        event_len = int.from_bytes(self._file_stream.read(1), sys.byteorder)
+        maybe_file_stream = self._file_stream
+        if maybe_file_stream is None:
+            return None
+        file_stream = cast(IO, maybe_file_stream)
+
+        event_len = int.from_bytes(file_stream.read(1), sys.byteorder)
         if event_len == 0:
             self.close()
             return None
 
         # Read that number of bytes as the message bytes
-        event: bytes = self._file_stream.read(event_len)
+        event: bytes = file_stream.read(event_len)
 
         event_proto = event_pb2.Event()
         event_proto.ParseFromString(event)
 
         # parse the message
         message_cls = getattr(
-            importlib.import_module(event_proto.module), event_proto.name
+            importlib.import_module(event_proto.uri.scheme), event_proto.uri.authority
         )
 
-        message: bytes = self._file_stream.read(event_proto.length_next)
+        message: bytes = file_stream.read(event_proto.payload_length)
 
         message_out = message_cls()
         message_out.ParseFromString(message)
 
-        return message_out
+        return event_proto, message_out
 
     def header(self) -> event_pb2.EventHeader:
-        self._file_stream.seek(0)
-        header_len = int.from_bytes(self._file_stream.read(1), sys.byteorder)
+        maybe_file_stream = self._file_stream
+        if maybe_file_stream is None:
+            return event_pb2.EventHeader()
+        file_stream = cast(IO, maybe_file_stream)
+        file_stream.seek(0)
+        header_len = int.from_bytes(file_stream.read(1), sys.byteorder)
 
         # Read that number of bytes as the message bytes
-        header: bytes = self._file_stream.read(header_len)
+        header: bytes = file_stream.read(header_len)
 
         header_proto = event_pb2.EventHeader()
         header_proto.ParseFromString(header)

py/farm_ng/core/events_file_writer.py

@@ -1,21 +1,26 @@
 import sys
 from pathlib import Path
 from typing import Any
-from typing import BinaryIO
+from typing import cast
+from typing import IO
 from typing import Optional
 
 from farm_ng.core import event_pb2
+from farm_ng.core import uri_pb2
 
 
 class EventsFileWriter:
     def __init__(self, file_name: Path) -> None:
         self._file_name = file_name.absolute()
         assert Path(self._file_name.parents[0]).is_dir()
 
-        self._file_stream: Optional[BinaryIO] = None
+        self._file_stream: Optional[IO] = None
 
     def __repr__(self) -> str:
-        return f"file_name: {str(self.file_name)}\nfile_stream: {self._file_stream}\nis_open: {self.is_open}"
+        return (
+            f"file_name: {str(self.file_name)}\n"
+            f"file_stream: {self._file_stream}\nis_open: {self.is_open}"
+        )
 
     @property
     def file_name(self) -> Optional[Path]:
@@ -36,24 +41,31 @@ def open(self) -> bool:
         return self.is_open()
 
     def close(self) -> bool:
-        self._file_stream.close()
+        maybe_file_stream = self._file_stream
+        if maybe_file_stream is None:
+            return False
+        file_stream = cast(IO, maybe_file_stream)
+
+        file_stream.close()
         self._file_stream = None
         return self.is_closed()
 
-    def write(self, message: Any, uri: Optional[event_pb2.Uri] = None) -> None:
+    def write(self, message: Any, uri: Optional[uri_pb2.Uri] = None) -> None:
+        maybe_file_stream = self._file_stream
+        if maybe_file_stream is None:
+            return None
+        file_stream = cast(IO, maybe_file_stream)
 
         if uri is None:
-            uri = event_pb2.Uri()
+            uri = uri_pb2.Uri()
 
         event = event_pb2.Event(
-            module=str(message.__module__),
-            name=str(message.__class__.__name__),
-            length_next=message.ByteSize(),
             uri=uri,
+            payload_length=message.ByteSize(),
         )
 
         event_len: bytes = event.ByteSize().to_bytes(1, sys.byteorder)
 
-        self._file_stream.write(event_len)
-        self._file_stream.write(event.SerializeToString())
-        self._file_stream.write(message.SerializeToString())
+        file_stream.write(event_len)
+        file_stream.write(event.SerializeToString())
+        file_stream.write(message.SerializeToString())

py/farm_ng/core/uri.py

@@ -1,8 +1,8 @@
-from farm_ng.core import event_pb2
+from farm_ng.core import uri_pb2
 
 
 class Uri:
-    def __init__(self, uri: event_pb2.Uri) -> None:
+    def __init__(self, uri: uri_pb2.Uri) -> None:
         self._uri = uri
 
     def __eq__(self, other: "Uri") -> bool:
@@ -13,17 +13,17 @@ def __repr__(self) -> str:
         return self.string()
 
     @property
-    def proto(self) -> event_pb2.Uri:
+    def proto(self) -> uri_pb2.Uri:
         return self._uri
 
     @property
     def scheme(self) -> str:
-        scheme_value = event_pb2.UriSchemeType.DESCRIPTOR.values[self._uri.scheme]
+        scheme_value = uri_pb2.UriSchemeType.DESCRIPTOR.values[self._uri.scheme]
         return scheme_value.name.lower()
 
     @classmethod
     def empty(cls) -> "Uri":
-        return Uri(event_pb2.Uri())
+        return Uri(uri_pb2.Uri())
 
     def string(self) -> str:
         return (
@@ -41,11 +41,11 @@ def from_string(self, string: str) -> "Uri":
     def from_strings(
         self, scheme_name: str, authority: str, path: str, query: str
     ) -> "Uri":
-        scheme_value = event_pb2.UriSchemeType.DESCRIPTOR.values_by_name[
+        scheme_value = uri_pb2.UriSchemeType.DESCRIPTOR.values_by_name[
             scheme_name.upper()
         ]
         return Uri(
-            event_pb2.Uri(
+            uri_pb2.Uri(
                 scheme=scheme_value.number, authority=authority, path=path, query=query
             )
         )