Skip to content
This repository was archived by the owner on Apr 4, 2024. It is now read-only.

Commit b66e44e

Browse files
TrickybrainTrickybrain
committedMar 28, 2024
SnapshotReader - Edwin - Not completed
1 parent d04a06d commit b66e44e

8 files changed

+358
-50
lines changed
 

‎python/selfie-lib/selfie_lib/ConvertToWindowsNewlines.py

+20
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,20 @@
1+
class ConvertToWindowsNewlines:
2+
def __init__(self, sink):
3+
self.sink = sink
4+
5+
def append(self, value, start_index=None, end_index=None):
6+
# If value is a single character
7+
if isinstance(value, str) and len(value) == 1:
8+
if value != '\n':
9+
self.sink.write(value)
10+
else:
11+
self.sink.write('\r\n')
12+
# If value is a CharSequence (in Python, a str)
13+
elif isinstance(value, str):
14+
# If start_index and end_index are provided, use the slice of the string
15+
if start_index is not None and end_index is not None:
16+
value_to_append = value[start_index:end_index]
17+
else:
18+
value_to_append = value
19+
self.sink.write(value_to_append.replace("\n", "\r\n"))
20+
return self

‎python/selfie-lib/selfie_lib/Snapshot.py

+64
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,64 @@
1+
from .SnapshotValue import SnapshotValue
2+
3+
class Snapshot:
4+
def __init__(self, subject, facet_data=None):
5+
if facet_data is None:
6+
facet_data = {}
7+
self.subject = subject
8+
self._facet_data = facet_data
9+
10+
@property
11+
def facets(self):
12+
return self._facet_data
13+
14+
def plus_facet(self, key, value):
15+
if not key:
16+
raise ValueError("The empty string is reserved for the subject.")
17+
new_facet_data = self._facet_data.copy()
18+
new_facet_data[self._unix_newlines(key)] = SnapshotValue.of(value)
19+
return Snapshot(self.subject, new_facet_data)
20+
21+
def plus_or_replace(self, key, value):
22+
if not key:
23+
return Snapshot(value, self._facet_data)
24+
new_facet_data = self._facet_data.copy()
25+
new_facet_data[self._unix_newlines(key)] = value
26+
return Snapshot(self.subject, new_facet_data)
27+
28+
def subject_or_facet_maybe(self, key):
29+
if not key:
30+
return self.subject
31+
return self._facet_data.get(key)
32+
33+
def subject_or_facet(self, key):
34+
result = self.subject_or_facet_maybe(key)
35+
if result is None:
36+
raise KeyError(f"'{key}' not found in {list(self._facet_data.keys())}")
37+
return result
38+
39+
def all_entries(self):
40+
return {**{"": self.subject}, **self._facet_data}
41+
42+
def __str__(self):
43+
return f"[{self.subject} {self._facet_data}]"
44+
45+
@staticmethod
46+
def _unix_newlines(key):
47+
return key.replace("\r\n", "\n")
48+
49+
@classmethod
50+
def of(cls, value):
51+
return cls(SnapshotValue.of(value))
52+
53+
@classmethod
54+
def of_entries(cls, entries):
55+
root = None
56+
facets = {}
57+
for key, value in entries:
58+
if not key:
59+
if root is not None:
60+
raise ValueError("Duplicate root snapshot.")
61+
root = value
62+
else:
63+
facets[key] = value
64+
return cls(root or SnapshotValue.of(""), facets)

‎python/selfie-lib/selfie_lib/SnapshotFile.py

+124
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,124 @@
1+
import threading
2+
from typing import List
3+
import base64
4+
5+
from .SnapshotValue import SnapshotValue
6+
from .ConvertToWindowsNewlines import ConvertToWindowsNewlines
7+
from .ParseException import ParseException
8+
from .SnapshotReader import SnapshotReader
9+
from .SnapshotValueReader import SnapshotValueReader
10+
11+
12+
13+
class SnapshotFile:
14+
HEADER_PREFIX = "📷 "
15+
END_OF_FILE = "[end of file]"
16+
17+
def __init__(self):
18+
self.unix_newlines = True
19+
self.metadata = None
20+
self._snapshots = {}
21+
self._lock = threading.Lock()
22+
self._was_set_at_test_time = False
23+
24+
@property
25+
def snapshots(self):
26+
return self._snapshots
27+
28+
@snapshots.setter
29+
def snapshots(self, value):
30+
with self._lock:
31+
self._snapshots = value
32+
33+
@property
34+
def was_set_at_test_time(self):
35+
return self._was_set_at_test_time
36+
37+
def set_at_test_time(self, key, snapshot):
38+
with self._lock:
39+
old_snapshots = self._snapshots.copy()
40+
self._snapshots[key] = snapshot
41+
self._was_set_at_test_time = True if self._snapshots != old_snapshots else self._was_set_at_test_time
42+
43+
def serialize(self, value_writer_raw):
44+
value_writer = value_writer_raw if self.unix_newlines else ConvertToWindowsNewlines(value_writer_raw)
45+
if self.metadata:
46+
self.write_entry(value_writer, f"📷 {self.metadata[0]}", None, SnapshotValue.of(self.metadata[1]))
47+
for key, snapshot in self._snapshots.items():
48+
self.write_entry(value_writer, key, None, snapshot.subject)
49+
for facet_key, facet_value in snapshot.facets.items():
50+
self.write_entry(value_writer, key, facet_key, facet_value)
51+
self.write_entry(value_writer, "", "end of file", SnapshotValue.of(""))
52+
53+
def write_entry(value_writer, key, facet, value):
54+
value_writer.write("╔═ ")
55+
value_writer.write(SnapshotValueReader.nameEsc.escape(key))
56+
if facet is not None:
57+
value_writer.write("[")
58+
value_writer.write(SnapshotValueReader.nameEsc.escape(facet))
59+
value_writer.write("]")
60+
value_writer.write(" ═╗")
61+
if value.is_binary:
62+
binary_length = len(value.value_binary())
63+
value_writer.write(f" base64 length {binary_length} bytes")
64+
value_writer.write("\n")
65+
66+
if key == "" and facet == "end of file":
67+
return
68+
69+
if value.is_binary:
70+
# Base64 encoding and replacing \r with an empty string
71+
binary_data = value.value_binary()
72+
encoded = base64.b64encode(binary_data).decode('utf-8')
73+
# Assuming efficientReplace is a more efficient method for replacing characters
74+
# Here, we just use the regular replace method for simplicity
75+
escaped = encoded.replace("\r", "")
76+
value_writer.write(escaped)
77+
else:
78+
# For string values, applying specific escape logic and then replacing "\n╔" with a special sequence
79+
text_data = value.value_string()
80+
# Assuming body_escape function handles the escaping as in SnapshotValueReader.bodyEsc.escape
81+
escaped = body_escape(text_data).replace("\n╔", "\n\uDF41")
82+
value_writer.write(escaped)
83+
value_writer.write("\n")
84+
85+
@staticmethod
86+
def parse(value_reader):
87+
try:
88+
result = SnapshotFile()
89+
result.unix_newlines = value_reader.unix_newlines
90+
reader = SnapshotReader(value_reader)
91+
92+
# Check if the first value starts with 📷
93+
if reader.peek_key() and reader.peek_key().startswith(SnapshotFile.HEADER_PREFIX):
94+
metadata_name = reader.peek_key()[len(SnapshotFile.HEADER_PREFIX):]
95+
metadata_value = reader.value_reader.next_value().value_string()
96+
# Assuming 'entry' function creates a dictionary entry in Python
97+
result.metadata = (metadata_name, metadata_value)
98+
99+
while reader.peek_key() is not None:
100+
key = reader.peek_key()
101+
snapshot = reader.next_snapshot()
102+
# Update snapshots dictionary with new key-value pair
103+
result.snapshots.update({key: snapshot})
104+
105+
return result
106+
107+
except ValueError as e:
108+
if isinstance(e, ParseException):
109+
raise e
110+
else:
111+
raise ParseException(value_reader.line_reader, e) from None
112+
113+
114+
@staticmethod
115+
def create_empty_with_unix_newlines(unix_newlines):
116+
result = SnapshotFile()
117+
result.unix_newlines = unix_newlines
118+
return result
119+
120+
def remove_all_indices(self, indices: List[int]):
121+
if not indices:
122+
return
123+
self._was_set_at_test_time = True
124+
self.snapshots = self.snapshots.minus_sorted_indices(indices)

‎python/selfie-lib/selfie_lib/SnapshotReader.py

+45
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,45 @@
1+
from .Snapshot import Snapshot
2+
from .SnapshotFile import SnapshotFile
3+
4+
5+
class SnapshotReader:
6+
def __init__(self, value_reader):
7+
self.value_reader = value_reader
8+
9+
def peek_key(self):
10+
next_key = self.value_reader.peek_key()
11+
if next_key is None or next_key == SnapshotFile.END_OF_FILE:
12+
return None
13+
if '[' in next_key:
14+
raise ValueError(f"Missing root snapshot, square brackets not allowed: '{next_key}'")
15+
return next_key
16+
17+
def next_snapshot(self):
18+
root_name = self.peek_key()
19+
snapshot = Snapshot.of(self.value_reader.next_value())
20+
while True:
21+
next_key = self.value_reader.peek_key()
22+
if next_key is None:
23+
return snapshot
24+
facet_idx = next_key.find('[')
25+
if facet_idx == -1 or (facet_idx == 0 and next_key == SnapshotFile.END_OF_FILE):
26+
return snapshot
27+
facet_root = next_key[:facet_idx]
28+
if facet_root != root_name:
29+
raise ValueError(f"Expected '{next_key}' to come after '{facet_root}', not '{root_name}'")
30+
facet_end_idx = next_key.find(']', facet_idx + 1)
31+
if facet_end_idx == -1:
32+
raise ValueError(f"Missing ] in {next_key}")
33+
facet_name = next_key[facet_idx + 1:facet_end_idx]
34+
snapshot = snapshot.plus_facet(facet_name, self.value_reader.next_value())
35+
36+
def skip_snapshot(self):
37+
root_name = self.peek_key()
38+
if root_name is None:
39+
raise ValueError("No snapshot to skip")
40+
self.value_reader.skip_value()
41+
while True:
42+
next_key = self.peek_key()
43+
if next_key is None or not next_key.startswith(f"{root_name}["):
44+
break
45+
self.value_reader.skip_value()

‎python/selfie-lib/selfie_lib/SnapshotValue.py

+47
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,47 @@
1+
from abc import ABC, abstractmethod
2+
from typing import Union
3+
4+
def unix_newlines(string: str) -> str:
5+
return string.replace("\r\n", "\n")
6+
7+
class SnapshotValue(ABC):
8+
@property
9+
def is_binary(self) -> bool:
10+
return isinstance(self, SnapshotValueBinary)
11+
12+
@abstractmethod
13+
def value_binary(self) -> bytes:
14+
pass
15+
16+
@abstractmethod
17+
def value_string(self) -> str:
18+
pass
19+
20+
@staticmethod
21+
def of(value: Union[bytes, str]) -> "SnapshotValue":
22+
if isinstance(value, bytes):
23+
return SnapshotValueBinary(value)
24+
elif isinstance(value, str):
25+
return SnapshotValueString(unix_newlines(value))
26+
else:
27+
raise TypeError("Value must be either bytes or str")
28+
29+
class SnapshotValueBinary(SnapshotValue):
30+
def __init__(self, value: bytes):
31+
self._value = value
32+
33+
def value_binary(self) -> bytes:
34+
return self._value
35+
36+
def value_string(self) -> str:
37+
raise NotImplementedError("This is a binary value.")
38+
39+
class SnapshotValueString(SnapshotValue):
40+
def __init__(self, value: str):
41+
self._value = value
42+
43+
def value_binary(self) -> bytes:
44+
raise NotImplementedError("This is a string value.")
45+
46+
def value_string(self) -> str:
47+
return self._value

‎python/selfie-lib/selfie_lib/SnapshotValueReader.py

+1-50
Original file line numberDiff line numberDiff line change
@@ -1,61 +1,12 @@
11
import base64
2-
3-
from abc import ABC, abstractmethod
4-
from typing import Union
52
from .PerCharacterEscaper import PerCharacterEscaper
63
from .ParseException import ParseException
74
from .LineReader import LineReader
8-
5+
from .SnapshotValue import SnapshotValue
96

107
def unix_newlines(string: str) -> str:
118
return string.replace("\r\n", "\n")
129

13-
14-
class SnapshotValue(ABC):
15-
@property
16-
def is_binary(self) -> bool:
17-
return isinstance(self, SnapshotValueBinary)
18-
19-
@abstractmethod
20-
def value_binary(self) -> bytes:
21-
pass
22-
23-
@abstractmethod
24-
def value_string(self) -> str:
25-
pass
26-
27-
@staticmethod
28-
def of(value: Union[bytes, str]) -> "SnapshotValue":
29-
if isinstance(value, bytes):
30-
return SnapshotValueBinary(value)
31-
elif isinstance(value, str):
32-
return SnapshotValueString(unix_newlines(value))
33-
else:
34-
raise TypeError("Value must be either bytes or str")
35-
36-
37-
class SnapshotValueBinary(SnapshotValue):
38-
def __init__(self, value: bytes):
39-
self._value = value
40-
41-
def value_binary(self) -> bytes:
42-
return self._value
43-
44-
def value_string(self) -> str:
45-
raise NotImplementedError("This is a binary value.")
46-
47-
48-
class SnapshotValueString(SnapshotValue):
49-
def __init__(self, value: str):
50-
self._value = value
51-
52-
def value_binary(self) -> bytes:
53-
raise NotImplementedError("This is a string value.")
54-
55-
def value_string(self) -> str:
56-
return self._value
57-
58-
5910
class SnapshotValueReader:
6011
KEY_FIRST_CHAR = "╔"
6112
KEY_START = "╔═ "

‎python/selfie-lib/selfie_lib/__init__.py

+6
Original file line numberDiff line numberDiff line change
@@ -4,3 +4,9 @@
44
from .PerCharacterEscaper import PerCharacterEscaper as PerCharacterEscaper
55
from .SnapshotValueReader import SnapshotValueReader as SnapshotValueReader
66
from .ParseException import ParseException as ParseException
7+
from .SnapshotReader import SnapshotReader as SnapshotReader
8+
from .Snapshot import Snapshot as Snapshot
9+
# from .SnapshotValue import SnapshotValue
10+
# from .SnapshotFile import SnapshotFile
11+
# from .SnapshotValueString import SnapshotValueString
12+
# from .SnapshotValueBinary import SnapshotValueBinary

0 commit comments

Comments
 (0)