Dev Tools: Source map decoder

Makefile

@@ -1,4 +1,4 @@
-UNITS = "@unit.abijson or @unit.abijson.byname or @unit.algod or @unit.algod.ledger_refactoring or @unit.applications or @unit.atc_method_args or @unit.atomic_transaction_composer or @unit.dryrun or @unit.dryrun.trace.application or @unit.feetest or @unit.indexer or @unit.indexer.ledger_refactoring or @unit.indexer.logs or @unit.offline or @unit.rekey or @unit.transactions.keyreg or @unit.responses or @unit.responses.231 or @unit.tealsign or @unit.transactions or @unit.transactions.payment or @unit.responses.unlimited_assets"
+UNITS = "@unit.abijson or @unit.abijson.byname or @unit.algod or @unit.algod.ledger_refactoring or @unit.applications or @unit.atc_method_args or @unit.atomic_transaction_composer or @unit.dryrun or @unit.dryrun.trace.application or @unit.feetest or @unit.indexer or @unit.indexer.ledger_refactoring or @unit.indexer.logs or @unit.offline or @unit.rekey or @unit.transactions.keyreg or @unit.responses or @unit.responses.231 or @unit.tealsign or @unit.transactions or @unit.transactions.payment or @unit.responses.unlimited_assets or @unit.sourcemap.pctoline"
 unit:
 	behave --tags=$(UNITS) tests -f progress2
 

algosdk/__init__.py

@@ -16,5 +16,6 @@
 from . import v2client
 from . import wallet
 from . import wordlist
+from . import source_map
 
 name = "algosdk"

algosdk/logic.py

@@ -2,6 +2,8 @@
 import json
 import os
 
+from typing import Tuple, Dict, List, Any
+
 from . import constants
 from . import error
 from . import encoding

algosdk/source_map.py

@@ -0,0 +1,97 @@
+from typing import Dict, Any, List, Tuple
+
+
+class SourceMap:
+    """
+    Decodes a VLQ-encoded source mapping between PC values and TEAL source code lines.
+    Spec available here: https://sourcemaps.info/spec.html
+
+    Args:
+        source_map (dict(str, Any)): source map JSON from algod
+        delimiter (str, optional): delimiter for mappings
+    """
+
+    def __init__(self, source_map: Dict[str, Any], delimiter: str = ";"):
+        self.delimter = delimiter
+
+        self.version: int = source_map["version"]
+        self.sources: List[str] = source_map["sources"]
+        self.mapping: str = source_map["mapping"]
+
+        pc_list = [
+            _decode_int_value(raw_val)
+            for raw_val in self.mapping.split(delimiter)
+        ]
+
+        # Initialize with 0,0 for pc/line
+        self.pc_to_line: Dict[int, int] = {0: 0}
+        self.line_to_pc: Dict[int, List[int]] = {0: [0]}
+
+        last_line = 0
+        for index, line_num in enumerate(pc_list):
+            if line_num is not None:  # be careful for '0' checks!
+                if line_num not in self.line_to_pc:
+                    self.line_to_pc[line_num] = []
+                self.line_to_pc[line_num].append(index)
+                last_line = line_num
+
+            self.pc_to_line[index] = last_line
+
+    def get_line_for_pc(self, pc: int) -> int:
+        return self.pc_to_line[pc]
+
+    def get_pcs_for_line(self, line: int) -> List[int]:
+        return self.line_to_pc[line]
+
+
+def _decode_int_value(value: str) -> int:
+    # Mappings may have up to 5 segments:
+    # Third segment represents the zero-based starting line in the original source represented.
+    decoded_value = base64vlq_decode(value)
+    return decoded_value[2] if decoded_value else None
+
+
+"""
+Source taken from: https://gist.github.com/mjpieters/86b0d152bb51d5f5979346d11005588b
+"""
+
+_b64chars = b"ABCDEFGHIJKLMNOPQRSTUVWXYZabcdefghijklmnopqrstuvwxyz0123456789+/"
+_b64table = [None] * (max(_b64chars) + 1)
+for i, b in enumerate(_b64chars):
+    _b64table[b] = i
+
+_shiftsize, _flag, _mask = 5, 1 << 5, (1 << 5) - 1
+
+
+def base64vlq_decode(vlqval: str) -> Tuple[int]:
+    """Decode Base64 VLQ value"""
+    results = []
+    add = results.append
+    shiftsize, flag, mask = _shiftsize, _flag, _mask
+    shift = value = 0
+    # use byte values and a table to go from base64 characters to integers
+    for v in map(_b64table.__getitem__, vlqval.encode("ascii")):
+        value += (v & mask) << shift
+        if v & flag:
+            shift += shiftsize
+            continue
+        # determine sign and add to results
+        add((value >> 1) * (-1 if value & 1 else 1))
+        shift = value = 0
+    return results
+
+
+def base64vlq_encode(*values: int) -> str:
+    """Encode integers to a VLQ value"""
+    results = []
+    add = results.append
+    shiftsize, flag, mask = _shiftsize, _flag, _mask
+    for v in values:
+        # add sign bit
+        v = (abs(v) << 1) | int(v < 0)
+        while True:
+            toencode, v = v & mask, v >> shiftsize
+            add(toencode | (v and flag))
+            if not v:
+                break
+    return bytes(map(_b64chars.__getitem__, results)).decode()

algosdk/v2client/algod.py

@@ -347,7 +347,7 @@ def suggested_params(self, **kwargs):
             res["min-fee"],
         )
 
-    def compile(self, source, **kwargs):
+    def compile(self, source, source_map=False, **kwargs):
         """
         Compile TEAL source with remote algod.
 
@@ -365,9 +365,9 @@ def compile(self, source, **kwargs):
             {"Content-Type": "application/x-binary"},
         )
         kwargs["headers"] = headers
-
+        params = {"sourcemap": source_map}
         return self.algod_request(
-            "POST", req, data=source.encode("utf-8"), **kwargs
+            "POST", req, params=params, data=source.encode("utf-8"), **kwargs
         )
 
     def dryrun(self, drr, **kwargs):

tests/steps/other_v2_steps.py

@@ -18,12 +18,7 @@
 from glom import glom
 import parse
 
-from algosdk import (
-    dryrun_results,
-    encoding,
-    error,
-    mnemonic,
-)
+from algosdk import dryrun_results, encoding, error, mnemonic, source_map
 from algosdk.error import AlgodHTTPError
 from algosdk.future import transaction
 from algosdk.v2client import *
@@ -1707,3 +1702,17 @@ def glom_app_eval_delta(context, i, path, field):
     assert field == str(
         actual_field
     ), f"path [{path}] expected value [{field}] but got [{actual_field}] instead"
+
+
+@given('a source map json file "{sourcemap_file}"')
+def parse_source_map(context, sourcemap_file):
+    jsmap = json.loads(load_resource(sourcemap_file, is_binary=False))
+    context.source_map = source_map.SourceMap(jsmap)
+
+
+@then('the string composed of pc:line number equals "{pc_to_line}"')
+def check_source_map(context, pc_to_line):
+    buff = [
+        f"{pc}:{line}" for pc, line in context.source_map.pc_to_line.items()
+    ]
+    assert ";".join(buff) == pc_to_line