python: add lens and camera (#426)

Author: nedtwigg

python/example-pytest-selfie/tests/cache_selfie_test.py

@@ -3,5 +3,10 @@
 from selfie_lib import cache_selfie
 
 
+def random_str() -> str:
+    return str(random.random())
+
+
 def test_cache_selfie():
-    cache_selfie(lambda: str(random.random())).to_be("0.6623096709843852")
+    cache_selfie(lambda: str(random.random())).to_be("0.46009462251400757")
+    cache_selfie(random_str).to_be("0.6134874512330031")

python/example-pytest-selfie/tests/get_started_test.py

@@ -0,0 +1,5 @@
+from selfie_lib import expect_selfie
+
+
+def test_quickstart():
+    expect_selfie([1, 2, 3]).to_be([1, 2, 3])

python/fullstack-pytest.code-workspace

@@ -8,6 +8,12 @@
 		},
 		{
 			"path": "selfie-lib"
+		},
+		{
+			"path": "../.github/workflows"
+		},
+		{
+			"path": "../selfie.dev"
 		}
 	],
 	"settings": {}

python/selfie-lib/selfie_lib/CacheSelfie.py

@@ -1,5 +1,5 @@
 import base64
-from typing import Any, Generic, Optional, Protocol, TypeVar, Union, overload
+from typing import Any, Callable, Generic, Optional, TypeVar, Union, overload
 
 from .Literals import LiteralString, LiteralValue, TodoStub
 from .Roundtrip import Roundtrip
@@ -10,24 +10,18 @@
 T = TypeVar("T", covariant=True)
 
 
-class Cacheable(Protocol[T]):
-    def __call__(self) -> T:
-        """Method to get the cached object."""
-        raise NotImplementedError
-
-
 @overload
-def cache_selfie(to_cache: Cacheable[str]) -> "CacheSelfie[str]": ...
+def cache_selfie(to_cache: Callable[..., str]) -> "CacheSelfie[str]": ...
 
 
 @overload
 def cache_selfie(
-    to_cache: Cacheable[T], roundtrip: Roundtrip[T, str]
+    to_cache: Callable[..., T], roundtrip: Roundtrip[T, str]
 ) -> "CacheSelfie[T]": ...
 
 
 def cache_selfie(
-    to_cache: Union[Cacheable[str], Cacheable[T]],
+    to_cache: Union[Callable[..., str], Callable[..., T]],
     roundtrip: Optional[Roundtrip[T, str]] = None,
 ) -> Union["CacheSelfie[str]", "CacheSelfie[T]"]:
     if roundtrip is None:
@@ -42,7 +36,10 @@ def cache_selfie(
 
 class CacheSelfie(Generic[T]):
     def __init__(
-        self, disk: DiskStorage, roundtrip: Roundtrip[T, str], generator: Cacheable[T]
+        self,
+        disk: DiskStorage,
+        roundtrip: Roundtrip[T, str],
+        generator: Callable[..., T],
     ):
         self.disk = disk
         self.roundtrip = roundtrip
@@ -110,7 +107,7 @@ def __init__(
         self,
         disk: DiskStorage,
         roundtrip: Roundtrip[T, bytes],
-        generator: Cacheable[T],
+        generator: Callable[..., T],
     ):
         self.disk = disk
         self.roundtrip = roundtrip

python/selfie-lib/selfie_lib/Lens.py

@@ -0,0 +1,93 @@
+import re
+from abc import ABC, abstractmethod
+from typing import Callable, Generic, Iterator, List, Optional, Protocol, TypeVar
+
+from .Snapshot import Snapshot, SnapshotValue
+
+T = TypeVar("T")
+
+
+class Lens(Protocol):
+    def __call__(self, snapshot: Snapshot) -> Snapshot:
+        raise NotImplementedError
+
+
+class CompoundLens(Lens):
+    def __init__(self):
+        self.lenses: List[Lens] = []
+
+    def __call__(self, snapshot: Snapshot) -> Snapshot:
+        current = snapshot
+        for lens in self.lenses:
+            current = lens(current)
+        return current
+
+    def add(self, lens: Lens) -> "CompoundLens":
+        self.lenses.append(lens)
+        return self
+
+    def mutate_all_facets(
+        self, perString: Callable[[str], Optional[str]]
+    ) -> "CompoundLens":
+        def _mutate_each(snapshot: Snapshot) -> Iterator[tuple[str, SnapshotValue]]:
+            for entry in snapshot.items():
+                if entry[1].is_binary:
+                    yield entry
+                else:
+                    mapped = perString(entry[1].value_string())
+                    if mapped is not None:
+                        yield (entry[0], SnapshotValue.of(mapped))
+
+        return self.add(lambda snapshot: Snapshot.of_items(_mutate_each(snapshot)))
+
+    def replace_all(self, toReplace: str, replacement: str) -> "CompoundLens":
+        return self.mutate_all_facets(lambda s: s.replace(toReplace, replacement))
+
+    def replace_all_regex(
+        self, pattern: str | re.Pattern[str], replacement: str
+    ) -> "CompoundLens":
+        return self.mutate_all_facets(lambda s: re.sub(pattern, replacement, s))
+
+    def set_facet_from(
+        self, target: str, source: str, function: Callable[[str], Optional[str]]
+    ) -> "CompoundLens":
+        def _set_facet_from(snapshot: Snapshot) -> Snapshot:
+            source_value = snapshot.subject_or_facet_maybe(source)
+            if source_value is None:
+                return snapshot
+            else:
+                return self.__set_facet_of(
+                    snapshot, target, function(source_value.value_string())
+                )
+
+        return self.add(_set_facet_from)
+
+    def __set_facet_of(
+        self, snapshot: Snapshot, target: str, new_value: Optional[str]
+    ) -> Snapshot:
+        if new_value is None:
+            return snapshot
+        else:
+            return snapshot.plus_or_replace(target, SnapshotValue.of(new_value))
+
+    def mutate_facet(
+        self, target: str, function: Callable[[str], Optional[str]]
+    ) -> "CompoundLens":
+        return self.set_facet_from(target, target, function)
+
+
+class Camera(Generic[T], ABC):
+    @abstractmethod
+    def snapshot(self, subject: T) -> Snapshot:
+        pass
+
+    def with_lens(self, lens: Lens) -> "Camera[T]":
+        class WithLensCamera(Camera):
+            def __init__(self, camera: Camera[T], lens: Callable[[Snapshot], Snapshot]):
+                self.__camera = camera
+                self.__lens = lens
+
+            def snapshot(self, subject: T) -> Snapshot:
+                return self.__lens(self.__camera.snapshot(subject))
+
+        return WithLensCamera(self, lens)

python/selfie-lib/selfie_lib/LineReader.py

@@ -1,6 +1,13 @@
 import io
 
 
+def _to_unix(s: str) -> str:
+    if s.find("\r\n") == -1:
+        return s
+    else:
+        return s.replace("\r\n", "\n")
+
+
 class LineReader:
     def __init__(self, content: bytes):
         self.__buffer = io.BytesIO(content)

python/selfie-lib/selfie_lib/Snapshot.py

@@ -1,6 +1,7 @@
-from typing import Dict, Iterable, Union
+from typing import Iterator, Union
 
 from .ArrayMap import ArrayMap
+from .LineReader import _to_unix
 from .SnapshotValue import SnapshotValue
 
 
@@ -34,8 +35,23 @@ def plus_facet(
     ) -> "Snapshot":
         if key == "":
             raise ValueError("The empty string is reserved for the subject.")
-        new_facet_data = self._facet_data.plus(key, SnapshotValue.of(value))
-        return Snapshot(self._subject, new_facet_data)
+        return Snapshot(
+            self._subject,
+            self._facet_data.plus(_to_unix(key), SnapshotValue.of(value)),
+        )
+
+    def plus_or_replace(
+        self, key: str, value: Union[bytes, str, SnapshotValue]
+    ) -> "Snapshot":
+        if key == "":
+            return Snapshot(SnapshotValue.of(value), self._facet_data)
+        else:
+            return Snapshot(
+                self._subject,
+                self._facet_data.plus_or_noop_or_replace(
+                    _to_unix(key), SnapshotValue.of(value)
+                ),
+            )
 
     def subject_or_facet_maybe(self, key: str) -> Union[SnapshotValue, None]:
         return self._subject if key == "" else self._facet_data.get(key)
@@ -53,19 +69,27 @@ def of(data: Union[bytes, str, SnapshotValue]) -> "Snapshot":
         return Snapshot(data, ArrayMap.empty())
 
     @staticmethod
-    def of_entries(entries: Iterable[Dict[str, SnapshotValue]]) -> "Snapshot":
-        root = None
+    def of_items(items: Iterator[tuple[str, SnapshotValue]]) -> "Snapshot":
+        subject = None
         facets = ArrayMap.empty()
-        for entry in entries:
-            key, value = entry["key"], entry["value"]
+        for entry in items:
+            (key, value) = entry
             if key == "":
-                if root is not None:
-                    raise ValueError("Duplicate root snapshot detected")
-                root = value
+                if subject is not None:
+                    raise ValueError(
+                        "Duplicate root snapshot value.\n   first: ${subject}\n  second: ${value}"
+                    )
+                subject = value
             else:
                 facets = facets.plus(key, value)
-        return Snapshot(root if root else SnapshotValue.of(""), facets)
+        return Snapshot(subject if subject else SnapshotValue.of(""), facets)
 
-    @staticmethod
-    def _unix_newlines(string: str) -> str:
-        return string.replace("\\r\\n", "\\n")
+    def items(self) -> Iterator[tuple[str, SnapshotValue]]:
+        yield ("", self._subject)
+        yield from self._facet_data.items()
+
+    def __repr__(self) -> str:
+        pieces = [f"Snapshot.of({self.subject.value_string()!r})"]
+        for e in self.facets.items():
+            pieces.append(f"\n  .plus_facet({e[0]!r}, {e[1].value_string()!r})")  # noqa: PERF401
+        return "".join(pieces)

python/selfie-lib/selfie_lib/SnapshotSystem.py

@@ -130,8 +130,8 @@ def msg(self, headline: str) -> str:
         if self == Mode.interactive:
             return (
                 f"{headline}\n"
-                "- update this snapshot by adding '_TODO' to the function name\n"
-                "- update all snapshots in this file by adding '# selfieonce' or '# SELFIEWRITE'"
+                "- update this snapshot by adding `_TODO` to the function name\n"
+                "- update all snapshots in this file by adding `#selfieonce` or `#SELFIEWRITE`"
             )
         elif self == Mode.readonly:
             return headline

python/selfie-lib/selfie_lib/SnapshotValue.py

@@ -1,9 +1,7 @@
 from abc import ABC, abstractmethod
 from typing import Union
 
-
-def unix_newlines(string: str) -> str:
-    return string.replace("\r\n", "\n")
+from .LineReader import _to_unix
 
 
 class SnapshotValue(ABC):
@@ -24,7 +22,7 @@ def of(data: Union[bytes, str, "SnapshotValue"]) -> "SnapshotValue":
         if isinstance(data, bytes):
             return SnapshotValueBinary(data)
         elif isinstance(data, str):
-            return SnapshotValueString(data)
+            return SnapshotValueString(_to_unix(data))
         elif isinstance(data, SnapshotValue):
             return data
         else:

python/selfie-lib/selfie_lib/SnapshotValueReader.py

@@ -7,10 +7,6 @@
 from .SnapshotValue import SnapshotValue
 
 
-def unix_newlines(string: str) -> str:
-    return string.replace("\r\n", "\n")
-
-
 class SnapshotValueReader:
     KEY_FIRST_CHAR = "╔"
     KEY_START = "╔═ "