Merge pull request #75 from oele-isis-vanderbilt/dev_branch

Bug Patch

Author: edavalosanaya

Dockerfile

@@ -17,4 +17,7 @@ RUN ls
 
 # Install ChimeraPy
 RUN python3 -m pip install --upgrade pip
-RUN cd ChimeraPy && python3 -m pip install '.[test]'
+RUN cd ChimeraPy && python3 -m pip install '.[test]' && cd ..
+
+# For a certain test, remove the mock
+RUN rm -r ChimeraPy/test/mock/test_package

README.md

@@ -1,4 +1,4 @@
-![Logo](docs/_static/logo/chimerapy_logo_with_name_theme_blue.png)
+![Logo](https://user-images.githubusercontent.com/40870026/204550212-a6e1b7c2-194b-4554-ab42-f5e456c6f402.png)
 
 [![PyPI](https://img.shields.io/pypi/v/chimerapy)](https://pypi.org/project/chimerapy/) [![Coverage Status](https://coveralls.io/repos/github/oele-isis-vanderbilt/ChimeraPy/badge.svg?branch=main)](https://coveralls.io/github/oele-isis-vanderbilt/ChimeraPy?branch=main) ![](https://img.shields.io/github/workflow/status/oele-isis-vanderbilt/ChimeraPy/Test) ![](https://img.shields.io/github/license/oele-isis-vanderbilt/ChimeraPy) ![](https://img.shields.io/badge/style-black-black)
 * [Docs](https://oele-isis-vanderbilt.github.io/ChimeraPy)

chimerapy/__init__.py

@@ -1,45 +1,10 @@
-# Setup the logging for the library
-# References:
-# https://docs.python-guide.org/writing/logging/
-# https://stackoverflow.com/questions/13649664/how-to-use-logging-with-pythons-fileconfig-and-configure-the-logfile-filename
-import logging.config
+# Meta data
+__version__ = "0.0.7"
 
-LOGGING_CONFIG = {
-    "version": 1,
-    "disable_existing_loggers": True,
-    "formatters": {
-        "standard": {
-            "format": "%(asctime)s [%(levelname)s] %(name)s: %(message)s",
-            "datefmt": "%Y-%m-%d %H:%M:%S",
-        },
-    },
-    "handlers": {
-        "console": {
-            "level": "INFO",
-            "formatter": "standard",
-            "class": "logging.StreamHandler",
-            "stream": "ext://sys.stdout",  # Default is stderr
-        },
-        "datagram": {
-            "level": "INFO",
-            "formatter": "standard",
-            "class": "logging.handlers.DatagramHandler",
-            "host": "127.0.0.1",
-            "port": 5555,
-        },
-    },
-    "loggers": {
-        "chimerapy": {
-            "handlers": ["console"],
-            "level": "INFO",
-            "propagate": True,
-        },
-        "subprocess": {"handlers": ["datagram"], "level": "INFO", "propagate": True},
-    },
-}
+# Package Setup
+from . import _logger
 
-# Setup the logging configuration
-logging.config.dictConfig(LOGGING_CONFIG)
+_logger.setup()
 
 # Interal Imports
 from .manager import Manager
@@ -58,3 +23,6 @@
 
 # Then define the entry points
 from . import entry
+
+# Debugging tools
+from ._debug import debug

chimerapy/_debug.py

@@ -0,0 +1,21 @@
+# Built-in Import
+from typing import List, Optional
+import logging
+import os
+
+# Internal Imports
+from ._logger import LOGGING_CONFIG
+
+
+def debug(loggers: Optional[List[str]] = None):
+
+    # Not provided, then get all
+    if type(loggers) == type(None):
+        loggers = [x for x in LOGGING_CONFIG["loggers"]]
+
+    assert loggers != None
+
+    # Change env variable and configurations
+    os.environ["CHIMERAPY_DEBUG_LOGGERS"] = os.pathsep.join(loggers)
+    for logger_name in loggers:
+        logging.getLogger(logger_name).setLevel(logging.DEBUG)

chimerapy/_logger.py

@@ -0,0 +1,68 @@
+# Setup the logging for the library
+# References:
+# https://docs.python-guide.org/writing/logging/
+# https://stackoverflow.com/questions/13649664/how-to-use-logging-with-pythons-fileconfig-and-configure-the-logfile-filename
+import logging.config
+import os
+
+LOGGING_CONFIG = {
+    "version": 1,
+    "disable_existing_loggers": True,
+    "formatters": {
+        "standard": {
+            "format": "%(asctime)s [%(levelname)s] %(name)s: %(message)s",
+            "datefmt": "%Y-%m-%d %H:%M:%S",
+        },
+    },
+    "handlers": {
+        "console": {
+            "level": "INFO",
+            "formatter": "standard",
+            "class": "logging.StreamHandler",
+            "stream": "ext://sys.stdout",  # Default is stderr
+        },
+        "datagram": {
+            "level": "DEBUG",
+            "formatter": "standard",
+            "class": "logging.handlers.DatagramHandler",
+            "host": "127.0.0.1",
+            "port": 5555,
+        },
+    },
+    "loggers": {
+        "chimerapy": {
+            "handlers": ["console"],
+            "level": "INFO",
+            "propagate": True,
+        },
+        "chimerapy-networking": {
+            "handlers": ["console"],
+            "level": "INFO",
+            "propagate": True,
+        },
+        "chimerapy-subprocess": {
+            "handlers": ["datagram"],
+            "level": "INFO",
+            "propagate": True,
+        },
+    },
+}
+
+# Setup the logging configuration
+def setup():
+
+    # Setting up the configureation
+    logging.config.dictConfig(LOGGING_CONFIG)
+
+
+def getLogger(name: str):
+
+    # Get the logging
+    logger = logging.getLogger(name)
+
+    # Ensure that the configuration is set
+    debug_loggers = os.environ.get("CHIMERAPY_DEBUG_LOGGERS", "").split(os.pathsep)
+    if name in debug_loggers:
+        logger.setLevel(logging.DEBUG)
+
+    return logger

chimerapy/client.py

@@ -1,5 +1,5 @@
 # Built-in Imports
-from typing import Literal, Callable, Dict
+from typing import Literal, Callable, Dict, Any
 import os
 import logging
 import collections
@@ -18,10 +18,12 @@
 # Third-party Imports
 
 # Internal Imports
+from . import socket_handling as sh
 from .utils import create_payload, decode_payload
 from . import enums
+from . import _logger
 
-logger = logging.getLogger("chimerapy")
+logger = _logger.getLogger("chimerapy-networking")
 
 
 class Client(threading.Thread):
@@ -55,6 +57,8 @@ def __init__(
 
         # Creating tempfolder to host items
         self.tempfolder = pathlib.Path(tempfile.mkdtemp())
+        self.handlers.update({enums.FILE_TRANSFER_START: self.receive_file})
+        self.file_transfer_records = collections.defaultdict(dict)
 
         # Create the socket and connect
         try:
@@ -115,69 +119,32 @@ def run(self):
         # Continue checking for messages from client until not running
         while self.is_running.is_set():
 
-            # Check if the socket is closed
-            if self.socket.fileno() == -1:
-                break
-
-            # Get message while not blocking
+            # Monitor the socket
             try:
-                # Get the data
-                bs = self.socket.recv(8)
-
-                # If null, skip
-                if bs == b"":
-                    continue
-
-                # Get the length
-                (length,) = struct.unpack(">Q", bs)
-                data = b""
-
-                # Use the length to get the entire message
-                while len(data) < int(length):
-                    to_read = int(length) - len(data)
-                    data += self.socket.recv(4096 if to_read > 4096 else to_read)
-
-            except socket.timeout:
-                continue
-
-            # If end, stop it
-            if data == b"":
+                success, msg = sh.monitor(self.name, self.socket)
+            except (ConnectionResetError, ConnectionAbortedError):
                 break
 
-            # Decode the message so we can process it
-            msg = decode_payload(data)
-
             # Process the msg
-            self.process_msg(msg)
+            if success:
+                self.process_msg(msg)
 
         self.socket.close()
 
     def send(self, msg: Dict, ack: bool = False):
 
-        # Create an uuid to track the message
-        msg_uuid = str(uuid.uuid4())
-
-        # Convert msg data to bytes
-        msg_bytes, msg_length = create_payload(
-            type=self.sender_msg_type,
-            signal=msg["signal"],
-            data=msg["data"],
+        # Sending the data
+        success, msg_uuid = sh.send(
+            name=self.name,
+            s=self.socket,
+            msg=msg,
+            sender_msg_type=self.sender_msg_type,
             ack=ack,
-            provided_uuid=msg_uuid,
         )
 
-        # Send the message
-        try:
-            self.socket.sendall(msg_length + msg_bytes)
-            logger.debug(f"{self}: send {msg['signal']}")
-        except socket.timeout:
-            logger.debug(f"{self}: Socket Timeout: skipping")
-            return
-        except:
-            logger.debug(
-                f"{self}: Broken Pipe Error, handled for {msg['signal']}", exc_info=True
-            )
-            return
+        # If not successful, skip ACK
+        if not success:
+            return None
 
         # If requested ACK, wait
         if ack:
@@ -198,61 +165,40 @@ def send(self, msg: Dict, ack: bool = False):
 
                     miss_counter += 1
 
-    def send_folder(self, name: str, dir: pathlib.Path, buffersize: int = 4096):
-
-        assert (
-            dir.is_dir() and dir.exists()
-        ), f"Sending {dir} needs to be a folder that exists."
-
-        # First, we need to archive the folder into a zip file
-        format = "zip"
-        shutil.make_archive(str(dir), format, dir.parent, dir.name)
-        zip_file = dir.parent / f"{dir.name}.{format}"
-
-        # Relocate zip to the tempfolder
-        temp_zip_file = self.tempfolder / f"_{zip_file.name}"
-        shutil.move(zip_file, temp_zip_file)
-
-        # Get information about the filesize
-        filesize = os.path.getsize(temp_zip_file)
-        max_num_steps = math.ceil(filesize / buffersize)
-
-        # Now start the process of sending content to the server
-        # First, we send the message inciting file transfer
-        init_msg = {
-            "type": enums.WORKER_MESSAGE,
-            "signal": enums.FILE_TRANSFER_START,
-            "data": {
-                "name": name,
-                "filename": f"{dir.name}.{format}",
-                "filesize": filesize,
-                "buffersize": buffersize,
-                "max_num_steps": max_num_steps,
-            },
-        }
-        self.send(init_msg)
-        logger.debug(f"{self}: sent file transfer initialization")
-
-        # Having counter tracking number of messages
-        msg_counter = 1
-
-        with open(temp_zip_file, "rb") as f:
-            while True:
-
-                # Read the data to be sent
-                data = f.read(buffersize)
-                if not data:
-                    break
+    def receive_file(self, msg: Dict[str, Any]):
+
+        # Obtain the file
+        success, sender_name, dst_filepath = sh.file_transfer_receive(
+            name=self.name, s=self.socket, msg=msg, tempfolder=self.tempfolder
+        )
+
+        # Create a record of the files transferred and from whom
+        if success:
+            self.file_transfer_records[sender_name][dst_filepath.name] = dst_filepath
 
-                logger.debug(
-                    f"{self}: file transfer, step {msg_counter}/{max_num_steps}"
-                )
+    def send_file(self, name: str, filepath: pathlib.Path):
 
-                # Send the data
-                self.socket.sendall(data)
-                msg_counter += 1
+        assert filepath.exists() and filepath.is_file()
 
-        logger.debug(f"{self}: finished file transfer")
+        sh.send_file(
+            sender_name=name,
+            sender_msg_type=self.sender_msg_type,
+            s=self.socket,
+            filepath=filepath,
+            buffersize=4096,
+        )
+
+    def send_folder(self, name: str, folderpath: pathlib.Path):
+
+        assert folderpath.exists() and folderpath.is_dir()
+
+        sh.send_folder(
+            name=name,
+            s=self.socket,
+            dir=folderpath,
+            tempfolder=self.tempfolder,
+            sender_msg_type=self.sender_msg_type,
+        )
 
     def shutdown(self):
 

chimerapy/data_handlers.py

@@ -4,7 +4,9 @@
 import threading
 import queue
 
-logger = logging.getLogger("chimerapy")
+from . import _logger
+
+logger = _logger.getLogger("chimerapy")
 
 # Internal Imports
 from . import enums