feat: adds --json to ais-decode

Author: M0r13n

.flake8

@@ -4,4 +4,4 @@ ignore = E501
 show_source = True
 statistics = True
 count = True
-exclude = .direnv,.git,__pycache__,venv,.eggs
+exclude = .direnv,.git,__pycache__,venv,.eggs,.venv

pyais/main.py

@@ -1,7 +1,9 @@
 import argparse
+import json
 import sys
-from typing import List, Tuple, Type, Any, Union
+from typing import List, Tuple, Type, Any, Union, TextIO, cast
 
+from pyais.messages import AISJSONEncoder
 from pyais.stream import ByteStream, TCPConnection, UDPReceiver, BinaryIOStream
 
 SOCKET_OPTIONS: Tuple[str, str] = ('udp', 'tcp')
@@ -10,140 +12,223 @@
 INVALID_CHECKSUM_ERROR = 21
 
 
-def arg_parser() -> argparse.ArgumentParser:
-    """Create a new ArgumentParser instance that serves as a entry point to the pyais application.
-    All possible commandline options and parameters must be defined here.
-    The goal is to create a grep-like interface:
-        Usage: ais-decode [OPTION]... PATTERNS [FILE]...
-    """
-    main_parser: argparse.ArgumentParser = argparse.ArgumentParser(
+def create_parser() -> argparse.ArgumentParser:
+    """Create and configure the argument parser for the AIS decoder application."""
+    parser = argparse.ArgumentParser(
         prog="ais-decode",
-        description="AIS message decoding. 100% pure Python."
-                    "Supports AIVDM/AIVDO messages. Supports single messages, files and TCP/UDP sockets.rst.",
+        description="Decode NMEA (AIVDM/AIVDO) AIS messages."
+                    "Supports single messages, files and TCP/UDP sockets.",
+        epilog="Examples:\n"
+               "  ais-decode -f input.txt                    # Decode from file\n"
+               "  ais-decode -j < input.txt                  # Decode from stdin with JSON output\n"
+               "  ais-decode socket localhost 5000           # Decode from UDP socket\n"
+               "  ais-decode single '!AIVDM,1,1,,A,13HOI:0P0000VOHLCnHQKwvL05Ip,0*23'\n"
+               "  nc 153.44.253.27 5631 | ais-decode --json | jq",
+        formatter_class=argparse.RawDescriptionHelpFormatter
     )
-    sub_parsers = main_parser.add_subparsers()
-
-    # Modes
-    # Currently three mutual exclusive modes are supported: TCP/UDP / file / single messages as arguments
-    # By default the program accepts input from STDIN
-    # Optional subparsers server as subcommands that handle socket connections and file reading
-    main_parser.add_argument(
-        '-f',
-        '--file',
+
+    # Global options
+    parser.add_argument(
+        '-j', '--json',
+        dest="json",
+        action='store_true',
+        help="Output messages in JSON format"
+    )
+
+    parser.add_argument(
+        '-o', '--out-file',
+        dest="out_file",
+        type=argparse.FileType("w"),
+        default=sys.stdout,
+        help="Output file (default: stdout)"
+    )
+
+    # Create subparsers for different input modes
+    subparsers = parser.add_subparsers(
+        title="Input modes",
+        description="Choose input source",
+        dest="mode",
+        required=False
+    )
+
+    # File input mode (also default for stdin)
+    parser.add_argument(
+        '-f', '--file',
         dest="in_file",
-        nargs="?",
         type=argparse.FileType("rb"),
-        default=None
+        nargs='?',
+        help="Input file (default: stdin if no subcommand specified)"
     )
+    parser.set_defaults(func=decode_from_file)
 
-    main_parser.set_defaults(func=decode_from_file)
-
-    socket_parser = sub_parsers.add_parser('socket')
+    # Socket input mode
+    socket_parser = subparsers.add_parser(
+        'socket',
+        help="Decode from TCP/UDP socket"
+    )
     socket_parser.add_argument(
         'destination',
-        type=str,
+        help="Hostname or IP address"
     )
     socket_parser.add_argument(
         'port',
-        type=int
+        type=int,
+        help="Port number"
     )
     socket_parser.add_argument(
-        '-t',
-        '--type',
+        '-t', '--type',
         default='udp',
-        nargs='?',
-        choices=SOCKET_OPTIONS
+        choices=SOCKET_OPTIONS,
+        help="Socket type (default: udp)"
     )
-
     socket_parser.set_defaults(func=decode_from_socket)
 
-    # Optional a single message can be decoded
-    # This has the highest precedence and will overwrite all other settings
-    single_msg_parser = sub_parsers.add_parser('single')
-    single_msg_parser.add_argument(
+    # Single message mode
+    single_parser = subparsers.add_parser(
+        'single',
+        help="Decode single message(s)"
+    )
+    single_parser.add_argument(
         'messages',
         nargs='+',
-        default=[]
+        help="One or more NMEA messages to decode"
     )
-    single_msg_parser.set_defaults(func=decode_single)
+    single_parser.set_defaults(func=decode_single)
 
-    # Output
-    # By default the application writes it output to STDOUT - but this can be any file
-    main_parser.add_argument(
-        "-o",
-        "--out-file",
-        dest="out_file",
-        type=argparse.FileType("w"),
-        default=sys.stdout
-    )
-
-    return main_parser
+    return parser
 
 
 def print_error(*args: Any, **kwargs: Any) -> None:
-    """Wrapper around the default print function that writes to STDERR."""
-    print(*args, **kwargs, file=sys.stdout)
+    """Print error messages to stderr."""
+    print(*args, **kwargs, file=sys.stderr)
+
+
+def output_message(msg: Any, out_file: TextIO, as_json: bool) -> None:
+    """Output a decoded message in the requested format."""
+    decoded = msg.decode()
+
+    if as_json:
+        json.dump(decoded.asdict(), out_file, cls=AISJSONEncoder)
+        out_file.write('\n')  # Add newline for readability
+    else:
+        out_file.write(str(decoded) + '\n')
+
+    out_file.flush()  # Ensure immediate output for streaming scenarios
 
 
 def decode_from_socket(args: argparse.Namespace) -> int:
-    """Connect a socket and start decoding."""
-    t: str = args.type
+    """Connect to a socket and decode incoming AIS messages."""
     stream_cls: Type[Union[UDPReceiver, TCPConnection]]
-    if t == "udp":
+
+    if args.type == "udp":
         stream_cls = UDPReceiver
-    elif t == "tcp":
+    elif args.type == "tcp":
         stream_cls = TCPConnection
     else:
-        raise ValueError("args.type must be either TCP or UDP.")
+        print_error(f"Invalid socket type: {args.type}")
+        return 1
+
+    try:
+        with stream_cls(args.destination, args.port) as stream:
+            print_error(f"Connected to {args.type.upper()} {args.destination}:{args.port}")
+
+            for msg in stream:
+                try:
+                    output_message(msg, args.out_file, args.json)
+
+                    if not msg.is_valid:
+                        print_error(f"WARNING: Invalid checksum for message: {msg}")
+
+                except Exception as e:
+                    print_error(f"ERROR decoding message: {e}")
+                    continue
+
+    except KeyboardInterrupt:
+        print_error("\nConnection closed by user")
+        return 0
+    except Exception as e:
+        print_error(f"ERROR: Failed to connect to {args.destination}:{args.port} - {e}")
+        return 1
 
-    with stream_cls(args.destination, args.port) as s:
-        try:
-            for msg in s:
-                decoded_message = msg.decode()
-                print(decoded_message, file=args.out_file)
-        except KeyboardInterrupt:
-            # Catch KeyboardInterrupts in order to close the socket and free associated resources
-            return 0
     return 0
 
 
 def decode_single(args: argparse.Namespace) -> int:
-    """Decode a list of messages."""
-    messages: List[str] = args.messages
-    messages_as_bytes: List[bytes] = [msg.encode() for msg in messages if isinstance(msg, str)]
-    for msg in ByteStream(messages_as_bytes):
-        print(msg.decode(), file=args.out_file)
-        if not msg.is_valid:
-            print_error("WARNING: Checksum invalid")
-    return 0
+    """Decode a list of single messages."""
+    messages_as_bytes: List[bytes] = [msg.encode() for msg in args.messages]
+    error_count = 0
+
+    for i, msg in enumerate(ByteStream(messages_as_bytes)):
+        try:
+            output_message(msg, args.out_file, args.json)
+
+            if not msg.is_valid:
+                print_error(f"WARNING: Invalid checksum for message {i + 1}: {args.messages[i]}")
+                error_count += 1
+
+        except Exception as e:
+            print_error(f"ERROR decoding message {i + 1}: {e}")
+            error_count += 1
+
+    return INVALID_CHECKSUM_ERROR if error_count > 0 else 0
 
 
 def decode_from_file(args: argparse.Namespace) -> int:
-    """Decode messages from a file-like object."""
-    if not args.in_file:
-        # This is needed, because it is not possible to open STDOUT in binary mode (it is text mode by default)
-        # Therefore it is None by default and we interact with the buffer directly
-        file = sys.stdin.buffer
-    else:
-        # If the file is not None, then it was opened during argument parsing
-        file = args.in_file
+    """Decode messages from a file or stdin."""
+    # Use stdin buffer if no file specified
+    file_obj = sys.stdin.buffer if args.in_file is None else args.in_file
+
+    try:
+        with BinaryIOStream(file_obj) as stream:
+            message_count = 0
+            error_count = 0
+
+            for msg in stream:
+                try:
+                    output_message(msg, args.out_file, args.json)
+                    message_count += 1
+
+                    if not msg.is_valid:
+                        print_error(f"WARNING: Invalid checksum at line {message_count}")
+                        error_count += 1
+
+                except Exception as e:
+                    print_error(f"ERROR decoding message at line {message_count + 1}: {e}")
+                    error_count += 1
+                    continue
+
+    except KeyboardInterrupt:
+        print_error(f"\nProcessing interrupted. Decoded {message_count} messages.")
+        return 0
+    except Exception as e:
+        print_error(f"ERROR reading input: {e}")
+        return 1
+
+    if args.in_file:
+        print_error(f"Processed {message_count} messages ({error_count} errors)")
 
-    with BinaryIOStream(file) as s:
-        try:
-            for msg in s:
-                decoded_message = msg.decode()
-                print(decoded_message, file=args.out_file)
-        except KeyboardInterrupt:
-            # Catch KeyboardInterrupts in order to close the file descriptor and free associated resources
-            return 0
     return 0
 
 
 def main() -> int:
-    main_parser = arg_parser()
-    namespace: argparse.Namespace = main_parser.parse_args()
-    exit_code: int = namespace.func(namespace)
-    return exit_code
+    """Main entry point for the AIS decoder application."""
+    parser = create_parser()
+    args = parser.parse_args()
+
+    # Validate arguments
+    if not hasattr(args, 'func'):
+        parser.print_help()
+        return 1
+
+    try:
+        return cast(int, args.func(args))
+    except Exception as e:
+        print_error(f"FATAL ERROR: {e}")
+        return 1
+    finally:
+        # Ensure output file is properly closed
+        if hasattr(args, 'out_file') and args.out_file != sys.stdout:
+            args.out_file.close()
 
 
 if __name__ == "__main__":

pyais/messages.py

@@ -67,13 +67,13 @@ def bit_field(width: int, d_type: typing.Type[typing.Any],
 ENUM_FIELDS = {'status', 'maneuver', 'epfd', 'ship_type', 'aid_type', 'station_type', 'txrx', 'interval'}
 
 
-class JSONEncoder(json.JSONEncoder):
+class AISJSONEncoder(json.JSONEncoder):
     """Custom JSON encoder to handle bytes objects"""
 
-    def default(self, obj: typing.Any) -> typing.Any:
-        if isinstance(obj, bytes):
-            return b64encode_str(obj)
-        return json.JSONEncoder.default(self, obj)
+    def default(self, o: typing.Any) -> typing.Any:
+        if isinstance(o, bytes):
+            return b64encode_str(o)
+        return json.JSONEncoder.default(self, o)
 
 
 class NMEASentenceFactory:
@@ -817,7 +817,7 @@ def asdict(self, enum_as_int: bool = False) -> typing.Dict[str, typing.Optional[
             return {slt: getattr(self, slt) for slt in self.__slots__}  # type: ignore
 
     def to_json(self) -> str:
-        return JSONEncoder(indent=4).encode(self.asdict())
+        return AISJSONEncoder(indent=4).encode(self.asdict())
 
 
 #