Command Line Interface

.gitignore

@@ -1,4 +1,8 @@
 # ---> Python
+
+# Configuration file
+*config.py
+
 # Byte-compiled / optimized / DLL files
 __pycache__/
 *.py[cod]

src/client.py

@@ -1,8 +1,9 @@
+import argparse
 import asyncio
 import collections
 import logging
 # import signal
-import sys
+import os
 
 
 class Client:
@@ -94,6 +95,34 @@ class Client:
             raise KeyboardInterrupt("Not working yet...")
 
 
+def get_action(action):
+    """Parse abbreviations for `action`."""
+    if not isinstance(action, str):
+        return
+    elif action.lower().startswith('r'):
+        return 'receive'
+    elif action.lower().startswith('s'):
+        return 'send'
+
+
+def get_file_path(path, action='receive'):
+    """Check that file `path` is correct and return it."""
+    if (
+            isinstance(path, str)
+            and action == 'send'
+            and os.path.isfile(path)
+    ):
+        return path
+    elif (
+            isinstance(path, str)
+            and action == 'receive'
+            and os.access(os.path.dirname(os.path.abspath(path)), os.W_OK)
+    ):
+        return path
+    elif path is not None:
+        logging.error(f"Invalid file: `{path}`")
+
+
 if __name__ == '__main__':
     log_formatter = logging.Formatter(
         "%(asctime)s [%(module)-15s %(levelname)-8s]     %(message)s",
@@ -107,31 +136,101 @@ if __name__ == '__main__':
     console_handler.setLevel(logging.DEBUG)
     root_logger.addHandler(console_handler)
 
-    if len(sys.argv) > 1:
+    # Parse command-line arguments
-        action = sys.argv[1]
+    cli_parser = argparse.ArgumentParser(description='Run client',
-    else:
+                                         allow_abbrev=False)
-        action = input("Do you want to (R)eceive or (S)end a file?\t\t")
+    cli_parser.add_argument('--host', type=str,
+                            default=None,
+                            required=False,
+                            help='server address')
+    cli_parser.add_argument('--port', type=int,
+                            default=None,
+                            required=False,
+                            help='server _port')
+    cli_parser.add_argument('--action', type=str,
+                            default=None,
+                            required=False,
+                            help='[S]end or [R]eceive')
+    cli_parser.add_argument('--path', type=str,
+                            default=None,
+                            required=False,
+                            help='File path')
+    cli_parser.add_argument('others',
+                            metavar='R or S',
+                            nargs='*',
+                            help='[S]end or [R]eceive (see `action`)')
+    args = vars(cli_parser.parse_args())
+    _host = args['host']
+    _port = args['port']
+    _action = get_action(args['action'])
+    _file_path = args['path']
 
-    action = (
+    # If _host and _port are not provided from command-line, try to import them
-        'send'
+    if _host is None:
-        if action.lower() == 's'
+        try:
-        else 'receive'
+            from config import host as _host
-    )
+        except ImportError:
-    if len(sys.argv) > 2:
+            _host = None
-        _file_path = sys.argv[2]
+    if _port is None:
-    else:
+        try:
-        _file_path = input(f"Enter file to {action}:\t\t\t\t\t\t")
+            from config import port as _port
+        except ImportError:
+            _port = None
+    # Take `s`, `r` etc. from command line as `_action`
+    if _action is None:
+        for arg in args['others']:
+            _action = get_action(arg)
+            if _action:
+                break
+    if _action is None:
+        try:
+            from config import action as _action
 
+            _action = get_action(_action)
+        except ImportError:
+            _action = None
+    if _file_path is None:
+        try:
+            from config import file_path as _file_path
+            _file_path = get_action(_file_path)
+        except ImportError:
+            _file_path = None
+
+    # If import fails, prompt user for _host or _port
+    while _host is None:
+        _host = input("Enter _host:\t\t\t\t\t\t")
+    while _port is None:
+        try:
+            _port = int(input("Enter _port:\t\t\t\t\t\t"))
+        except ValueError:
+            logging.info("Invalid _port. Enter a valid _port number!")
+            _port = None
+    while _action is None:
+        _action = get_action(
+            input("Do you want to (R)eceive or (S)end a file?\t\t")
+        )
+    while _file_path is None:
+        _file_path = get_file_path(
+            path=input(f"Enter file to {_action}:\t\t\t\t\t\t"),
+            action=_action
+        )
     loop = asyncio.get_event_loop()
     client = Client(
-        host='davte.it',
+        host=_host,
-        port=5000,
+        port=_port,
     )
-    # loop.add_signal_handler(signal.SIGINT, client.stop, loop)
     logging.info("Starting client...")
-    if action.lower() == 'send':
+    if _action == 'send':
-        loop.run_until_complete(client.run_sending_client(file_path=_file_path))
+        loop.run_until_complete(
+            client.run_sending_client(
+                file_path=_file_path
+            )
+        )
     else:
-        loop.run_until_complete(client.run_receiving_client(file_path=_file_path))
+        loop.run_until_complete(
+            client.run_receiving_client(
+                file_path=_file_path
+            )
+        )
     loop.close()
     logging.info("Stopped client")

src/server.py

@@ -1,3 +1,4 @@
+import argparse
 import asyncio
 import collections
 import logging
@@ -96,7 +97,6 @@ class Server:
             logging.info("Receiver is connecting...")
             await self.run_writer(writer=writer)
             logging.info("Outgoing transmission ended")
-            self.stop()
         return
 
     async def run_server(self):
@@ -131,10 +131,47 @@ if __name__ == '__main__':
     console_handler.setLevel(logging.DEBUG)
     root_logger.addHandler(console_handler)
 
+    # Parse command-line arguments
+    parser = argparse.ArgumentParser(description='Run server',
+                                     allow_abbrev=False)
+    parser.add_argument('--_host', type=str,
+                        default=None,
+                        required=False,
+                        help='server address')
+    parser.add_argument('--_port', type=int,
+                        default=None,
+                        required=False,
+                        help='server _port')
+    args = vars(parser.parse_args())
+    host = args['_host']
+    port = args['_port']
+
+    # If _host and _port are not provided from command-line, try to import them
+    if host is None:
+        try:
+            from config import host
+        except ImportError:
+            host = None
+    if port is None:
+        try:
+            from config import port
+        except ImportError:
+            port = None
+
+    # If import fails, prompt user for _host or _port
+    while host is None:
+        host = input("Enter _host:\t\t\t\t\t\t")
+    while port is None:
+        try:
+            port = int(input("Enter _port:\t\t\t\t\t\t"))
+        except ValueError:
+            logging.info("Invalid _port. Enter a valid _port number!")
+            port = None
+
     loop = asyncio.get_event_loop()
     server = Server(
-        host='davte.it',
+        host=host,
-        port=5000,
+        port=port,
     )
     logging.info("Starting file bridging server...")
     try: