From b66e44e01d317b60876f8278184dc3d5de9e198d Mon Sep 17 00:00:00 2001 From: Edwin Date: Thu, 28 Mar 2024 11:15:30 -0700 Subject: [PATCH 1/7] SnapshotReader - Edwin - Not completed --- .../selfie_lib/ConvertToWindowsNewlines.py | 20 +++ python/selfie-lib/selfie_lib/Snapshot.py | 64 +++++++++ python/selfie-lib/selfie_lib/SnapshotFile.py | 124 ++++++++++++++++++ .../selfie-lib/selfie_lib/SnapshotReader.py | 45 +++++++ python/selfie-lib/selfie_lib/SnapshotValue.py | 47 +++++++ .../selfie_lib/SnapshotValueReader.py | 51 +------ python/selfie-lib/selfie_lib/__init__.py | 6 + .../selfie-lib/tests/SnapshotReader_test.py | 51 +++++++ 8 files changed, 358 insertions(+), 50 deletions(-) create mode 100644 python/selfie-lib/selfie_lib/ConvertToWindowsNewlines.py create mode 100644 python/selfie-lib/selfie_lib/Snapshot.py create mode 100644 python/selfie-lib/selfie_lib/SnapshotFile.py create mode 100644 python/selfie-lib/selfie_lib/SnapshotReader.py create mode 100644 python/selfie-lib/selfie_lib/SnapshotValue.py create mode 100644 python/selfie-lib/tests/SnapshotReader_test.py diff --git a/python/selfie-lib/selfie_lib/ConvertToWindowsNewlines.py b/python/selfie-lib/selfie_lib/ConvertToWindowsNewlines.py new file mode 100644 index 00000000..f4615d6f --- /dev/null +++ b/python/selfie-lib/selfie_lib/ConvertToWindowsNewlines.py @@ -0,0 +1,20 @@ +class ConvertToWindowsNewlines: + def __init__(self, sink): + self.sink = sink + + def append(self, value, start_index=None, end_index=None): + # If value is a single character + if isinstance(value, str) and len(value) == 1: + if value != '\n': + self.sink.write(value) + else: + self.sink.write('\r\n') + # If value is a CharSequence (in Python, a str) + elif isinstance(value, str): + # If start_index and end_index are provided, use the slice of the string + if start_index is not None and end_index is not None: + value_to_append = value[start_index:end_index] + else: + value_to_append = value + self.sink.write(value_to_append.replace("\n", "\r\n")) + return self diff --git a/python/selfie-lib/selfie_lib/Snapshot.py b/python/selfie-lib/selfie_lib/Snapshot.py new file mode 100644 index 00000000..7528a0ff --- /dev/null +++ b/python/selfie-lib/selfie_lib/Snapshot.py @@ -0,0 +1,64 @@ +from .SnapshotValue import SnapshotValue + +class Snapshot: + def __init__(self, subject, facet_data=None): + if facet_data is None: + facet_data = {} + self.subject = subject + self._facet_data = facet_data + + @property + def facets(self): + return self._facet_data + + def plus_facet(self, key, value): + if not key: + raise ValueError("The empty string is reserved for the subject.") + new_facet_data = self._facet_data.copy() + new_facet_data[self._unix_newlines(key)] = SnapshotValue.of(value) + return Snapshot(self.subject, new_facet_data) + + def plus_or_replace(self, key, value): + if not key: + return Snapshot(value, self._facet_data) + new_facet_data = self._facet_data.copy() + new_facet_data[self._unix_newlines(key)] = value + return Snapshot(self.subject, new_facet_data) + + def subject_or_facet_maybe(self, key): + if not key: + return self.subject + return self._facet_data.get(key) + + def subject_or_facet(self, key): + result = self.subject_or_facet_maybe(key) + if result is None: + raise KeyError(f"'{key}' not found in {list(self._facet_data.keys())}") + return result + + def all_entries(self): + return {**{"": self.subject}, **self._facet_data} + + def __str__(self): + return f"[{self.subject} {self._facet_data}]" + + @staticmethod + def _unix_newlines(key): + return key.replace("\r\n", "\n") + + @classmethod + def of(cls, value): + return cls(SnapshotValue.of(value)) + + @classmethod + def of_entries(cls, entries): + root = None + facets = {} + for key, value in entries: + if not key: + if root is not None: + raise ValueError("Duplicate root snapshot.") + root = value + else: + facets[key] = value + return cls(root or SnapshotValue.of(""), facets) diff --git a/python/selfie-lib/selfie_lib/SnapshotFile.py b/python/selfie-lib/selfie_lib/SnapshotFile.py new file mode 100644 index 00000000..f2316661 --- /dev/null +++ b/python/selfie-lib/selfie_lib/SnapshotFile.py @@ -0,0 +1,124 @@ +import threading +from typing import List +import base64 + +from .SnapshotValue import SnapshotValue +from .ConvertToWindowsNewlines import ConvertToWindowsNewlines +from .ParseException import ParseException +from .SnapshotReader import SnapshotReader +from .SnapshotValueReader import SnapshotValueReader + + + +class SnapshotFile: + HEADER_PREFIX = "šŸ“· " + END_OF_FILE = "[end of file]" + + def __init__(self): + self.unix_newlines = True + self.metadata = None + self._snapshots = {} + self._lock = threading.Lock() + self._was_set_at_test_time = False + + @property + def snapshots(self): + return self._snapshots + + @snapshots.setter + def snapshots(self, value): + with self._lock: + self._snapshots = value + + @property + def was_set_at_test_time(self): + return self._was_set_at_test_time + + def set_at_test_time(self, key, snapshot): + with self._lock: + old_snapshots = self._snapshots.copy() + self._snapshots[key] = snapshot + self._was_set_at_test_time = True if self._snapshots != old_snapshots else self._was_set_at_test_time + + def serialize(self, value_writer_raw): + value_writer = value_writer_raw if self.unix_newlines else ConvertToWindowsNewlines(value_writer_raw) + if self.metadata: + self.write_entry(value_writer, f"šŸ“· {self.metadata[0]}", None, SnapshotValue.of(self.metadata[1])) + for key, snapshot in self._snapshots.items(): + self.write_entry(value_writer, key, None, snapshot.subject) + for facet_key, facet_value in snapshot.facets.items(): + self.write_entry(value_writer, key, facet_key, facet_value) + self.write_entry(value_writer, "", "end of file", SnapshotValue.of("")) + + def write_entry(value_writer, key, facet, value): + value_writer.write("ā•”ā• ") + value_writer.write(SnapshotValueReader.nameEsc.escape(key)) + if facet is not None: + value_writer.write("[") + value_writer.write(SnapshotValueReader.nameEsc.escape(facet)) + value_writer.write("]") + value_writer.write(" ā•ā•—") + if value.is_binary: + binary_length = len(value.value_binary()) + value_writer.write(f" base64 length {binary_length} bytes") + value_writer.write("\n") + + if key == "" and facet == "end of file": + return + + if value.is_binary: + # Base64 encoding and replacing \r with an empty string + binary_data = value.value_binary() + encoded = base64.b64encode(binary_data).decode('utf-8') + # Assuming efficientReplace is a more efficient method for replacing characters + # Here, we just use the regular replace method for simplicity + escaped = encoded.replace("\r", "") + value_writer.write(escaped) + else: + # For string values, applying specific escape logic and then replacing "\nā•”" with a special sequence + text_data = value.value_string() + # Assuming body_escape function handles the escaping as in SnapshotValueReader.bodyEsc.escape + escaped = body_escape(text_data).replace("\nā•”", "\n\uDF41") + value_writer.write(escaped) + value_writer.write("\n") + + @staticmethod + def parse(value_reader): + try: + result = SnapshotFile() + result.unix_newlines = value_reader.unix_newlines + reader = SnapshotReader(value_reader) + + # Check if the first value starts with šŸ“· + if reader.peek_key() and reader.peek_key().startswith(SnapshotFile.HEADER_PREFIX): + metadata_name = reader.peek_key()[len(SnapshotFile.HEADER_PREFIX):] + metadata_value = reader.value_reader.next_value().value_string() + # Assuming 'entry' function creates a dictionary entry in Python + result.metadata = (metadata_name, metadata_value) + + while reader.peek_key() is not None: + key = reader.peek_key() + snapshot = reader.next_snapshot() + # Update snapshots dictionary with new key-value pair + result.snapshots.update({key: snapshot}) + + return result + + except ValueError as e: + if isinstance(e, ParseException): + raise e + else: + raise ParseException(value_reader.line_reader, e) from None + + + @staticmethod + def create_empty_with_unix_newlines(unix_newlines): + result = SnapshotFile() + result.unix_newlines = unix_newlines + return result + + def remove_all_indices(self, indices: List[int]): + if not indices: + return + self._was_set_at_test_time = True + self.snapshots = self.snapshots.minus_sorted_indices(indices) diff --git a/python/selfie-lib/selfie_lib/SnapshotReader.py b/python/selfie-lib/selfie_lib/SnapshotReader.py new file mode 100644 index 00000000..846b2a41 --- /dev/null +++ b/python/selfie-lib/selfie_lib/SnapshotReader.py @@ -0,0 +1,45 @@ +from .Snapshot import Snapshot +from .SnapshotFile import SnapshotFile + + +class SnapshotReader: + def __init__(self, value_reader): + self.value_reader = value_reader + + def peek_key(self): + next_key = self.value_reader.peek_key() + if next_key is None or next_key == SnapshotFile.END_OF_FILE: + return None + if '[' in next_key: + raise ValueError(f"Missing root snapshot, square brackets not allowed: '{next_key}'") + return next_key + + def next_snapshot(self): + root_name = self.peek_key() + snapshot = Snapshot.of(self.value_reader.next_value()) + while True: + next_key = self.value_reader.peek_key() + if next_key is None: + return snapshot + facet_idx = next_key.find('[') + if facet_idx == -1 or (facet_idx == 0 and next_key == SnapshotFile.END_OF_FILE): + return snapshot + facet_root = next_key[:facet_idx] + if facet_root != root_name: + raise ValueError(f"Expected '{next_key}' to come after '{facet_root}', not '{root_name}'") + facet_end_idx = next_key.find(']', facet_idx + 1) + if facet_end_idx == -1: + raise ValueError(f"Missing ] in {next_key}") + facet_name = next_key[facet_idx + 1:facet_end_idx] + snapshot = snapshot.plus_facet(facet_name, self.value_reader.next_value()) + + def skip_snapshot(self): + root_name = self.peek_key() + if root_name is None: + raise ValueError("No snapshot to skip") + self.value_reader.skip_value() + while True: + next_key = self.peek_key() + if next_key is None or not next_key.startswith(f"{root_name}["): + break + self.value_reader.skip_value() diff --git a/python/selfie-lib/selfie_lib/SnapshotValue.py b/python/selfie-lib/selfie_lib/SnapshotValue.py new file mode 100644 index 00000000..2f76c30d --- /dev/null +++ b/python/selfie-lib/selfie_lib/SnapshotValue.py @@ -0,0 +1,47 @@ +from abc import ABC, abstractmethod +from typing import Union + +def unix_newlines(string: str) -> str: + return string.replace("\r\n", "\n") + +class SnapshotValue(ABC): + @property + def is_binary(self) -> bool: + return isinstance(self, SnapshotValueBinary) + + @abstractmethod + def value_binary(self) -> bytes: + pass + + @abstractmethod + def value_string(self) -> str: + pass + + @staticmethod + def of(value: Union[bytes, str]) -> "SnapshotValue": + if isinstance(value, bytes): + return SnapshotValueBinary(value) + elif isinstance(value, str): + return SnapshotValueString(unix_newlines(value)) + else: + raise TypeError("Value must be either bytes or str") + +class SnapshotValueBinary(SnapshotValue): + def __init__(self, value: bytes): + self._value = value + + def value_binary(self) -> bytes: + return self._value + + def value_string(self) -> str: + raise NotImplementedError("This is a binary value.") + +class SnapshotValueString(SnapshotValue): + def __init__(self, value: str): + self._value = value + + def value_binary(self) -> bytes: + raise NotImplementedError("This is a string value.") + + def value_string(self) -> str: + return self._value \ No newline at end of file diff --git a/python/selfie-lib/selfie_lib/SnapshotValueReader.py b/python/selfie-lib/selfie_lib/SnapshotValueReader.py index 11d1735e..b3731da3 100644 --- a/python/selfie-lib/selfie_lib/SnapshotValueReader.py +++ b/python/selfie-lib/selfie_lib/SnapshotValueReader.py @@ -1,61 +1,12 @@ import base64 - -from abc import ABC, abstractmethod -from typing import Union from .PerCharacterEscaper import PerCharacterEscaper from .ParseException import ParseException from .LineReader import LineReader - +from .SnapshotValue import SnapshotValue def unix_newlines(string: str) -> str: return string.replace("\r\n", "\n") - -class SnapshotValue(ABC): - @property - def is_binary(self) -> bool: - return isinstance(self, SnapshotValueBinary) - - @abstractmethod - def value_binary(self) -> bytes: - pass - - @abstractmethod - def value_string(self) -> str: - pass - - @staticmethod - def of(value: Union[bytes, str]) -> "SnapshotValue": - if isinstance(value, bytes): - return SnapshotValueBinary(value) - elif isinstance(value, str): - return SnapshotValueString(unix_newlines(value)) - else: - raise TypeError("Value must be either bytes or str") - - -class SnapshotValueBinary(SnapshotValue): - def __init__(self, value: bytes): - self._value = value - - def value_binary(self) -> bytes: - return self._value - - def value_string(self) -> str: - raise NotImplementedError("This is a binary value.") - - -class SnapshotValueString(SnapshotValue): - def __init__(self, value: str): - self._value = value - - def value_binary(self) -> bytes: - raise NotImplementedError("This is a string value.") - - def value_string(self) -> str: - return self._value - - class SnapshotValueReader: KEY_FIRST_CHAR = "ā•”" KEY_START = "ā•”ā• " diff --git a/python/selfie-lib/selfie_lib/__init__.py b/python/selfie-lib/selfie_lib/__init__.py index 87b47f7a..57d835af 100644 --- a/python/selfie-lib/selfie_lib/__init__.py +++ b/python/selfie-lib/selfie_lib/__init__.py @@ -4,3 +4,9 @@ from .PerCharacterEscaper import PerCharacterEscaper as PerCharacterEscaper from .SnapshotValueReader import SnapshotValueReader as SnapshotValueReader from .ParseException import ParseException as ParseException +from .SnapshotReader import SnapshotReader as SnapshotReader +from .Snapshot import Snapshot as Snapshot +# from .SnapshotValue import SnapshotValue +# from .SnapshotFile import SnapshotFile +# from .SnapshotValueString import SnapshotValueString +# from .SnapshotValueBinary import SnapshotValueBinary diff --git a/python/selfie-lib/tests/SnapshotReader_test.py b/python/selfie-lib/tests/SnapshotReader_test.py new file mode 100644 index 00000000..524176f9 --- /dev/null +++ b/python/selfie-lib/tests/SnapshotReader_test.py @@ -0,0 +1,51 @@ +from base64 import b64decode +from selfie_lib import SnapshotValueReader, Snapshot, SnapshotReader + +class TestSnapshotReader: + def test_facet(self): + reader = SnapshotReader( + SnapshotValueReader.of( + """ + ā•”ā• Apple ā•ā•— + Apple + ā•”ā• Apple[color] ā•ā•— + green + ā•”ā• Apple[crisp] ā•ā•— + yes + ā•”ā• Orange ā•ā•— + Orange + """.strip() + ) + ) + assert reader.peek_key() == "Apple" + assert reader.peek_key() == "Apple" + apple_snapshot = Snapshot.of("Apple").plus_facet("color", "green").plus_facet("crisp", "yes") + assert reader.next_snapshot() == apple_snapshot + assert reader.peek_key() == "Orange" + assert reader.peek_key() == "Orange" + assert reader.next_snapshot() == Snapshot.of("Orange") + assert reader.peek_key() is None + + def test_binary(self): + reader = SnapshotReader( + SnapshotValueReader.of( + """ + ā•”ā• Apple ā•ā•— + Apple + ā•”ā• Apple[color] ā•ā•— base64 length 3 bytes + c2Fk + ā•”ā• Apple[crisp] ā•ā•— + yes + ā•”ā• Orange ā•ā•— base64 length 3 bytes + c2Fk + """.strip() + ) + ) + assert reader.peek_key() == "Apple" + assert reader.peek_key() == "Apple" + apple_snapshot = Snapshot.of("Apple").plus_facet("color", b64decode("c2Fk")).plus_facet("crisp", "yes") + assert reader.next_snapshot() == apple_snapshot + assert reader.peek_key() == "Orange" + assert reader.peek_key() == "Orange" + assert reader.next_snapshot() == Snapshot.of(b64decode("c2Fk")) + assert reader.peek_key() is None From 0612d8d3ec97c91449da0b0191189109934ef2b5 Mon Sep 17 00:00:00 2001 From: Edwin Date: Thu, 28 Mar 2024 16:41:47 -0700 Subject: [PATCH 2/7] SnapshotReader determine value type error --- python/selfie-lib/selfie_lib/Snapshot.py | 76 +++++++++++-------- python/selfie-lib/selfie_lib/SnapshotFile.py | 5 +- .../selfie-lib/selfie_lib/SnapshotReader.py | 6 +- python/selfie-lib/selfie_lib/SnapshotValue.py | 6 +- python/selfie-lib/selfie_lib/__init__.py | 6 +- .../selfie-lib/tests/SnapshotReader_test.py | 20 ++--- 6 files changed, 65 insertions(+), 54 deletions(-) diff --git a/python/selfie-lib/selfie_lib/Snapshot.py b/python/selfie-lib/selfie_lib/Snapshot.py index 7528a0ff..c4de20ac 100644 --- a/python/selfie-lib/selfie_lib/Snapshot.py +++ b/python/selfie-lib/selfie_lib/Snapshot.py @@ -1,64 +1,78 @@ from .SnapshotValue import SnapshotValue +from collections import OrderedDict class Snapshot: - def __init__(self, subject, facet_data=None): - if facet_data is None: - facet_data = {} - self.subject = subject + def __init__(self, subject, facet_data): + self._subject = subject self._facet_data = facet_data @property def facets(self): - return self._facet_data + return OrderedDict(sorted(self._facet_data.items())) def plus_facet(self, key, value): + if isinstance(value, bytes): + value = SnapshotValue.of(value) + elif isinstance(value, str): + value = SnapshotValue.of(value) + return self._plus_facet(key, value) + + def _plus_facet(self, key, value): if not key: raise ValueError("The empty string is reserved for the subject.") - new_facet_data = self._facet_data.copy() - new_facet_data[self._unix_newlines(key)] = SnapshotValue.of(value) - return Snapshot(self.subject, new_facet_data) + facet_data = dict(self._facet_data) + facet_data[self._unix_newlines(key)] = value + return Snapshot(self._subject, facet_data) def plus_or_replace(self, key, value): if not key: return Snapshot(value, self._facet_data) - new_facet_data = self._facet_data.copy() - new_facet_data[self._unix_newlines(key)] = value - return Snapshot(self.subject, new_facet_data) + facet_data = dict(self._facet_data) + facet_data[self._unix_newlines(key)] = value + return Snapshot(self._subject, facet_data) def subject_or_facet_maybe(self, key): if not key: - return self.subject + return self._subject return self._facet_data.get(key) def subject_or_facet(self, key): - result = self.subject_or_facet_maybe(key) - if result is None: + value = self.subject_or_facet_maybe(key) + if value is None: raise KeyError(f"'{key}' not found in {list(self._facet_data.keys())}") - return result + return value def all_entries(self): - return {**{"": self.subject}, **self._facet_data} + entries = [("", self._subject)] + entries.extend(self._facet_data.items()) + return entries def __str__(self): - return f"[{self.subject} {self._facet_data}]" + return f"[{self._subject} {self._facet_data}]" @staticmethod - def _unix_newlines(key): - return key.replace("\r\n", "\n") + def of(binary): + return Snapshot(SnapshotValue.of(binary), {}) - @classmethod - def of(cls, value): - return cls(SnapshotValue.of(value)) + @staticmethod + def of(string): + return Snapshot(SnapshotValue.of(string), {}) - @classmethod - def of_entries(cls, entries): - root = None - facets = {} + @staticmethod + def of_entries(entries): + subject = None + facet_data = {} for key, value in entries: if not key: - if root is not None: - raise ValueError("Duplicate root snapshot.") - root = value + if subject is not None: + raise ValueError(f"Duplicate root snapshot.\n first: {subject}\nsecond: {value}") + subject = value else: - facets[key] = value - return cls(root or SnapshotValue.of(""), facets) + facet_data[key] = value + if subject is None: + subject = SnapshotValue.of("") + return Snapshot(subject, facet_data) + + @staticmethod + def _unix_newlines(string): + return string.replace("\\r\\n", "\\n") \ No newline at end of file diff --git a/python/selfie-lib/selfie_lib/SnapshotFile.py b/python/selfie-lib/selfie_lib/SnapshotFile.py index f2316661..3e70d455 100644 --- a/python/selfie-lib/selfie_lib/SnapshotFile.py +++ b/python/selfie-lib/selfie_lib/SnapshotFile.py @@ -8,8 +8,6 @@ from .SnapshotReader import SnapshotReader from .SnapshotValueReader import SnapshotValueReader - - class SnapshotFile: HEADER_PREFIX = "šŸ“· " END_OF_FILE = "[end of file]" @@ -77,8 +75,7 @@ def write_entry(value_writer, key, facet, value): else: # For string values, applying specific escape logic and then replacing "\nā•”" with a special sequence text_data = value.value_string() - # Assuming body_escape function handles the escaping as in SnapshotValueReader.bodyEsc.escape - escaped = body_escape(text_data).replace("\nā•”", "\n\uDF41") + escaped = SnapshotValueReader.bodyEsc(text_data).replace("\nā•”", "\n\uDF41") value_writer.write(escaped) value_writer.write("\n") diff --git a/python/selfie-lib/selfie_lib/SnapshotReader.py b/python/selfie-lib/selfie_lib/SnapshotReader.py index 846b2a41..0eaee273 100644 --- a/python/selfie-lib/selfie_lib/SnapshotReader.py +++ b/python/selfie-lib/selfie_lib/SnapshotReader.py @@ -1,6 +1,4 @@ from .Snapshot import Snapshot -from .SnapshotFile import SnapshotFile - class SnapshotReader: def __init__(self, value_reader): @@ -8,7 +6,7 @@ def __init__(self, value_reader): def peek_key(self): next_key = self.value_reader.peek_key() - if next_key is None or next_key == SnapshotFile.END_OF_FILE: + if next_key is None or next_key == "[end of file]": return None if '[' in next_key: raise ValueError(f"Missing root snapshot, square brackets not allowed: '{next_key}'") @@ -22,7 +20,7 @@ def next_snapshot(self): if next_key is None: return snapshot facet_idx = next_key.find('[') - if facet_idx == -1 or (facet_idx == 0 and next_key == SnapshotFile.END_OF_FILE): + if facet_idx == -1 or (facet_idx == 0 and next_key == "[end of file]"): return snapshot facet_root = next_key[:facet_idx] if facet_root != root_name: diff --git a/python/selfie-lib/selfie_lib/SnapshotValue.py b/python/selfie-lib/selfie_lib/SnapshotValue.py index 2f76c30d..84b7fca9 100644 --- a/python/selfie-lib/selfie_lib/SnapshotValue.py +++ b/python/selfie-lib/selfie_lib/SnapshotValue.py @@ -1,9 +1,11 @@ from abc import ABC, abstractmethod from typing import Union +# from .SnapshotValueBinary import SnapshotValueBinary +# from .SnapshotValueString import SnapshotValueString def unix_newlines(string: str) -> str: return string.replace("\r\n", "\n") - + class SnapshotValue(ABC): @property def is_binary(self) -> bool: @@ -25,6 +27,7 @@ def of(value: Union[bytes, str]) -> "SnapshotValue": return SnapshotValueString(unix_newlines(value)) else: raise TypeError("Value must be either bytes or str") + class SnapshotValueBinary(SnapshotValue): def __init__(self, value: bytes): @@ -36,6 +39,7 @@ def value_binary(self) -> bytes: def value_string(self) -> str: raise NotImplementedError("This is a binary value.") + class SnapshotValueString(SnapshotValue): def __init__(self, value: str): self._value = value diff --git a/python/selfie-lib/selfie_lib/__init__.py b/python/selfie-lib/selfie_lib/__init__.py index 57d835af..354dda2b 100644 --- a/python/selfie-lib/selfie_lib/__init__.py +++ b/python/selfie-lib/selfie_lib/__init__.py @@ -6,7 +6,5 @@ from .ParseException import ParseException as ParseException from .SnapshotReader import SnapshotReader as SnapshotReader from .Snapshot import Snapshot as Snapshot -# from .SnapshotValue import SnapshotValue -# from .SnapshotFile import SnapshotFile -# from .SnapshotValueString import SnapshotValueString -# from .SnapshotValueBinary import SnapshotValueBinary +from .SnapshotValue import SnapshotValue as SnapshotValue + diff --git a/python/selfie-lib/tests/SnapshotReader_test.py b/python/selfie-lib/tests/SnapshotReader_test.py index 524176f9..19386047 100644 --- a/python/selfie-lib/tests/SnapshotReader_test.py +++ b/python/selfie-lib/tests/SnapshotReader_test.py @@ -5,16 +5,16 @@ class TestSnapshotReader: def test_facet(self): reader = SnapshotReader( SnapshotValueReader.of( - """ - ā•”ā• Apple ā•ā•— - Apple - ā•”ā• Apple[color] ā•ā•— - green - ā•”ā• Apple[crisp] ā•ā•— - yes - ā•”ā• Orange ā•ā•— - Orange - """.strip() +""" +ā•”ā• Apple ā•ā•— +Apple +ā•”ā• Apple[color] ā•ā•— +green +ā•”ā• Apple[crisp] ā•ā•— +yes +ā•”ā• Orange ā•ā•— +Orange +""".strip() ) ) assert reader.peek_key() == "Apple" From 0e340ed6015a70da25697e3bd000c1f508e62923 Mon Sep 17 00:00:00 2001 From: Edwin Date: Thu, 28 Mar 2024 16:59:07 -0700 Subject: [PATCH 3/7] Fix the method overloading issues in python --- python/selfie-lib/selfie_lib/Snapshot.py | 15 +++++++++------ python/selfie-lib/selfie_lib/SnapshotValue.py | 14 ++++++++------ 2 files changed, 17 insertions(+), 12 deletions(-) diff --git a/python/selfie-lib/selfie_lib/Snapshot.py b/python/selfie-lib/selfie_lib/Snapshot.py index c4de20ac..60192841 100644 --- a/python/selfie-lib/selfie_lib/Snapshot.py +++ b/python/selfie-lib/selfie_lib/Snapshot.py @@ -51,12 +51,15 @@ def __str__(self): return f"[{self._subject} {self._facet_data}]" @staticmethod - def of(binary): - return Snapshot(SnapshotValue.of(binary), {}) - - @staticmethod - def of(string): - return Snapshot(SnapshotValue.of(string), {}) + def of(data): + if isinstance(data, bytes): + # Handling binary data + return Snapshot(SnapshotValue.of(data), {}) + elif isinstance(data, str): + # Handling string data + return Snapshot(SnapshotValue.of(data), {}) + else: + raise TypeError("Data must be either binary or string") @staticmethod def of_entries(entries): diff --git a/python/selfie-lib/selfie_lib/SnapshotValue.py b/python/selfie-lib/selfie_lib/SnapshotValue.py index 84b7fca9..2776da5a 100644 --- a/python/selfie-lib/selfie_lib/SnapshotValue.py +++ b/python/selfie-lib/selfie_lib/SnapshotValue.py @@ -20,13 +20,15 @@ def value_string(self) -> str: pass @staticmethod - def of(value: Union[bytes, str]) -> "SnapshotValue": - if isinstance(value, bytes): - return SnapshotValueBinary(value) - elif isinstance(value, str): - return SnapshotValueString(unix_newlines(value)) + def of(cls, data): + if isinstance(data, bytes): + return cls(SnapshotValue.of(data), {}) + elif isinstance(data, str): + return cls(SnapshotValue.of(data), {}) + elif isinstance(data, SnapshotValue): + return cls(data, {}) else: - raise TypeError("Value must be either bytes or str") + raise TypeError("Unsupported type for Snapshot creation") class SnapshotValueBinary(SnapshotValue): From 43ad165d6b1450c8d64b6f0a9b77fb1b812075cd Mon Sep 17 00:00:00 2001 From: Edwin Date: Fri, 29 Mar 2024 14:26:52 -0700 Subject: [PATCH 4/7] SnapshotReader passing all the test --- python/selfie-lib/selfie_lib/ArrayMap.py | 6 +- .../selfie_lib/ConvertToWindowsNewlines.py | 4 +- python/selfie-lib/selfie_lib/Snapshot.py | 25 +- python/selfie-lib/selfie_lib/SnapshotFile.py | 255 +++++++++--------- .../selfie-lib/selfie_lib/SnapshotReader.py | 19 +- python/selfie-lib/selfie_lib/SnapshotValue.py | 56 +++- .../selfie_lib/SnapshotValueReader.py | 2 + python/selfie-lib/selfie_lib/__init__.py | 1 - .../selfie-lib/tests/SnapshotReader_test.py | 31 ++- 9 files changed, 242 insertions(+), 157 deletions(-) diff --git a/python/selfie-lib/selfie_lib/ArrayMap.py b/python/selfie-lib/selfie_lib/ArrayMap.py index 49d317b1..485fb296 100644 --- a/python/selfie-lib/selfie_lib/ArrayMap.py +++ b/python/selfie-lib/selfie_lib/ArrayMap.py @@ -9,10 +9,12 @@ class ListBackedSet(Set[T], ABC): @abstractmethod - def __len__(self) -> int: ... + def __len__(self) -> int: + ... @abstractmethod - def __getitem__(self, index: Union[int, slice]) -> Union[T, List[T]]: ... + def __getitem__(self, index: Union[int, slice]) -> Union[T, List[T]]: + ... def __contains__(self, item: object) -> bool: for i in range(len(self)): diff --git a/python/selfie-lib/selfie_lib/ConvertToWindowsNewlines.py b/python/selfie-lib/selfie_lib/ConvertToWindowsNewlines.py index f4615d6f..5718a2b9 100644 --- a/python/selfie-lib/selfie_lib/ConvertToWindowsNewlines.py +++ b/python/selfie-lib/selfie_lib/ConvertToWindowsNewlines.py @@ -5,10 +5,10 @@ def __init__(self, sink): def append(self, value, start_index=None, end_index=None): # If value is a single character if isinstance(value, str) and len(value) == 1: - if value != '\n': + if value != "\n": self.sink.write(value) else: - self.sink.write('\r\n') + self.sink.write("\r\n") # If value is a CharSequence (in Python, a str) elif isinstance(value, str): # If start_index and end_index are provided, use the slice of the string diff --git a/python/selfie-lib/selfie_lib/Snapshot.py b/python/selfie-lib/selfie_lib/Snapshot.py index 60192841..2a551b6b 100644 --- a/python/selfie-lib/selfie_lib/Snapshot.py +++ b/python/selfie-lib/selfie_lib/Snapshot.py @@ -1,5 +1,10 @@ from .SnapshotValue import SnapshotValue from collections import OrderedDict +import logging + +logging.basicConfig(level=logging.DEBUG) +logger = logging.getLogger(__name__) + class Snapshot: def __init__(self, subject, facet_data): @@ -10,6 +15,14 @@ def __init__(self, subject, facet_data): def facets(self): return OrderedDict(sorted(self._facet_data.items())) + def __eq__(self, other): + if not isinstance(other, Snapshot): + return NotImplemented + return self._subject == other._subject and self._facet_data == other._facet_data + + def __hash__(self): + return hash((self._subject, frozenset(self._facet_data.items()))) + def plus_facet(self, key, value): if isinstance(value, bytes): value = SnapshotValue.of(value) @@ -47,7 +60,7 @@ def all_entries(self): entries.extend(self._facet_data.items()) return entries - def __str__(self): + def __bytes__(self): return f"[{self._subject} {self._facet_data}]" @staticmethod @@ -58,8 +71,10 @@ def of(data): elif isinstance(data, str): # Handling string data return Snapshot(SnapshotValue.of(data), {}) + elif isinstance(data, SnapshotValue): + return Snapshot(data, {}) else: - raise TypeError("Data must be either binary or string") + raise TypeError("Data must be either binary or string" + data) @staticmethod def of_entries(entries): @@ -68,7 +83,9 @@ def of_entries(entries): for key, value in entries: if not key: if subject is not None: - raise ValueError(f"Duplicate root snapshot.\n first: {subject}\nsecond: {value}") + raise ValueError( + f"Duplicate root snapshot.\n first: {subject}\nsecond: {value}" + ) subject = value else: facet_data[key] = value @@ -78,4 +95,4 @@ def of_entries(entries): @staticmethod def _unix_newlines(string): - return string.replace("\\r\\n", "\\n") \ No newline at end of file + return string.replace("\\r\\n", "\\n") diff --git a/python/selfie-lib/selfie_lib/SnapshotFile.py b/python/selfie-lib/selfie_lib/SnapshotFile.py index 3e70d455..dda89c0a 100644 --- a/python/selfie-lib/selfie_lib/SnapshotFile.py +++ b/python/selfie-lib/selfie_lib/SnapshotFile.py @@ -1,121 +1,134 @@ -import threading -from typing import List -import base64 - -from .SnapshotValue import SnapshotValue -from .ConvertToWindowsNewlines import ConvertToWindowsNewlines -from .ParseException import ParseException -from .SnapshotReader import SnapshotReader -from .SnapshotValueReader import SnapshotValueReader - -class SnapshotFile: - HEADER_PREFIX = "šŸ“· " - END_OF_FILE = "[end of file]" - - def __init__(self): - self.unix_newlines = True - self.metadata = None - self._snapshots = {} - self._lock = threading.Lock() - self._was_set_at_test_time = False - - @property - def snapshots(self): - return self._snapshots - - @snapshots.setter - def snapshots(self, value): - with self._lock: - self._snapshots = value - - @property - def was_set_at_test_time(self): - return self._was_set_at_test_time - - def set_at_test_time(self, key, snapshot): - with self._lock: - old_snapshots = self._snapshots.copy() - self._snapshots[key] = snapshot - self._was_set_at_test_time = True if self._snapshots != old_snapshots else self._was_set_at_test_time - - def serialize(self, value_writer_raw): - value_writer = value_writer_raw if self.unix_newlines else ConvertToWindowsNewlines(value_writer_raw) - if self.metadata: - self.write_entry(value_writer, f"šŸ“· {self.metadata[0]}", None, SnapshotValue.of(self.metadata[1])) - for key, snapshot in self._snapshots.items(): - self.write_entry(value_writer, key, None, snapshot.subject) - for facet_key, facet_value in snapshot.facets.items(): - self.write_entry(value_writer, key, facet_key, facet_value) - self.write_entry(value_writer, "", "end of file", SnapshotValue.of("")) - - def write_entry(value_writer, key, facet, value): - value_writer.write("ā•”ā• ") - value_writer.write(SnapshotValueReader.nameEsc.escape(key)) - if facet is not None: - value_writer.write("[") - value_writer.write(SnapshotValueReader.nameEsc.escape(facet)) - value_writer.write("]") - value_writer.write(" ā•ā•—") - if value.is_binary: - binary_length = len(value.value_binary()) - value_writer.write(f" base64 length {binary_length} bytes") - value_writer.write("\n") - - if key == "" and facet == "end of file": - return - - if value.is_binary: - # Base64 encoding and replacing \r with an empty string - binary_data = value.value_binary() - encoded = base64.b64encode(binary_data).decode('utf-8') - # Assuming efficientReplace is a more efficient method for replacing characters - # Here, we just use the regular replace method for simplicity - escaped = encoded.replace("\r", "") - value_writer.write(escaped) - else: - # For string values, applying specific escape logic and then replacing "\nā•”" with a special sequence - text_data = value.value_string() - escaped = SnapshotValueReader.bodyEsc(text_data).replace("\nā•”", "\n\uDF41") - value_writer.write(escaped) - value_writer.write("\n") - - @staticmethod - def parse(value_reader): - try: - result = SnapshotFile() - result.unix_newlines = value_reader.unix_newlines - reader = SnapshotReader(value_reader) - - # Check if the first value starts with šŸ“· - if reader.peek_key() and reader.peek_key().startswith(SnapshotFile.HEADER_PREFIX): - metadata_name = reader.peek_key()[len(SnapshotFile.HEADER_PREFIX):] - metadata_value = reader.value_reader.next_value().value_string() - # Assuming 'entry' function creates a dictionary entry in Python - result.metadata = (metadata_name, metadata_value) - - while reader.peek_key() is not None: - key = reader.peek_key() - snapshot = reader.next_snapshot() - # Update snapshots dictionary with new key-value pair - result.snapshots.update({key: snapshot}) - - return result - - except ValueError as e: - if isinstance(e, ParseException): - raise e - else: - raise ParseException(value_reader.line_reader, e) from None - - - @staticmethod - def create_empty_with_unix_newlines(unix_newlines): - result = SnapshotFile() - result.unix_newlines = unix_newlines - return result - - def remove_all_indices(self, indices: List[int]): - if not indices: - return - self._was_set_at_test_time = True - self.snapshots = self.snapshots.minus_sorted_indices(indices) +# import threading +# from typing import List +# import base64 + +# from .SnapshotValue import SnapshotValue +# from .ConvertToWindowsNewlines import ConvertToWindowsNewlines +# from .ParseException import ParseException +# from .SnapshotReader import SnapshotReader +# from .SnapshotValueReader import SnapshotValueReader + + +# class SnapshotFile: +# HEADER_PREFIX = "šŸ“· " +# END_OF_FILE = "[end of file]" + +# def __init__(self): +# self.unix_newlines = True +# self.metadata = None +# self._snapshots = {} +# self._lock = threading.Lock() +# self._was_set_at_test_time = False + +# @property +# def snapshots(self): +# return self._snapshots + +# @snapshots.setter +# def snapshots(self, value): +# with self._lock: +# self._snapshots = value + +# @property +# def was_set_at_test_time(self): +# return self._was_set_at_test_time + +# def set_at_test_time(self, key, snapshot): +# with self._lock: +# old_snapshots = self._snapshots.copy() +# self._snapshots[key] = snapshot +# self._was_set_at_test_time = ( +# True if self._snapshots != old_snapshots else self._was_set_at_test_time +# ) + +# def serialize(self, value_writer_raw): +# value_writer = ( +# value_writer_raw +# if self.unix_newlines +# else ConvertToWindowsNewlines(value_writer_raw) +# ) +# if self.metadata: +# self.write_entry( +# value_writer, +# f"šŸ“· {self.metadata[0]}", +# None, +# SnapshotValue.of(self.metadata[1]), +# ) +# for key, snapshot in self._snapshots.items(): +# self.write_entry(value_writer, key, None, snapshot.subject) +# for facet_key, facet_value in snapshot.facets.items(): +# self.write_entry(value_writer, key, facet_key, facet_value) +# self.write_entry(value_writer, "", "end of file", SnapshotValue.of("")) + +# def write_entry(value_writer, key, facet, value): +# value_writer.write("ā•”ā• ") +# value_writer.write(SnapshotValueReader.nameEsc.escape(key)) +# if facet is not None: +# value_writer.write("[") +# value_writer.write(SnapshotValueReader.nameEsc.escape(facet)) +# value_writer.write("]") +# value_writer.write(" ā•ā•—") +# if value.is_binary: +# binary_length = len(value.value_binary()) +# value_writer.write(f" base64 length {binary_length} bytes") +# value_writer.write("\n") + +# if key == "" and facet == "end of file": +# return + +# if value.is_binary: +# # Base64 encoding and replacing \r with an empty string +# binary_data = value.value_binary() +# encoded = base64.b64encode(binary_data).decode("utf-8") +# # Assuming efficientReplace is a more efficient method for replacing characters +# # Here, we just use the regular replace method for simplicity +# escaped = encoded.replace("\r", "") +# value_writer.write(escaped) +# else: +# # For string values, applying specific escape logic and then replacing "\nā•”" with a special sequence +# text_data = value.value_string() +# escaped = SnapshotValueReader.bodyEsc(text_data).replace("\nā•”", "\n\uDF41") +# value_writer.write(escaped) +# value_writer.write("\n") + +# @staticmethod +# def parse(value_reader): +# try: +# result = SnapshotFile() +# result.unix_newlines = value_reader.unix_newlines +# reader = SnapshotReader(value_reader) + +# # Check if the first value starts with šŸ“· +# if reader.peek_key() and reader.peek_key().startswith( +# SnapshotFile.HEADER_PREFIX +# ): +# metadata_name = reader.peek_key()[len(SnapshotFile.HEADER_PREFIX) :] +# metadata_value = reader.value_reader.next_value().value_string() +# # Assuming 'entry' function creates a dictionary entry in Python +# result.metadata = (metadata_name, metadata_value) + +# while reader.peek_key() is not None: +# key = reader.peek_key() +# snapshot = reader.next_snapshot() +# # Update snapshots dictionary with new key-value pair +# result.snapshots.update({key: snapshot}) + +# return result + +# except ValueError as e: +# if isinstance(e, ParseException): +# raise e +# else: +# raise ParseException(value_reader.line_reader, e) from None + +# @staticmethod +# def create_empty_with_unix_newlines(unix_newlines): +# result = SnapshotFile() +# result.unix_newlines = unix_newlines +# return result + +# def remove_all_indices(self, indices: List[int]): +# if not indices: +# return +# self._was_set_at_test_time = True +# self.snapshots = self.snapshots.minus_sorted_indices(indices) diff --git a/python/selfie-lib/selfie_lib/SnapshotReader.py b/python/selfie-lib/selfie_lib/SnapshotReader.py index 0eaee273..4271dac0 100644 --- a/python/selfie-lib/selfie_lib/SnapshotReader.py +++ b/python/selfie-lib/selfie_lib/SnapshotReader.py @@ -1,5 +1,6 @@ from .Snapshot import Snapshot + class SnapshotReader: def __init__(self, value_reader): self.value_reader = value_reader @@ -8,8 +9,10 @@ def peek_key(self): next_key = self.value_reader.peek_key() if next_key is None or next_key == "[end of file]": return None - if '[' in next_key: - raise ValueError(f"Missing root snapshot, square brackets not allowed: '{next_key}'") + if "[" in next_key: + raise ValueError( + f"Missing root snapshot, square brackets not allowed: '{next_key}'" + ) return next_key def next_snapshot(self): @@ -19,18 +22,20 @@ def next_snapshot(self): next_key = self.value_reader.peek_key() if next_key is None: return snapshot - facet_idx = next_key.find('[') + facet_idx = next_key.find("[") if facet_idx == -1 or (facet_idx == 0 and next_key == "[end of file]"): return snapshot facet_root = next_key[:facet_idx] if facet_root != root_name: - raise ValueError(f"Expected '{next_key}' to come after '{facet_root}', not '{root_name}'") - facet_end_idx = next_key.find(']', facet_idx + 1) + raise ValueError( + f"Expected '{next_key}' to come after '{facet_root}', not '{root_name}'" + ) + facet_end_idx = next_key.find("]", facet_idx + 1) if facet_end_idx == -1: raise ValueError(f"Missing ] in {next_key}") - facet_name = next_key[facet_idx + 1:facet_end_idx] + facet_name = next_key[facet_idx + 1 : facet_end_idx] snapshot = snapshot.plus_facet(facet_name, self.value_reader.next_value()) - + def skip_snapshot(self): root_name = self.peek_key() if root_name is None: diff --git a/python/selfie-lib/selfie_lib/SnapshotValue.py b/python/selfie-lib/selfie_lib/SnapshotValue.py index 2776da5a..665734e8 100644 --- a/python/selfie-lib/selfie_lib/SnapshotValue.py +++ b/python/selfie-lib/selfie_lib/SnapshotValue.py @@ -3,9 +3,11 @@ # from .SnapshotValueBinary import SnapshotValueBinary # from .SnapshotValueString import SnapshotValueString + def unix_newlines(string: str) -> str: return string.replace("\r\n", "\n") - + + class SnapshotValue(ABC): @property def is_binary(self) -> bool: @@ -20,16 +22,16 @@ def value_string(self) -> str: pass @staticmethod - def of(cls, data): + def of(data): if isinstance(data, bytes): - return cls(SnapshotValue.of(data), {}) + return SnapshotValueBinary(data) elif isinstance(data, str): - return cls(SnapshotValue.of(data), {}) + return SnapshotValueString(data) elif isinstance(data, SnapshotValue): - return cls(data, {}) + return data else: raise TypeError("Unsupported type for Snapshot creation") - + class SnapshotValueBinary(SnapshotValue): def __init__(self, value: bytes): @@ -40,7 +42,26 @@ def value_binary(self) -> bytes: def value_string(self) -> str: raise NotImplementedError("This is a binary value.") - + + # def __bytes__(self): + # return self._value + + # def __repr__ (self): + # return self.__bytes__() + + # def __eq__(self, other): + # if not isinstance(other, SnapshotValueBinary): + # return False + # return self._value == other._value + + def __eq__(self, other): + if isinstance(other, SnapshotValueBinary): + return self.value_binary() == other.value_binary() + return False + + def __hash__(self): + return hash(self._value) + class SnapshotValueString(SnapshotValue): def __init__(self, value: str): @@ -50,4 +71,23 @@ def value_binary(self) -> bytes: raise NotImplementedError("This is a string value.") def value_string(self) -> str: - return self._value \ No newline at end of file + return self._value + + # def __str__(self): + # return self._value + + # def __repr__ (self): + # return self.__str__() + + # def __eq__(self, other): + # if not isinstance(other, SnapshotValueString): + # return False + # return self._value == other._value + + def __eq__(self, other): + if isinstance(other, SnapshotValueString): + return self.value_string() == other.value_string() + return False + + def __hash__(self): + return hash(self._value) diff --git a/python/selfie-lib/selfie_lib/SnapshotValueReader.py b/python/selfie-lib/selfie_lib/SnapshotValueReader.py index b3731da3..4f937619 100644 --- a/python/selfie-lib/selfie_lib/SnapshotValueReader.py +++ b/python/selfie-lib/selfie_lib/SnapshotValueReader.py @@ -4,9 +4,11 @@ from .LineReader import LineReader from .SnapshotValue import SnapshotValue + def unix_newlines(string: str) -> str: return string.replace("\r\n", "\n") + class SnapshotValueReader: KEY_FIRST_CHAR = "ā•”" KEY_START = "ā•”ā• " diff --git a/python/selfie-lib/selfie_lib/__init__.py b/python/selfie-lib/selfie_lib/__init__.py index 354dda2b..3204174f 100644 --- a/python/selfie-lib/selfie_lib/__init__.py +++ b/python/selfie-lib/selfie_lib/__init__.py @@ -7,4 +7,3 @@ from .SnapshotReader import SnapshotReader as SnapshotReader from .Snapshot import Snapshot as Snapshot from .SnapshotValue import SnapshotValue as SnapshotValue - diff --git a/python/selfie-lib/tests/SnapshotReader_test.py b/python/selfie-lib/tests/SnapshotReader_test.py index 19386047..236c6a2e 100644 --- a/python/selfie-lib/tests/SnapshotReader_test.py +++ b/python/selfie-lib/tests/SnapshotReader_test.py @@ -1,11 +1,12 @@ from base64 import b64decode from selfie_lib import SnapshotValueReader, Snapshot, SnapshotReader + class TestSnapshotReader: def test_facet(self): reader = SnapshotReader( SnapshotValueReader.of( -""" + """ ā•”ā• Apple ā•ā•— Apple ā•”ā• Apple[color] ā•ā•— @@ -19,7 +20,9 @@ def test_facet(self): ) assert reader.peek_key() == "Apple" assert reader.peek_key() == "Apple" - apple_snapshot = Snapshot.of("Apple").plus_facet("color", "green").plus_facet("crisp", "yes") + apple_snapshot = ( + Snapshot.of("Apple").plus_facet("color", "green").plus_facet("crisp", "yes") + ) assert reader.next_snapshot() == apple_snapshot assert reader.peek_key() == "Orange" assert reader.peek_key() == "Orange" @@ -30,20 +33,24 @@ def test_binary(self): reader = SnapshotReader( SnapshotValueReader.of( """ - ā•”ā• Apple ā•ā•— - Apple - ā•”ā• Apple[color] ā•ā•— base64 length 3 bytes - c2Fk - ā•”ā• Apple[crisp] ā•ā•— - yes - ā•”ā• Orange ā•ā•— base64 length 3 bytes - c2Fk - """.strip() +ā•”ā• Apple ā•ā•— +Apple +ā•”ā• Apple[color] ā•ā•— base64 length 3 bytes +c2Fk +ā•”ā• Apple[crisp] ā•ā•— +yes +ā•”ā• Orange ā•ā•— base64 length 3 bytes +c2Fk +""".strip() ) ) assert reader.peek_key() == "Apple" assert reader.peek_key() == "Apple" - apple_snapshot = Snapshot.of("Apple").plus_facet("color", b64decode("c2Fk")).plus_facet("crisp", "yes") + apple_snapshot = ( + Snapshot.of("Apple") + .plus_facet("color", b64decode("c2Fk")) + .plus_facet("crisp", "yes") + ) assert reader.next_snapshot() == apple_snapshot assert reader.peek_key() == "Orange" assert reader.peek_key() == "Orange" From 9f026f12a53794a09067c1daa5a3bdbd33a05afa Mon Sep 17 00:00:00 2001 From: Edwin Date: Fri, 29 Mar 2024 14:29:23 -0700 Subject: [PATCH 5/7] Fixing some format issues --- python/selfie-lib/selfie_lib/Snapshot.py | 4 ---- 1 file changed, 4 deletions(-) diff --git a/python/selfie-lib/selfie_lib/Snapshot.py b/python/selfie-lib/selfie_lib/Snapshot.py index 2a551b6b..9d22057a 100644 --- a/python/selfie-lib/selfie_lib/Snapshot.py +++ b/python/selfie-lib/selfie_lib/Snapshot.py @@ -1,9 +1,5 @@ from .SnapshotValue import SnapshotValue from collections import OrderedDict -import logging - -logging.basicConfig(level=logging.DEBUG) -logger = logging.getLogger(__name__) class Snapshot: From d4da1f6ac2b271ae214601923a89db779ace5009 Mon Sep 17 00:00:00 2001 From: Edwin Date: Fri, 29 Mar 2024 14:32:34 -0700 Subject: [PATCH 6/7] fix format issues --- python/selfie-lib/selfie_lib/SnapshotFile.py | 134 ------------------ python/selfie-lib/selfie_lib/SnapshotValue.py | 25 ---- 2 files changed, 159 deletions(-) delete mode 100644 python/selfie-lib/selfie_lib/SnapshotFile.py diff --git a/python/selfie-lib/selfie_lib/SnapshotFile.py b/python/selfie-lib/selfie_lib/SnapshotFile.py deleted file mode 100644 index dda89c0a..00000000 --- a/python/selfie-lib/selfie_lib/SnapshotFile.py +++ /dev/null @@ -1,134 +0,0 @@ -# import threading -# from typing import List -# import base64 - -# from .SnapshotValue import SnapshotValue -# from .ConvertToWindowsNewlines import ConvertToWindowsNewlines -# from .ParseException import ParseException -# from .SnapshotReader import SnapshotReader -# from .SnapshotValueReader import SnapshotValueReader - - -# class SnapshotFile: -# HEADER_PREFIX = "šŸ“· " -# END_OF_FILE = "[end of file]" - -# def __init__(self): -# self.unix_newlines = True -# self.metadata = None -# self._snapshots = {} -# self._lock = threading.Lock() -# self._was_set_at_test_time = False - -# @property -# def snapshots(self): -# return self._snapshots - -# @snapshots.setter -# def snapshots(self, value): -# with self._lock: -# self._snapshots = value - -# @property -# def was_set_at_test_time(self): -# return self._was_set_at_test_time - -# def set_at_test_time(self, key, snapshot): -# with self._lock: -# old_snapshots = self._snapshots.copy() -# self._snapshots[key] = snapshot -# self._was_set_at_test_time = ( -# True if self._snapshots != old_snapshots else self._was_set_at_test_time -# ) - -# def serialize(self, value_writer_raw): -# value_writer = ( -# value_writer_raw -# if self.unix_newlines -# else ConvertToWindowsNewlines(value_writer_raw) -# ) -# if self.metadata: -# self.write_entry( -# value_writer, -# f"šŸ“· {self.metadata[0]}", -# None, -# SnapshotValue.of(self.metadata[1]), -# ) -# for key, snapshot in self._snapshots.items(): -# self.write_entry(value_writer, key, None, snapshot.subject) -# for facet_key, facet_value in snapshot.facets.items(): -# self.write_entry(value_writer, key, facet_key, facet_value) -# self.write_entry(value_writer, "", "end of file", SnapshotValue.of("")) - -# def write_entry(value_writer, key, facet, value): -# value_writer.write("ā•”ā• ") -# value_writer.write(SnapshotValueReader.nameEsc.escape(key)) -# if facet is not None: -# value_writer.write("[") -# value_writer.write(SnapshotValueReader.nameEsc.escape(facet)) -# value_writer.write("]") -# value_writer.write(" ā•ā•—") -# if value.is_binary: -# binary_length = len(value.value_binary()) -# value_writer.write(f" base64 length {binary_length} bytes") -# value_writer.write("\n") - -# if key == "" and facet == "end of file": -# return - -# if value.is_binary: -# # Base64 encoding and replacing \r with an empty string -# binary_data = value.value_binary() -# encoded = base64.b64encode(binary_data).decode("utf-8") -# # Assuming efficientReplace is a more efficient method for replacing characters -# # Here, we just use the regular replace method for simplicity -# escaped = encoded.replace("\r", "") -# value_writer.write(escaped) -# else: -# # For string values, applying specific escape logic and then replacing "\nā•”" with a special sequence -# text_data = value.value_string() -# escaped = SnapshotValueReader.bodyEsc(text_data).replace("\nā•”", "\n\uDF41") -# value_writer.write(escaped) -# value_writer.write("\n") - -# @staticmethod -# def parse(value_reader): -# try: -# result = SnapshotFile() -# result.unix_newlines = value_reader.unix_newlines -# reader = SnapshotReader(value_reader) - -# # Check if the first value starts with šŸ“· -# if reader.peek_key() and reader.peek_key().startswith( -# SnapshotFile.HEADER_PREFIX -# ): -# metadata_name = reader.peek_key()[len(SnapshotFile.HEADER_PREFIX) :] -# metadata_value = reader.value_reader.next_value().value_string() -# # Assuming 'entry' function creates a dictionary entry in Python -# result.metadata = (metadata_name, metadata_value) - -# while reader.peek_key() is not None: -# key = reader.peek_key() -# snapshot = reader.next_snapshot() -# # Update snapshots dictionary with new key-value pair -# result.snapshots.update({key: snapshot}) - -# return result - -# except ValueError as e: -# if isinstance(e, ParseException): -# raise e -# else: -# raise ParseException(value_reader.line_reader, e) from None - -# @staticmethod -# def create_empty_with_unix_newlines(unix_newlines): -# result = SnapshotFile() -# result.unix_newlines = unix_newlines -# return result - -# def remove_all_indices(self, indices: List[int]): -# if not indices: -# return -# self._was_set_at_test_time = True -# self.snapshots = self.snapshots.minus_sorted_indices(indices) diff --git a/python/selfie-lib/selfie_lib/SnapshotValue.py b/python/selfie-lib/selfie_lib/SnapshotValue.py index 665734e8..95f72df7 100644 --- a/python/selfie-lib/selfie_lib/SnapshotValue.py +++ b/python/selfie-lib/selfie_lib/SnapshotValue.py @@ -1,7 +1,4 @@ from abc import ABC, abstractmethod -from typing import Union -# from .SnapshotValueBinary import SnapshotValueBinary -# from .SnapshotValueString import SnapshotValueString def unix_newlines(string: str) -> str: @@ -43,17 +40,6 @@ def value_binary(self) -> bytes: def value_string(self) -> str: raise NotImplementedError("This is a binary value.") - # def __bytes__(self): - # return self._value - - # def __repr__ (self): - # return self.__bytes__() - - # def __eq__(self, other): - # if not isinstance(other, SnapshotValueBinary): - # return False - # return self._value == other._value - def __eq__(self, other): if isinstance(other, SnapshotValueBinary): return self.value_binary() == other.value_binary() @@ -73,17 +59,6 @@ def value_binary(self) -> bytes: def value_string(self) -> str: return self._value - # def __str__(self): - # return self._value - - # def __repr__ (self): - # return self.__str__() - - # def __eq__(self, other): - # if not isinstance(other, SnapshotValueString): - # return False - # return self._value == other._value - def __eq__(self, other): if isinstance(other, SnapshotValueString): return self.value_string() == other.value_string() From 639f36ddfbd49a69ca0479b664066e996495ec7d Mon Sep 17 00:00:00 2001 From: Ned Twigg Date: Tue, 2 Apr 2024 10:56:10 -0700 Subject: [PATCH 7/7] `poetry run ruff format` --- python/selfie-lib/selfie_lib/ArrayMap.py | 6 ++---- 1 file changed, 2 insertions(+), 4 deletions(-) diff --git a/python/selfie-lib/selfie_lib/ArrayMap.py b/python/selfie-lib/selfie_lib/ArrayMap.py index 485fb296..49d317b1 100644 --- a/python/selfie-lib/selfie_lib/ArrayMap.py +++ b/python/selfie-lib/selfie_lib/ArrayMap.py @@ -9,12 +9,10 @@ class ListBackedSet(Set[T], ABC): @abstractmethod - def __len__(self) -> int: - ... + def __len__(self) -> int: ... @abstractmethod - def __getitem__(self, index: Union[int, slice]) -> Union[T, List[T]]: - ... + def __getitem__(self, index: Union[int, slice]) -> Union[T, List[T]]: ... def __contains__(self, item: object) -> bool: for i in range(len(self)):