Merge pull request #310 from ICRAR/binary_fix

Fix writing of BytesIO data

Author: awicenec

daliuge-engine/dlg/data/drops/memory.py

@@ -42,7 +42,7 @@ def parse_pydata(pd_dict: dict) -> bytes:
     :returns a byte encoded value
     """
     pydata = pd_dict["value"]
-    logger.debug(f"pydata value provided: {pydata}, {type(pydata)}")
+    logger.debug(f"pydata value provided: {pydata}, {pd_dict['type'].lower()}")
 
     if pd_dict["type"].lower() == "json":
         try:

daliuge-engine/dlg/drop_loaders.py

@@ -102,6 +102,8 @@ def load_numpy(drop: "DataDROP", allow_pickle=True):
 
 
 import dill
+
+
 def load_dill(drop: "DataDROP"):
     """
     Load dill
@@ -116,28 +118,30 @@ def load_dill(drop: "DataDROP"):
     drop.close(desc)
     return dill.loads(buf.getbuffer())
 
-def load_binary(drop: "DataDROP"): 
-   """
-   Load binary 
-   """
-   buf = io.BytesIO()
-   desc = drop.open()
-   read = True
-   while read:
-       data = drop.read(desc)
-       if data: 
-           buf.write(data)
-           drop.close(desc)
-           return buf.getvalue().decode()
-        
-       return 0
+
+def load_binary(drop: "DataDROP"):
+    """
+    Load binary
+    """
+    buf = io.BytesIO()
+    desc = drop.open()
+    read = True
+    while read:
+        data = drop.read(desc)
+        if data:
+            buf.write(data)
+            drop.close(desc)
+            return buf.getvalue().decode()
+
+        return 0
+
 
 def save_binary(drop: "DataDROP", data: bytes):
     """
-    Save binary 
+    Save binary
     """
     bytes_data = io.BytesIO(data)
     dropio = drop.getIO()
     dropio.open(OpenMode.OPEN_WRITE)
-    dropio.write(bytes_data)
-    dropio.close()
+    dropio.write(bytes_data.getbuffer())
+    dropio.close()

daliuge-engine/dlg/named_port_utils.py

@@ -12,7 +12,7 @@
 class DropParser(Enum):
     RAW = "raw"
     PICKLE = "pickle"
-    EVAL = "eval" 
+    EVAL = "eval"
     NPY = "npy"
     DILL = "dill"
     # JSON = "json"
@@ -68,9 +68,7 @@ def serialize_applicationArgs(applicationArgs, prefix="--", separator=" "):
     positional arguments and one for kw arguments that can be used to construct
     the final command line.
     """
-    applicationArgs = clean_applicationArgs(
-        applicationArgs
-    )
+    applicationArgs = clean_applicationArgs(applicationArgs)
     pargs = []
     kwargs = {}
     for name, vdict in applicationArgs.items():
@@ -144,7 +142,7 @@ def identify_named_ports(
         if value is None:
             value = ""  # make sure we are passing NULL drop events
         if key in positionalArgs:
-            encoding = DropParser(positionalPortArgs[key]['encoding'])
+            encoding = DropParser(positionalPortArgs[key]["encoding"])
             parser = get_port_reader_function(encoding)
             if parser:
                 logger.debug("Reading from port using %s", parser.__repr__())
@@ -156,7 +154,7 @@ def identify_named_ports(
             if addPositionalToKeyword:
                 keywordPortArgs.update({key: positionalPortArgs[key]})
         elif key in keywordArgs:
-            encoding = DropParser(keywordArgs[key]['encoding'])
+            encoding = DropParser(keywordArgs[key]["encoding"])
             parser = get_port_reader_function(encoding)
             if parser:
                 logger.debug("Reading from port using %s", parser.__repr__())
@@ -299,12 +297,14 @@ def replace_named_ports(
     keywordArgs = _get_args(appArgs, positional=False)
 
     # Extract values from dictionaries - "encoding" etc. are irrelevant
-    appArgs = {arg: subdict['value'] for arg, subdict in appArgs.items()}
-    positionalArgs = {arg: subdict['value'] for arg, subdict in positionalArgs.items()}
-    keywordArgs = {arg: subdict['value'] for arg, subdict in keywordArgs.items()}
-    keywordPortArgs = {arg: subdict['value'] for arg, subdict in keywordPortArgs.items()}
- 
-     # Construct the final keywordArguments and positionalPortArguments
+    appArgs = {arg: subdict["value"] for arg, subdict in appArgs.items()}
+    positionalArgs = {arg: subdict["value"] for arg, subdict in positionalArgs.items()}
+    keywordArgs = {arg: subdict["value"] for arg, subdict in keywordArgs.items()}
+    keywordPortArgs = {
+        arg: subdict["value"] for arg, subdict in keywordPortArgs.items()
+    }
+
+    # Construct the final keywordArguments and positionalPortArguments
     for k, v in keywordPortArgs.items():
         if v not in [None, ""]:
             keywordArgs.update({k: v})
@@ -376,8 +376,10 @@ def _get_args(appArgs, positional=False):
     Separate out the arguments dependening on if we want positional or keyword style
     """
     args = {
-        arg: {"value": appArgs[arg]["value"], 
-              "encoding": appArgs[arg].get("encoding", "dill")}
+        arg: {
+            "value": appArgs[arg]["value"],
+            "encoding": appArgs[arg].get("encoding", "dill"),
+        }
         for arg in appArgs
         if (appArgs[arg]["positional"] == positional)
     }
@@ -386,6 +388,7 @@ def _get_args(appArgs, positional=False):
     logger.debug("%s arguments: %s", argType, args)
     return args
 
+
 def get_port_reader_function(input_parser: DropParser):
     """
     Return the function used to read input from a named port
@@ -413,7 +416,7 @@ def optionalEval(x):
     elif input_parser is DropParser.DILL:
         reader = drop_loaders.load_dill
     elif input_parser is DropParser.BINARY:
-         reader = drop_loaders.load_binary
+        reader = drop_loaders.load_binary
     else:
         raise ValueError(input_parser.__repr__())
     return reader