feat: json logging support

README.md

@@ -43,6 +43,11 @@ the clutter
 There's also an option to simply print all requests or send them to Discord. You can use those 
 if you don't like the UI.
 
+Set `output = "json"` to emit one decoded-JSON object per RPC (JSONL) to stdout instead of a
+rendered view — each line carries `timestamp`, `rpc_id`, `rpc_status`, `rpc_handle`, and a `protos`
+list where every request/response is the **decoded** proto as JSON (no base64 blobs). Pipe it
+straight to a file or `jq`, e.g. `trafficlight run > traffic.jsonl`.
+
 ## TrafficLight CLI
 
 - `trafficlight run` to run the TUI

config.example.toml

@@ -2,10 +2,11 @@ host = "0.0.0.0"
 port = 3335
 # Host and Port for the Receiver
 
-output = "ui/print/discord"
+output = "ui/print/discord/json"
 # ui      = cool interactive ui
 # print   = simple log
 # discord = discord webhooks
+# json    = one decoded-JSON object per rpc (JSONL) to stdout — pipe to a file / jq
 
 webhook = ""
 # your webhook url in case you set "discord" for output

tests/test_json_output.py

@@ -0,0 +1,91 @@
+from __future__ import annotations
+
+import asyncio
+import base64
+import io
+import json
+import unittest
+from contextlib import redirect_stdout
+
+from trafficlight import protos
+from trafficlight.output.json_ import JsonOutput
+from trafficlight.proto_utils.proto import Proto, Request
+
+
+class MessageToJsonTests(unittest.TestCase):
+    def test_decoded_request_reports_type_and_dict_data(self) -> None:
+        # method 106 = GET_MAP_OBJECTS; empty bytes decode to an all-defaults GetMapObjectsProto.
+        obj = Request(106, "").to_json_obj()
+        self.assertEqual(obj["type"], "GetMapObjectsProto")
+        self.assertTrue(obj["decoded"])
+        self.assertIsInstance(obj["data"], dict)
+        # No base64 blob — the raw bytes never appear in the JSON object.
+        self.assertNotIn("payload", obj)
+
+    def test_unknown_method_falls_back_to_blackbox(self) -> None:
+        # Unknown method id => no proto name; still decode the wire bytes generically.
+        # field 1 (varint) = 5  ->  b"\x08\x05"
+        obj = Request(9_999_999, "CAU=").to_json_obj()
+        self.assertIsNone(obj["type"])
+        self.assertFalse(obj["decoded"])
+        self.assertIsInstance(obj["data"], dict)
+        self.assertEqual(obj["data"].get("1"), 5)
+
+
+class ProtoToJsonTests(unittest.TestCase):
+    def test_proxy_rpc_nests_the_inner_decoded_method(self) -> None:
+        # method 5012 wraps a real method inside ProxyRequestProto.action + .payload.
+        inner_req = protos.GetMapObjectsProto(cell_id=[7]).SerializeToString()
+        inner_resp = protos.GetMapObjectsOutProto().SerializeToString()
+        outer_req = protos.ProxyRequestProto(action=106, payload=inner_req).SerializeToString()
+        outer_resp = protos.ProxyResponseProto(payload=inner_resp).SerializeToString()
+
+        proto = Proto(
+            rpc_id=1,
+            method_value=5012,
+            raw_request=base64.b64encode(outer_req).decode(),
+            raw_response=base64.b64encode(outer_resp).decode(),
+        )
+        obj = proto.to_json_obj()
+
+        self.assertEqual(obj["method"], 5012)
+        self.assertEqual(obj["request"]["type"], "ProxyRequestProto")
+        self.assertIn("proxy", obj)
+        proxy = obj["proxy"]
+        self.assertEqual(proxy["method"], 106)
+        self.assertEqual(proxy["request"]["type"], "GetMapObjectsProto")
+        self.assertEqual(proxy["request"]["data"], {"cell_id": ["7"]})
+
+    def test_non_proxy_rpc_has_no_proxy_key(self) -> None:
+        obj = Proto(rpc_id=1, method_value=106, raw_request="", raw_response="").to_json_obj()
+        self.assertNotIn("proxy", obj)
+
+
+class JsonOutputEnvelopeTests(unittest.TestCase):
+    def test_add_record_emits_one_jsonl_object_with_rotom_fields(self) -> None:
+        proto = Proto(rpc_id=42, method_value=106, raw_request="", raw_response="")
+        out = JsonOutput()
+
+        buf = io.StringIO()
+        with redirect_stdout(buf):
+            asyncio.run(out.add_record(rpc_id=42, rpc_status=1, protos=[proto], rpc_handle=7))
+
+        lines = [ln for ln in buf.getvalue().splitlines() if ln.strip()]
+        self.assertEqual(len(lines), 1, "exactly one JSONL record per rpc envelope")
+        rec = json.loads(lines[0])
+
+        self.assertEqual(rec["rpc_id"], 42)
+        self.assertEqual(rec["rpc_status"], 1)
+        self.assertEqual(rec["rpc_handle"], 7)
+        self.assertIn("timestamp", rec)
+
+        self.assertEqual(len(rec["protos"]), 1)
+        p = rec["protos"][0]
+        self.assertEqual(p["method"], 106)
+        self.assertEqual(p["method_name"], "METHOD_GET_MAP_OBJECTS")
+        self.assertEqual(p["request"]["type"], "GetMapObjectsProto")
+        self.assertIn("response", p)
+
+
+if __name__ == "__main__":
+    unittest.main()

trafficlight/config.py

@@ -11,6 +11,7 @@ class Output(Enum):
     UI = "ui"
     PRINT = "print"
     DISCORD = "discord"
+    JSON = "json"
 
 
 class Config(BaseModel):

trafficlight/output/__init__.py

@@ -1,6 +1,7 @@
 from trafficlight.config import Output as _OutputType
 from .base import BaseOutput
 from .discord import DiscordOutput
+from .json_ import JsonOutput
 from .print_ import PrintOutput
 from .ui import UiOutput
 
@@ -10,5 +11,7 @@ def get_output(output_type: _OutputType) -> BaseOutput:
         return PrintOutput()
     elif output_type == _OutputType.DISCORD:
         return DiscordOutput()
+    elif output_type == _OutputType.JSON:
+        return JsonOutput()
 
     return UiOutput()

trafficlight/output/json_.py

@@ -0,0 +1,35 @@
+from __future__ import annotations
+
+import json
+from datetime import datetime
+from typing import TYPE_CHECKING
+
+from .base import BaseOutput
+
+if TYPE_CHECKING:
+    from trafficlight.proto_utils.proto import Proto
+
+
+class JsonOutput(BaseOutput):
+    """Emit one JSON object per RPC envelope (JSONL) to stdout.
+
+    Same information the `print` output shows, but as machine-readable JSON instead of a rendered
+    table — pipe it to a file (`trafficlight run > traffic.jsonl`) and query it with jq/scripts.
+
+    The record mirrors the Rotom-style envelope the receiver ingests (rpc id / status / handle +
+    a `protos` list keyed by `method`), except the request/response are the DECODED protos as JSON
+    objects — never the raw base64 payloads.
+    """
+
+    async def start(self) -> None:
+        pass
+
+    async def add_record(self, rpc_id: int, rpc_status: int, protos: list[Proto], rpc_handle: int | None = None) -> None:
+        record = {
+            "timestamp": datetime.now().isoformat(),
+            "rpc_id": rpc_id,
+            "rpc_status": rpc_status,
+            "rpc_handle": rpc_handle,
+            "protos": [proto.to_json_obj() for proto in protos],
+        }
+        print(json.dumps(record, ensure_ascii=False))

trafficlight/proto_utils/proto.py

@@ -8,6 +8,7 @@
 from blackboxprotobuf.lib.api import decode_message, _json_safe_transform
 from google.protobuf import text_format, descriptor
 from google.protobuf.internal.enum_type_wrapper import EnumTypeWrapper
+from google.protobuf.json_format import MessageToDict
 from google.protobuf.message import Message as ProtobufMessage
 
 from trafficlight import protos
@@ -122,6 +123,24 @@ def to_string(self, one_line: bool = True) -> str:
         else:
             return text_format.MessageToString(self.payload, as_one_line=one_line)
 
+    def to_json_obj(self) -> dict:
+        """A JSON-safe view of this message: the proto decoded to a dict (never a base64 blob).
+
+        - decoded proto  -> {"type": <ProtoName>, "decoded": True,  "data": {..}}
+        - undecodable    -> {"type": <name|None>, "decoded": False, "data": <blackbox|None>}
+
+        `data` holds the actual field values, keyed by proto field name (blackbox falls back to
+        wire field numbers). Genuine `bytes` sub-fields still render base64 — JSON has no bytes type —
+        but the message envelope itself is never dumped as one opaque base64 string.
+        """
+        if self.payload is not None:
+            return {
+                "type": self.name,
+                "decoded": True,
+                "data": MessageToDict(self.payload, preserving_proto_field_name=True),
+            }
+        return {"type": self.name, "decoded": False, "data": self.blackbox}
+
 
 class Request(Message):
     messages = MESSAGES
@@ -155,6 +174,18 @@ def messages(self) -> Iterable[Message]:
         yield self.request
         yield self.response
 
+    def to_json_obj(self) -> dict:
+        """This RPC method's request+response as decoded JSON, plus a nested proxy when present."""
+        obj = {
+            "method": self.method_value,
+            "method_name": self.method_name,
+            "request": self.request.to_json_obj(),
+            "response": self.response.to_json_obj(),
+        }
+        if self.proxy is not None:
+            obj["proxy"] = self.proxy.to_json_obj()
+        return obj
+
     @staticmethod
     def get_message_name(messages: dict, value: int) -> str | None:
         try: