diff options
Diffstat (limited to 'python/src')
| -rw-r--r-- | python/src/types/buffer.py | 16 | ||||
| -rw-r--r-- | python/src/types/monad.py | 80 | ||||
| -rw-r--r-- | python/src/types/selection.py | 161 | ||||
| -rw-r--r-- | python/src/types/state.py | 93 |
4 files changed, 350 insertions, 0 deletions
diff --git a/python/src/types/buffer.py b/python/src/types/buffer.py new file mode 100644 index 0000000..04e1def --- /dev/null +++ b/python/src/types/buffer.py @@ -0,0 +1,16 @@ +from dataclasses import dataclass + +@dataclass +class Buffer: + content: str + + @property + def min_pos(self) -> int: + return 0 + + @property + def max_pos(self) -> int: + return len(self.content) + + def valid_pos(self, pos : int) -> bool: + return self.min_pos <= pos < self.max_pos diff --git a/python/src/types/monad.py b/python/src/types/monad.py new file mode 100644 index 0000000..f09ff33 --- /dev/null +++ b/python/src/types/monad.py @@ -0,0 +1,80 @@ +from functools import wraps +from typing import Callable + +from .selection import Selection +from .state import EditorState, Result, StateChange + + +class StateMonad: + def __init__(self, run_fn: Callable[[EditorState], Result]): + self._run_fn = run_fn + + def run(self, state: EditorState) -> Result: + return self._run_fn(state) + + def commit(self, state: EditorState) -> tuple[Selection, EditorState]: + sel, changes = self.run(state) + state.commit_changes(changes) + return (sel, state) + + def bind(self, f: Callable[[Selection], "StateMonad"]) -> "StateMonad": + def run_bound(state: EditorState) -> Result: + sel1, changes1 = self.run(state) + expr2 = f(sel1) + sel2, changes2 = expr2.run(state) + return (sel2, changes1 + changes2) + + return StateMonad(run_bound) + + @staticmethod + def pure(selection: Selection) -> "StateMonad": + def run_pure(_: EditorState) -> Result: + return (selection, []) + + return StateMonad(run_pure) + + def modify(change: StateChange, selection: Selection) -> "StateMonad": + def run_modify(_: EditorState) -> Result: + return (selection, [change]) + + return StateMonad(run_modify) + + +def monadic( + func: Callable[[Selection, EditorState], Result], +) -> Callable[[Selection | StateMonad], StateMonad]: + def make_monad(sel: Selection) -> StateMonad: + def run_fn(state: EditorState) -> Result: + return func(sel, state) + + return StateMonad(run_fn) + + @wraps(func) + def wrapper(arg: Selection | StateMonad) -> StateMonad: + if isinstance(arg, StateMonad): + return arg.bind(make_monad) + else: + return make_monad(arg) + + return wrapper + + +def group(*exprs: StateMonad) -> StateMonad: + def run_grouped(state: EditorState) -> Result: + all_selections = [] + all_changes = [] + + for expr in exprs: + sel, changes = expr.run(state) + all_selections.append(sel) + all_changes.extend(changes) + + # TODO: Check for any overlapping changes + # TODO: Merge selections into union + + return ( + all_selections[0] if all_selections else state.current_selection, + all_changes, + ) + + return StateMonad(run_grouped) diff --git a/python/src/types/selection.py b/python/src/types/selection.py new file mode 100644 index 0000000..478b5b9 --- /dev/null +++ b/python/src/types/selection.py @@ -0,0 +1,161 @@ +from dataclasses import dataclass, field +from typing import Callable + +from .buffer import Buffer + + +@dataclass +class Position: + pos: int + + +@dataclass +class Range: + start: Position + end: Position + + +@dataclass +class Ranges: + ranges: list[Range] + + +@dataclass +class MultiRanges: + multi_ranges: dict[Buffer, Ranges] + + +@dataclass +class Selection: + selection: Position | Range | Ranges | MultiRanges + buffers: set[Buffer] = field(default_factory=set) + + def __post_init__(self): + if type(self.selection) is MultiRanges: + self.buffers = set(self.selection.multi_ranges) + else: + assert len(self.buffers) == 1 + + @classmethod + def empty(cls) -> "Selection": + return cls(selection=Position(0), buffers=set()) + + def promote(self, to_rank: int) -> "Selection": + if to_rank < 0 or to_rank > 3: + raise ValueError(f"Invalid rank: {to_rank}") + if self.rank > to_rank: + raise ValueError( + f"Cannot coerce selection of rank {self.rank} to rank {to_rank}" + ) + if self.rank == to_rank: + return self + match self.selection: + case Position(_): + as_range = Range( + start=Position(self.selection.pos), + end=Position(self.selection.pos + 1), + ) + as_ranges = Ranges(ranges=[as_range]) + if to_rank == 1: + return Selection(selection=as_range, buffers=self.buffers) + if to_rank == 2: + return Selection(selection=as_ranges, buffers=self.buffers) + if to_rank == 3: + return Selection( + MultiRanges({buffer: as_ranges for buffer in self.buffers}) + ) + case Range(_, _): + as_ranges = Ranges(ranges=[self.selection]) + if to_rank == 2: + return Selection(selection=as_ranges, buffers=self.buffers) + if to_rank == 3: + return Selection( + MultiRanges({buffer: as_ranges for buffer in self.buffers}) + ) + case Ranges(_): + return Selection( + MultiRanges({buffer: self.selection for buffer in self.buffers}) + ) + + raise RuntimeError("Unexpected promotion issue") + + @classmethod + def union(cls, *selections: "Selection") -> "Selection": + if not selections: + return cls.empty() + + all_buffers = set.union(*[s.buffers for s in selections]) + + if len(all_buffers) == 0: + return cls.empty() + + if len(all_buffers) > 1: + max_rank = 3 + else: + max_rank = max(s.rank for s in selections) + if max_rank < 2: + max_rank = 2 + + promoted = [s.promote(max_rank) for s in selections] + + if max_rank == 2: + all_ranges = [] + for sel in promoted: + if not type(sel.selection) is Ranges: + raise ValueError( + f"Selection of rank {sel.rank} is not a range. Should have been promoted to rank 2." + ) + all_ranges.extend(sel.selection.ranges) + return cls(selection=Ranges(all_ranges), buffers=all_buffers) + + if max_rank == 3: + buffer_ranges: dict[Buffer, list[Range]] = {buf: [] for buf in all_buffers} + for sel in promoted: + if not type(sel.selection) is MultiRanges: + raise ValueError( + f"Selection of rank {sel.rank} is not a multi-range. Should have been promoted to rank 3." + ) + for buffer, ranges in sel.selection.multi_ranges.items(): + buffer_ranges[buffer].extend(ranges.ranges) + + multi = MultiRanges( + {buffer: Ranges(ranges) for buffer, ranges in buffer_ranges.items()} + ) + return cls(selection=multi) + + raise RuntimeError("Unexpected union issue") + + @property + def rank(self) -> int: + match self.selection: + case Position(_): + return 0 + case Range(_, _): + return 1 + case Ranges(_): + return 2 + case MultiRanges(_): + return 3 + + def broadcast( + self, + rank_zero: Callable[[Position, Buffer], "Selection"], + rank_one: Callable[[Range, Buffer], "Selection"], + ) -> "Selection": + match self.selection: + case Position(_): + return rank_zero(self.selection, list(self.buffers)[0]) + + case Range(_, _) as r: + return rank_one(r, list(self.buffers)[0]) + + case Ranges(ranges): + buffer = list(self.buffers)[0] + return Selection.union(*[rank_one(r, buffer) for r in ranges]) + + case MultiRanges(multi): + all_results = [] + for buffer, ranges in multi.items(): + for r in ranges.ranges: + all_results.append(rank_one(r, buffer)) + return Selection.union(*all_results) diff --git a/python/src/types/state.py b/python/src/types/state.py new file mode 100644 index 0000000..ca048aa --- /dev/null +++ b/python/src/types/state.py @@ -0,0 +1,93 @@ +from dataclasses import dataclass +from pathlib import Path + +from .buffer import Buffer +from .selection import Selection + + +class StateChange: + def invert(self) -> "StateChange" | None: + return None + + +@dataclass +class SetCurrentSelection(StateChange): + old_selection: Selection + new_selection: Selection + + def invert(self) -> "SetCurrentSelection": + return SetCurrentSelection( + old_selection=self.new_selection, + new_selection=self.old_selection, + ) + + +@dataclass +class SetCurrentBuffer(StateChange): + old_buffer: Buffer + new_buffer: Buffer + + def invert(self) -> "SetCurrentBuffer": + return SetCurrentBuffer( + old_buffer=self.new_buffer, + new_buffer=self.old_buffer, + ) + + +@dataclass +class ModifyBuffer(StateChange): + buffer: Buffer + old_start: int + old_end: int + old_content: str + new_content: str + + def invert(self) -> None: + # TODO: implement + return None + + +@dataclass +class CreateBuffer(StateChange): + buffer: Buffer + file: Path | None + + +@dataclass +class DeleteBuffer(StateChange): + buffer: Buffer + + +@dataclass +class EditorState: + current_buffer: Buffer + current_selection: Selection + buffer_to_file: dict[Buffer, Path | None] + history: list[StateChange] + + def commit(self, change: StateChange) -> "EditorState": + match change: + case SetCurrentSelection(_, new_selection): + self.current_selection = new_selection + case SetCurrentBuffer(_, new_buffer): + self.current_buffer = new_buffer + case ModifyBuffer(buffer, old_start, old_end, _, new_content): + self.current_buffer.content = ( + self.current_buffer.content[:old_start] + + new_content + + self.current_buffer.content[old_end:] + ) + case CreateBuffer(buffer, file): + self.buffer_to_file[buffer] = file + case DeleteBuffer(buffer): + del self.buffer_to_file[buffer] + self.history.append(change) + return self + + def commit_changes(self, changes: list[StateChange]) -> "EditorState": + for change in changes: + self.commit(change) + return self + + +Result = tuple[Selection, list[StateChange]] |