(path)
| 18 | |
| 19 | |
| 20 | def tokenizer(path): |
| 21 | # pipe byte stream |
| 22 | process = subprocess.Popen(["gunzip", "--stdout", path], stdout=subprocess.PIPE) |
| 23 | buffer = bytearray() |
| 24 | assert process.stdout |
| 25 | head = process.stdout.read(4) |
| 26 | if head != b"SLF0": |
| 27 | raise ValueError(f"invalid file {path}, should be a xcactivitylog") |
| 28 | |
| 29 | def null_handler(index): |
| 30 | del buffer[: index + 1] |
| 31 | return (TokenType.Null, None) |
| 32 | |
| 33 | def int_handler(type): |
| 34 | def handler(index): |
| 35 | v = int(buffer[:index]) |
| 36 | del buffer[: index + 1] |
| 37 | return (type, v) |
| 38 | |
| 39 | return handler |
| 40 | |
| 41 | def double_handler(index): |
| 42 | v = buffer[:index] |
| 43 | v = bytes.fromhex(v.decode()) |
| 44 | v = struct.unpack("<d", v)[0] |
| 45 | del buffer[: index + 1] |
| 46 | return (TokenType.Double, v) |
| 47 | |
| 48 | def str_handler(type): |
| 49 | def handler(index): |
| 50 | length = int(buffer[:index]) |
| 51 | start = index + 1 |
| 52 | available = len(buffer) - start |
| 53 | if length > available: |
| 54 | bstr = buffer[start:] |
| 55 | bstr += process.stdout.read(length - available) # type: ignore |
| 56 | del buffer[:] |
| 57 | else: |
| 58 | bstr = buffer[start : (length + start)] |
| 59 | del buffer[: start + length] |
| 60 | return (type, bstr.decode()) |
| 61 | |
| 62 | return handler |
| 63 | |
| 64 | handler_map = { |
| 65 | ord(b'"'): str_handler(TokenType.String), |
| 66 | ord(b"-"): null_handler, |
| 67 | ord(b"#"): int_handler(TokenType.Integer), |
| 68 | ord(b"^"): double_handler, |
| 69 | ord(b"("): int_handler(TokenType.Array), |
| 70 | ord(b"%"): str_handler(TokenType.Class), |
| 71 | ord(b"@"): int_handler(TokenType.Instance), |
| 72 | ord(b"*"): str_handler(TokenType.Json), |
| 73 | } |
| 74 | |
| 75 | i = 0 |
| 76 | while v := process.stdout.read(16): |
| 77 | buffer.extend(v) |
no test coverage detected