+ if not self._refresh():
+ raise RuntimeError("Refresh too soon")
+
+ def _refresh(self) -> bool:
+ if self._refreshing and self._busy is not None:
+ if self._busy.value != self._busy_state:
+ self._ticks_disabled = True
+ self._refreshing = False
+ self._send_command_sequence(False, self._stop_sequence)
+ else:
+ return False
+ if self._core.current_group is None:
+ return True
+ # Refresh at seconds per frame rate
+ if self.time_to_refresh > 0:
+ return False
+
+ if not self._core.bus_free():
+ # Can't acquire display bus; skip updating this display. Try next display
+ return False
+
+ areas_to_refresh = self._get_refresh_areas()
+ if not areas_to_refresh:
+ return True
+
+ if self._acep:
+ self._start_refresh()
+ self._clean_area()
+ self._finish_refresh()
+ while self._refreshing:
+ # TODO: Add something here that can change self._refreshing
+ pass
+
+ self._start_refresh()
+ for area in areas_to_refresh:
+ self._refresh_area(area)
+ self._finish_refresh()
+
+ return True
+
+ def _release(self) -> None:
+ """Release the display and free its resources"""
+ if self._refreshing:
+ self._wait_for_busy()
+ self._ticks_disabled = True
+ self._refreshing = False
+ # Run stop sequence but don't wait for busy because busy is set when sleeping
+ self._send_command_sequence(False, self._stop_sequence)
+ self._core.release_display_core()
+ if self._busy is not None:
+ self._busy.deinit()
+
+ def _background(self) -> None:
+ """Run background refresh tasks."""
+
+ # Wait until initialized
+ if not hasattr(self, "_core"):
+ return
+
+ if self._ticks_disabled:
+ return
+
+ if self._refreshing:
+ refresh_done = False
+ if self._busy is not None:
+ busy = self._busy.value
+ refresh_done = busy == self._busy_state
+ else:
+ refresh_done = (
+ time.monotonic() * 1000 - self._core.last_refresh
+ > self._refresh_time
+ )
+
+ if refresh_done:
+ self._ticks_disabled = True
+ self._refreshing = False
+ # Run stop sequence but don't wait for busy because busy is set when sleeping
+ self._send_command_sequence(False, self._stop_sequence)
+
+ def _get_refresh_areas(self) -> list[Area]:
+ """Get a list of areas to be refreshed"""
+ areas = []
+ if self._core.full_refresh:
+ areas.append(self._core.area)
+ return areas
+ first_area = None
+ if self._core.current_group is not None:
+ self._core.current_group._get_refresh_areas( # pylint: disable=protected-access
+ areas
+ )
+ first_area = areas[0]
+ if first_area is not None and self._core.row_command == NO_COMMAND:
+ # Do a full refresh if the display doesn't support partial updates
+ areas = [self._core.area]
+ return areas
+
+ def _refresh_area(self, area: Area) -> bool:
+ """Redraw the area."""
+ # pylint: disable=too-many-locals, too-many-branches
+ clipped = Area()
+ # Clip the area to the display by overlapping the areas.
+ # If there is no overlap then we're done.
+ if not self._core.clip_area(area, clipped):
+ return True
+ print("---------------")
+ print(area.x1, area.y1, area.x2, area.y2)
+ print(clipped.x1, clipped.y1, clipped.x2, clipped.y2)
+ print("---------------")
+ subrectangles = 1
+ rows_per_buffer = clipped.height()
+ pixels_per_word = 32 // self._core.colorspace.depth
+ pixels_per_buffer = clipped.size()
+
+ # We should have lots of memory
+ buffer_size = clipped.size() // pixels_per_word
+
+ if clipped.size() > buffer_size * pixels_per_word:
+ rows_per_buffer = buffer_size * pixels_per_word // clipped.width()
+ if rows_per_buffer == 0:
+ rows_per_buffer = 1
+ subrectangles = clipped.height() // rows_per_buffer
+ if clipped.height() % rows_per_buffer != 0:
+ subrectangles += 1
+ pixels_per_buffer = rows_per_buffer * clipped.width()
+ buffer_size = pixels_per_buffer // pixels_per_word
+ if pixels_per_buffer % pixels_per_word:
+ buffer_size += 1
+
+ mask_length = (pixels_per_buffer // 32) + 1 # 1 bit per pixel + 1
+
+ passes = 1
+ if self._write_color_ram_command != NO_COMMAND:
+ passes = 2
+ for pass_index in range(passes):
+ remaining_rows = clipped.height()
+ if self._core.row_command != NO_COMMAND:
+ self._core.set_region_to_update(clipped)
+
+ write_command = self._write_black_ram_command
+ if pass_index == 1:
+ write_command = self._write_color_ram_command
+
+ self._core.begin_transaction()
+ self._core.send(DISPLAY_COMMAND, self._chip_select, bytes([write_command]))
+ self._core.end_transaction()
+
+ for subrect_index in range(subrectangles):
+ subrectangle = Area(
+ x1=clipped.x1,
+ y1=clipped.y1 + rows_per_buffer * subrect_index,
+ x2=clipped.x2,
+ y2=clipped.y1 + rows_per_buffer * (subrect_index + 1),
+ )
+ if remaining_rows < rows_per_buffer:
+ subrectangle.y2 = subrectangle.y1 + remaining_rows
+ remaining_rows -= rows_per_buffer
+
+ subrectangle_size_bytes = subrectangle.size() // (
+ 8 // self._core.colorspace.depth
+ )
+
+ buffer = memoryview(bytearray([0] * (buffer_size * 4))).cast("I")
+ mask = memoryview(bytearray([0] * (mask_length * 4))).cast("I")
+
+ if not self._acep:
+ self._core.colorspace.grayscale = True
+ self._core.colorspace.grayscale_bit = 7
+ if pass_index == 1:
+ if self._grayscale: # 4-color grayscale
+ self._core.colorspace.grayscale_bit = 6
+ self._core.fill_area(subrectangle, mask, buffer)
+ elif self._core.colorspace.tricolor:
+ self._core.colorspace.grayscale = False
+ self._core.fill_area(subrectangle, mask, buffer)
+ elif self._core.colorspace.sevencolor:
+ self._core.fill_area(subrectangle, mask, buffer)
+ else:
+ self._core.fill_area(subrectangle, mask, buffer)
+
+ # Invert it all
+ if (pass_index == 1 and self._color_bits_inverted) or (
+ pass_index == 0 and self._black_bits_inverted
+ ):
+ for i, _ in enumerate(buffer):
+ buffer[i] = ~buffer[i]
+
+ if not self._core.begin_transaction():
+ # Can't acquire display bus; skip the rest of the data. Try next display.
+ return False
+ self._core.send(
+ DISPLAY_DATA,
+ self._chip_select,
+ buffer.tobytes()[:subrectangle_size_bytes],
+ )
+ self._core.end_transaction()
+ return True
+
+ def _send_command_sequence(
+ self, should_wait_for_busy: bool, sequence: ReadableBuffer
+ ) -> None:
+ i = 0
+ while i < len(sequence):
+ command = sequence[i]
+ data_size = sequence[i + 1]
+ delay = (data_size & DELAY) != 0
+ data_size &= ~DELAY
+ data = sequence[i + 2 : i + 2 + data_size]
+ if self._two_byte_sequence_length:
+ data_size = ((data_size & ~DELAY) << 8) + sequence[i + 2]
+ data = sequence[i + 3 : i + 3 + data_size]
+
+ self._core.begin_transaction()
+ self._core.send(
+ DISPLAY_COMMAND, CHIP_SELECT_TOGGLE_EVERY_BYTE, bytes([command])
+ )
+ if data_size > 0:
+ self._core.send(
+ DISPLAY_DATA,
+ CHIP_SELECT_UNTOUCHED,
+ bytes(data),
+ )
+ self._core.end_transaction()
+ delay_time_ms = 0
+ if delay:
+ data_size += 1
+ delay_time_ms = sequence[i + 1 + data_size]
+ if delay_time_ms == 255:
+ delay_time_ms = 500
+ time.sleep(delay_time_ms / 1000)
+ if should_wait_for_busy:
+ self._wait_for_busy()
+ i += 2 + data_size
+ if self._two_byte_sequence_length:
+ i += 1
+
+ def _start_refresh(self) -> None:
+ # Run start sequence
+ self._core._bus_reset() # pylint: disable=protected-access
+ time.sleep(self._start_up_time)
+ self._send_command_sequence(True, self._start_sequence)
+ self._core.start_refresh()
+
+ def _finish_refresh(self) -> None:
+ # Actually refresh the display now that all pixel RAM has been updated
+ self._send_command_sequence(False, self._refresh_sequence)
+ self._ticks_disabled = False
+ self._refreshing = True
+ self._core.finish_refresh()
+
+ def _wait_for_busy(self) -> None:
+ if self._busy is not None:
+ while self._busy.value == self._busy_state:
+ time.sleep(0.001)
+
+ def _clean_area(self) -> bool:
+ width = self._core.width
+ height = self._core.height
+
+ buffer = bytearray([0x77] * (width // 2))
+ self._core.begin_transaction()
+ self._core.send(
+ DISPLAY_COMMAND, self._chip_select, bytes([self._write_black_ram_command])
+ )
+ self._core.end_transaction()
+ for _ in range(height):
+ if not self._core.begin_transaction():
+ return False
+ self._core.send(DISPLAY_DATA, self._chip_select, buffer)
+ self._core.end_transaction()
+ return True
+
+ @property
+ def rotation(self) -> int:
+ """The rotation of the display as an int in degrees"""
+ return self._core.rotation
+
+ @rotation.setter
+ def rotation(self, value: int) -> None:
+ if value % 90 != 0:
+ raise ValueError("Display rotation must be in 90 degree increments")
+ transposed = self._core.rotation in (90, 270)
+ will_transposed = value in (90, 270)
+ if transposed != will_transposed:
+ self._core.width, self._core.height = self._core.height, self._core.width
+ self._core.set_rotation(value)
+ if self._core.current_group is not None:
+ self._core.current_group._update_transform( # pylint: disable=protected-access
+ self._core.transform
+ )