-
-
Notifications
You must be signed in to change notification settings - Fork 29
Expand file tree
/
Copy pathscroll_helper.py
More file actions
1316 lines (1141 loc) · 59.8 KB
/
Copy pathscroll_helper.py
File metadata and controls
1316 lines (1141 loc) · 59.8 KB
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317
318
319
320
321
322
323
324
325
326
327
328
329
330
331
332
333
334
335
336
337
338
339
340
341
342
343
344
345
346
347
348
349
350
351
352
353
354
355
356
357
358
359
360
361
362
363
364
365
366
367
368
369
370
371
372
373
374
375
376
377
378
379
380
381
382
383
384
385
386
387
388
389
390
391
392
393
394
395
396
397
398
399
400
401
402
403
404
405
406
407
408
409
410
411
412
413
414
415
416
417
418
419
420
421
422
423
424
425
426
427
428
429
430
431
432
433
434
435
436
437
438
439
440
441
442
443
444
445
446
447
448
449
450
451
452
453
454
455
456
457
458
459
460
461
462
463
464
465
466
467
468
469
470
471
472
473
474
475
476
477
478
479
480
481
482
483
484
485
486
487
488
489
490
491
492
493
494
495
496
497
498
499
500
501
502
503
504
505
506
507
508
509
510
511
512
513
514
515
516
517
518
519
520
521
522
523
524
525
526
527
528
529
530
531
532
533
534
535
536
537
538
539
540
541
542
543
544
545
546
547
548
549
550
551
552
553
554
555
556
557
558
559
560
561
562
563
564
565
566
567
568
569
570
571
572
573
574
575
576
577
578
579
580
581
582
583
584
585
586
587
588
589
590
591
592
593
594
595
596
597
598
599
600
601
602
603
604
605
606
607
608
609
610
611
612
613
614
615
616
617
618
619
620
621
622
623
624
625
626
627
628
629
630
631
632
633
634
635
636
637
638
639
640
641
642
643
644
645
646
647
648
649
650
651
652
653
654
655
656
657
658
659
660
661
662
663
664
665
666
667
668
669
670
671
672
673
674
675
676
677
678
679
680
681
682
683
684
685
686
687
688
689
690
691
692
693
694
695
696
697
698
699
700
701
702
703
704
705
706
707
708
709
710
711
712
713
714
715
716
717
718
719
720
721
722
723
724
725
726
727
728
729
730
731
732
733
734
735
736
737
738
739
740
741
742
743
744
745
746
747
748
749
750
751
752
753
754
755
756
757
758
759
760
761
762
763
764
765
766
767
768
769
770
771
772
773
774
775
776
777
778
779
780
781
782
783
784
785
786
787
788
789
790
791
792
793
794
795
796
797
798
799
800
801
802
803
804
805
806
807
808
809
810
811
812
813
814
815
816
817
818
819
820
821
822
823
824
825
826
827
828
829
830
831
832
833
834
835
836
837
838
839
840
841
842
843
844
845
846
847
848
849
850
851
852
853
854
855
856
857
858
859
860
861
862
863
864
865
866
867
868
869
870
871
872
873
874
875
876
877
878
879
880
881
882
883
884
885
886
887
888
889
890
891
892
893
894
895
896
897
898
899
900
901
902
903
904
905
906
907
908
909
910
911
912
913
914
915
916
917
918
919
920
921
922
923
924
925
926
927
928
929
930
931
932
933
934
935
936
937
938
939
940
941
942
943
944
945
946
947
948
949
950
951
952
953
954
955
956
957
958
959
960
961
962
963
964
965
966
967
968
969
970
971
972
973
974
975
976
977
978
979
980
981
982
983
984
985
986
987
988
989
990
991
992
993
994
995
996
997
998
999
1000
"""
Scroll Helper
Handles scrolling text and image content for LED matrix displays.
Extracted from LEDMatrix core to provide reusable functionality for plugins.
Features:
- Pre-rendered scrolling image caching with numpy array optimization
- Fast numpy-based image slicing for high-performance scrolling (100+ FPS)
- Scroll position management with wrap-around
- Dynamic duration calculation based on content width
- Frame rate tracking and logging
- Scrolling state management integration with display_manager
- Support for both continuous and bounded scrolling modes
- Pre-allocated buffers to minimize memory allocations
"""
import logging
import math
import time
from typing import Optional, Dict, Any, List, Tuple
from PIL import Image
import numpy as np
# How often the frame-stats line is emitted, and therefore also the ceiling
# on a believable frame time: a scroll that renders at all cannot take this
# long over one frame, so a sample this large is an idle gap between scrolls.
FPS_LOG_INTERVAL = 5.0
# The stats line goes to INFO only when a window is worth an operator's
# attention, as Vegas's FPS line does (src/vegas_mode/coordinator.py): every
# 5s from every scroller was most of the journal on a healthy rig. A window is
# degraded when its frame rate falls below this fraction of the rate it was
# locked to (1 / its own median frame time; same 0.9 as Vegas) ...
STATS_HEALTHY_FRACTION = 0.9
# ... or when more than this share of its frames stalled (past 1.5x the
# median). A 1% stall rate barely moves the mean, so the fps test alone would
# miss the judder this line exists to show.
STATS_DEGRADED_STALL_RATE = 0.01
# A healthy scroller still logs at INFO this often, so silence in the journal
# means stopped rather than fine. Every window is still logged at DEBUG.
STATS_HEARTBEAT_INTERVAL = 300.0
def frame_stats_degraded(stats: Dict[str, Any]) -> bool:
"""Whether one frame_stats() window is worth logging at INFO."""
n = stats["frames"]
if n == 0 or stats["median"] <= 0:
return False
locked_fps = 1.0 / stats["median"]
return (stats["fps"] < locked_fps * STATS_HEALTHY_FRACTION
or stats["stalls"] > n * STATS_DEGRADED_STALL_RATE)
def _rgb_pixels(item) -> np.ndarray:
"""An appended item's pixels as an RGB array, as pasting it would draw them."""
if isinstance(item, np.ndarray):
return item
if item.mode != 'RGB':
item = item.convert('RGB')
return np.asarray(item)
def frame_stats(frame_times: list) -> Dict[str, Any]:
"""Summary statistics over one window of frame durations (seconds).
Split out of log_frame_rate() so the arithmetic can be tested without a
clock. Median and p95 are the real ones: the median takes both middle
samples on an even window, and p95 is nearest-rank, so a 100-frame window
reports the 95th sorted sample rather than the 96th. That matters twice
over, because the median is also the threshold the stall and skip counts
are measured against.
"""
window = sorted(frame_times)
n = len(window)
median = (window[n // 2] if n % 2
else (window[n // 2 - 1] + window[n // 2]) / 2.0)
mean = sum(window) / n
# Anything past 1.5x the median missed a panel refresh; anything under
# half of it never reached the panel at all (dirty tracking skipped the
# swap, so the frame did not wait for vsync).
return {
"frames": n,
"fps": (1.0 / mean) if mean > 0 else 0.0,
"median": median,
"p95": window[max(0, math.ceil(0.95 * n) - 1)],
"max": window[-1],
"min": window[0],
"stalls": sum(1 for f in window if f > median * 1.5),
"skips": sum(1 for f in window if f < median * 0.5),
}
def format_frame_stats(frame_times: list) -> str:
"""The one-line rendering of frame_stats(), in milliseconds."""
s = frame_stats(frame_times)
n = s["frames"]
return (
f"{s['fps']:.1f} fps over {n} frames | "
f"median {s['median'] * 1000:.2f}ms p95 {s['p95'] * 1000:.2f}ms "
f"max {s['max'] * 1000:.2f}ms min {s['min'] * 1000:.2f}ms | "
f"stalls {s['stalls']} ({100.0 * s['stalls'] / n:.1f}%) "
f"skips {s['skips']} ({100.0 * s['skips'] / n:.1f}%)"
)
class ScrollHelper:
"""
Helper class for scrolling text and image content on LED displays.
Provides functionality for:
- Creating and caching scrolling images (with numpy array optimization)
- Fast numpy-based image slicing for high-performance scrolling
- Managing scroll position with wrap-around
- Calculating dynamic display duration
- Frame rate tracking and performance monitoring
- Integration with display manager scrolling state
- Pre-allocated buffers for minimal memory allocations
Performance optimizations:
- Uses numpy arrays for fast array slicing instead of PIL crop operations
- Pre-computes numpy array from PIL image to avoid repeated conversions
- Reuses pre-allocated frame buffer to minimize allocations
- Optimized for 100+ FPS scrolling performance
"""
def __init__(self, display_width: int, display_height: int,
logger: Optional[logging.Logger] = None):
"""
Initialize the ScrollHelper.
Args:
display_width: Width of the LED matrix display
display_height: Height of the LED matrix display
logger: Optional logger instance
"""
self.display_width = display_width
self.display_height = display_height
self.logger = logger or logging.getLogger(__name__)
# Scrolling state
self.scroll_position = 0.0
self.total_distance_scrolled = 0.0 # Track total distance including wrap-arounds
self.scroll_speed = 1.0
self.scroll_delay = 0.001 # Minimal delay for high FPS (1ms)
self.cached_image = None # see the property below
self.cached_array: Optional[np.ndarray] = None # Numpy array cache for fast operations
self.total_scroll_width = 0
# An extended strip lives in a buffer with spare room after it, and
# cached_array is a view of the buffer's live columns: an append writes
# only the new columns, and a trim only moves the view's start. See
# append_content. _strip_view is the view this helper last made; a
# cached_array that is anything else was set from outside and is not
# written through.
self._strip_buffer: Optional[np.ndarray] = None
self._strip_view: Optional[np.ndarray] = None
self._strip_start = 0
#: Bytes the last append_content / drop_scrolled_prefix copied, for
#: frame-timing attribution (src/common/frame_timing.py note_op).
self.last_copy_bytes = 0
# Pre-allocated buffer for output frame (reused to avoid allocations)
self._frame_buffer: Optional[np.ndarray] = None
# Sub-pixel scrolling: OFF by default, and that is deliberate.
# Blending renders a half-step by mixing two adjacent columns 50/50.
# On a high-resolution screen that reads as smooth motion; on a coarse
# LED matrix showing pixel-font text it does not. A one-pixel stroke
# becomes two half-brightness pixels, so frames alternate between crisp
# and smeared and the text appears to shimmer and jump a pixel ahead --
# tested on a 2x128x64 panel and clearly worse than integer stepping.
#
# The rule this display obeys: motion is smooth when it advances a
# whole number of pixels per refresh. Anything slower must either
# blend (blur) or repeat frames (judder); blending is the worse of the
# two here. Vegas mode still opts in via set_sub_pixel_scrolling().
self.sub_pixel_scrolling = False
# Frame-based scrolling settings
self.frame_based_scrolling = False
#: Whole pixels to advance per presented frame, or None to pace
#: off elapsed time. See set_pixels_per_frame.
self.fixed_pixels_per_frame = None
self.last_step_time = 0.0 # Time of the last position update
# Time tracking for scroll updates
self.last_update_time: Optional[float] = None
#: Informational only: the presentation rate scroll_config chose
#: (panel refresh / frame hold). Nothing paces off it -- the helper
#: steps per call and SwapOnVSync paces the calls. Kept because
#: plugins and their tests read it back.
self.target_fps = 120
# Dynamic duration settings
self.dynamic_duration_enabled = True
self.min_duration = 30
self.max_duration = 300
self.duration_buffer = 0.1
self.calculated_duration = 60
self.scroll_start_time: Optional[float] = None
self.last_progress_log_time: Optional[float] = None
self.progress_log_interval = 5.0 # seconds
# Frame rate tracking. last_frame_time is None until the first frame
# of a scroll is rendered -- see log_frame_rate() for why timing from
# construction (or from the end of the previous scroll) is wrong.
self.frame_count = 0
self.last_frame_time: Optional[float] = None
self.last_fps_log_time = time.time()
self.frame_times = []
# Every frame time since the last stats line, so the 5s summary can
# report the tail rather than one arbitrary sample. Cleared on log.
self._window: list = []
# INFO-level stats bookkeeping (see STATS_HEARTBEAT_INTERVAL). Kept
# across reset_scroll(): a heartbeat per scroll start would bring the
# chatter back. 0.0 so the first window after start-up is at INFO.
self._stats_last_info_log = 0.0
self._stats_was_degraded = False
# Scrolling state management
self.is_scrolling = False
self.scroll_complete = False
# -- the strip as a PIL image ---------------------------------------------
#
# Every frame is cut from cached_array; nothing on the frame path reads the
# PIL image's pixels. Extending and trimming a strip (append_content,
# drop_scrolled_prefix) used to rebuild that image in full each time
# anyway: Image.fromarray of a Vegas-sized strip is 1.7-3.8ms on a Pi 4,
# twice per extension, on the render thread. Those two now leave it to be
# built from the array on first read, which in Vegas means only by a
# multi-display sync push -- and the strip is not held twice in memory.
#
# Assigning cached_image still stores exactly what was assigned; a lazy
# image is only ever one the helper derived from its own array.
@property
def cached_image(self) -> Optional[Image.Image]:
"""The strip as a PIL image, built from ``cached_array`` if deferred."""
image = self.__dict__.get('_cached_image')
if image is not None:
return image
source = self.__dict__.get('_image_source')
if source is None:
return None
# Built from the array this read started with. Another thread (the
# sync push) may read while the render thread extends the strip; it
# then gets the strip as it was, as it did when the image was built
# eagerly, and the stale build is not kept.
image = Image.fromarray(source)
if self.__dict__.get('_image_source') is source:
self._cached_image = image
return image
@cached_image.setter
def cached_image(self, image: Optional[Image.Image]) -> None:
self._cached_image = image
self._image_source = None
def _defer_image(self) -> None:
"""The array just changed under the image: rebuild it only if read."""
self._cached_image = None
self._image_source = self.cached_array
def has_strip(self) -> bool:
"""Whether there is a strip (an image, or one deferred), not reading it."""
return (self.__dict__.get('_cached_image') is not None
or self.__dict__.get('_image_source') is not None)
def create_scrolling_image(self, content_items: list,
item_gap: int = 32,
element_gap: int = 16,
lead_gap: Optional[int] = None) -> Image.Image:
"""
Create a wide image containing all content items for scrolling.
Args:
content_items: List of PIL Images to include in scroll
item_gap: Gap between different items
element_gap: Gap between elements within an item
lead_gap: Blank columns before the first item. Defaults to a full
display width, which makes a standalone ticker scroll in from
off-screen. Callers that loop many plugins back-to-back (Vegas
mode) pass a smaller value, since a full display width of black
reads as the panel being switched off at the start of every
cycle.
Returns:
PIL Image containing all content arranged horizontally
"""
if lead_gap is None:
lead_gap = self.display_width
lead_gap = max(0, int(lead_gap))
if not content_items:
# Create empty image if no content
# Still set total_scroll_width to 0 to indicate no scrollable content
self.total_scroll_width = 0
self.cached_image = Image.new('RGB', (self.display_width, self.display_height), (0, 0, 0))
self.cached_array = np.array(self.cached_image)
self._forget_strip_buffer()
self.scroll_position = 0.0
self.total_distance_scrolled = 0.0
self.scroll_complete = False
return self.cached_image
# Calculate total width needed
# Sum of all item widths
total_width = sum(img.width for img in content_items)
# Add item gaps between items (not after last item)
total_width += item_gap * (len(content_items) - 1)
# Add element_gap after each item (matches positioning logic)
total_width += element_gap * len(content_items)
# Add initial gap before first item
total_width += lead_gap
# Create the full scrolling image
full_image = Image.new('RGB', (total_width, self.display_height), (0, 0, 0))
# Position items
current_x = lead_gap # Start with initial gap
for i, img in enumerate(content_items):
# Paste the item image
full_image.paste(img, (current_x, 0))
current_x += img.width + element_gap
# Add gap between items (except after last item)
if i < len(content_items) - 1:
current_x += item_gap
# Store the image and update scroll width
self.cached_image = full_image
# Convert to numpy array for fast operations
self.cached_array = np.array(full_image)
self._forget_strip_buffer()
actual_image_width = full_image.width
self.total_scroll_width = actual_image_width
self.scroll_position = 0.0
self.total_distance_scrolled = 0.0
self.scroll_complete = False
# Pre-allocate frame buffer if needed
if self._frame_buffer is None or self._frame_buffer.shape != (self.display_height, self.display_width, 3):
self._frame_buffer = np.zeros((self.display_height, self.display_width, 3), dtype=np.uint8)
# Calculate dynamic duration
self._calculate_dynamic_duration()
now = time.time()
self.scroll_start_time = now
self.last_progress_log_time = now
# The position just went back to 0; the first update must not advance
# it by however long the helper sat idle (off-screen) before this.
self.last_update_time = now
self.logger.info(
"Dynamic duration target set to %ds (min=%ds, max=%ds, buffer=%.2f)",
self.calculated_duration,
self.min_duration,
self.max_duration,
self.duration_buffer,
)
self.logger.info(
"Created scrolling image: %dx%dpx (total_scroll_width=%dpx, %d items, item_gap=%d, element_gap=%d)",
actual_image_width, self.display_height, self.total_scroll_width,
len(content_items), item_gap, element_gap
)
return full_image
def update_scroll_position(self) -> None:
"""
Advance the scroll by one presented frame and handle wrap-around.
With a fixed per-frame step (set_pixels_per_frame, which
scroll_config.configure sets for a crisp speed) every call moves
exactly that many pixels and no clock is read; the caller's
vsync-blocking swap, held for ``frame_hold`` refreshes, sets the rate.
Otherwise the position advances by elapsed time at the configured
speed.
"""
if not self.has_strip():
return
# Calculate frame time for consistent scroll speed regardless of FPS
current_time = time.time()
if self.last_update_time is None:
self.last_update_time = current_time
delta_time = current_time - self.last_update_time
self.last_update_time = current_time
if self.scroll_start_time is None:
self.scroll_start_time = current_time
self.last_progress_log_time = current_time
# Update scroll position
if self.fixed_pixels_per_frame:
# One presented frame, one fixed whole-pixel step. No clock is
# consulted, so no jitter reaches the position and every frame
# moves the eye by the same amount. See set_pixels_per_frame.
pixels_to_move = self.fixed_pixels_per_frame
self.last_step_time = current_time
elif self.frame_based_scrolling:
# Frame-based: move fixed amount when scroll_delay has passed
# This matches stock ticker behavior: move pixels, then wait scroll_delay
# Initialize last_step_time on first call to prevent huge initial jump
if self.last_step_time == 0.0:
self.last_step_time = current_time
# Frame-based mode advances by elapsed time, exactly like the
# time-based branch below, at the same configured speed
# (scroll_speed px per scroll_delay seconds).
#
# It used to step discretely: 0, 1 or 2 whole pixels depending on
# whether a wall clock had passed scroll_delay. Plugins set
# scroll_delay to the target frame period, so that comparison sits
# exactly on its own threshold and the decision flips on sub-
# millisecond jitter -- a frame a hair early moved nothing and
# rendered an identical frame, a frame a hair late moved two
# pixels. Rounding the step count fixed the stalls but still
# discarded the remainder, so the error never corrected.
#
# Accumulating elapsed time keeps position exactly proportional to
# real time: jitter shifts a pixel boundary by a fraction of a
# frame instead of flipping a whole step, and nothing is lost or
# gained. This is what the one visibly smooth scroller on the
# hardware (the stock ticker) was already doing by virtue of never
# enabling frame-based mode.
# set_scroll_delay clamps scroll_delay to at least 0.001.
pixels_per_second = self.scroll_speed / self.scroll_delay
pixels_to_move = pixels_per_second * delta_time
self.last_step_time = current_time
else:
# Time-based: move based on time delta (correct speed over time)
# scroll_speed is pixels per second
pixels_to_move = self.scroll_speed * delta_time
self.scroll_position += pixels_to_move
self.total_distance_scrolled += pixels_to_move
# One pass is total_scroll_width. With the default lead_gap the strip
# starts with display_width of blank, so by then the last item has
# fully left the panel; a caller passing a smaller lead_gap (Vegas)
# decides for itself where its cycle ends. Adding display_width here
# caused 1-2 extra wrap-arounds on wide chains.
required_total_distance = self.total_scroll_width
# Guard: zero-width content has nothing to scroll — keep position at 0 and skip
# completion/wrap logic to avoid producing an invalid -1 position.
if required_total_distance == 0:
self.scroll_position = 0
return
# Check completion FIRST (before wrap-around) to prevent visual loop
# When dynamic duration is enabled and cycle is complete, stop at end instead of wrapping
is_complete = self.total_distance_scrolled >= required_total_distance
if is_complete:
# Only log completion once to avoid spam
if not self.scroll_complete:
elapsed = current_time - (self.scroll_start_time or current_time)
scroll_percent = (self.total_distance_scrolled / required_total_distance * 100) if required_total_distance > 0 else 0.0
position_percent = (self.scroll_position / self.total_scroll_width * 100) if self.total_scroll_width > 0 else 0.0
self.logger.info(
"Scroll cycle COMPLETE: scrolled %.0f/%d px (%.1f%%, position=%.0f/%.0f px, %.1f%%) - elapsed %.2fs, target %.2fs",
self.total_distance_scrolled,
required_total_distance,
scroll_percent,
self.scroll_position,
self.total_scroll_width,
position_percent,
elapsed,
self.calculated_duration,
)
self.scroll_complete = True
# Clamp position to prevent wrap when complete
if self.scroll_position >= self.total_scroll_width:
self.scroll_position = self.total_scroll_width - 1
self.logger.debug("Clamped scroll position to %d (max=%d)", self.scroll_position, self.total_scroll_width - 1)
else:
self.scroll_complete = False
# Only wrap-around if cycle is not complete yet
if self.scroll_position >= self.total_scroll_width:
elapsed = current_time - self.scroll_start_time
self.scroll_position = self.scroll_position - self.total_scroll_width
self.logger.info(
"Scroll wrap-around detected: position reset, total_distance=%.0f/%d px (elapsed %.2fs, target %.2fs)",
self.total_distance_scrolled,
required_total_distance,
elapsed,
self.calculated_duration,
)
if (
self.dynamic_duration_enabled
and self.last_progress_log_time is not None
and current_time - self.last_progress_log_time >= self.progress_log_interval
):
elapsed_time = current_time - (self.scroll_start_time or current_time)
required_total_distance = self.total_scroll_width
# Progress telemetry, emitted every few seconds for the whole of
# every scroll. It says how far along a marquee is, which is what
# you turn debug on to watch and not something an operator needs
# in the journal on a device that scrolls all day.
self.logger.debug(
"Scroll progress: elapsed=%.2fs, target=%.2fs, total_scrolled=%.0f/%d px (%.1f%%)",
elapsed_time,
self.calculated_duration,
self.total_distance_scrolled,
required_total_distance,
(self.total_distance_scrolled / required_total_distance * 100) if required_total_distance > 0 else 0.0,
)
self.last_progress_log_time = current_time
def get_visible_portion(self) -> Optional[Image.Image]:
"""
Get the currently visible portion of the scrolling image using fast numpy operations.
Uses integer pixel positioning for high-performance scrolling.
Returns:
PIL Image showing the visible portion, or None if no cached image
"""
if self.cached_array is None or not self.has_strip():
return None
start_x_int = int(self.scroll_position)
end_x_int = start_x_int + self.display_width
# Integer positioning quantises motion to whole pixels, so the number of
# distinct frames per second equals the scroll speed in px/s, no matter
# how fast the loop renders. At 50px/s and 78fps that made 36% of frames
# identical: the extra frames cost work and bought nothing. Blending
# between the two neighbouring positions gives motion at the frame rate
# instead of the step rate.
if self.sub_pixel_scrolling:
fractional = self.scroll_position - start_x_int
if fractional > 0.0:
return self._blend_visible_portion(start_x_int, fractional)
return self._get_visible_portion_integer(start_x_int, end_x_int)
def _blend_visible_portion(self, start_x: int, fractional: float) -> Image.Image:
"""
Linear blend between the frames at ``start_x`` and ``start_x + 1``.
Implemented with numpy rather than scipy.ndimage.shift, which is not
installed on the target devices.
Args:
start_x: Left column of the earlier of the two frames
fractional: How far between the two, in [0, 1)
Returns:
The blended frame
"""
width = self.display_width
strip_width = self.cached_array.shape[1]
if 0 <= start_x and start_x + width + 1 <= strip_width:
# Slice the backing array directly. Going via
# _get_visible_portion_integer would build two PIL images only for
# them to be converted straight back to arrays, which measured 15x
# the cost of the integer path.
near = self.cached_array[:, start_x:start_x + width]
far = self.cached_array[:, start_x + 1:start_x + 1 + width]
else:
# One of the slices wraps (close to the end, or a strip narrower
# than the panel); let the integer path handle that and pay the
# conversion. Continuous mode extends the strip before reaching
# here, so this is the rare case.
near = np.asarray(
self._get_visible_portion_integer(start_x, start_x + width))
far = np.asarray(
self._get_visible_portion_integer(start_x + 1, start_x + 1 + width))
# Fixed-point rather than float32: integer multiply-add on uint16 is
# markedly faster than float maths on the Pi's ARM cores, and 8 bits of
# weight is finer than the panel can show.
weight = int(fractional * 256.0)
blended = (
(near.astype(np.uint16) * (256 - weight)
+ far.astype(np.uint16) * weight) >> 8
).astype(np.uint8)
return Image.frombytes(
'RGB', (width, self.display_height),
np.ascontiguousarray(blended).tobytes()
)
def _get_visible_portion_integer(self, start_x: int, end_x: int) -> Image.Image:
"""Fast integer pixel extraction (no interpolation).
Uses Image.frombytes instead of Image.fromarray: frombytes skips
numpy's array-protocol overhead and is ~50% faster for the display-sized
slices (128×32 = 12 KB) used here.
"""
_size = (self.display_width, self.display_height)
img_w = self.cached_array.shape[1]
if 0 <= start_x and end_x <= img_w:
# Normal case: single contiguous slice (fastest path). tobytes()
# on the column-slice view already returns C-order bytes, so
# ascontiguousarray() first only added a second full-frame copy.
return Image.frombytes(
'RGB', _size,
self.cached_array[:, start_x:end_x].tobytes())
# Ensure frame buffer is allocated for all non-simple paths
if self._frame_buffer is None or self._frame_buffer.shape != (self.display_height, self.display_width, 3):
self._frame_buffer = np.zeros((self.display_height, self.display_width, 3), dtype=np.uint8)
if img_w == 0:
self._frame_buffer[:] = 0
else:
# The frame runs off the strip, so it carries on from the head:
# frame column j is strip column (start_x + j) modulo the strip's
# width -- the tail and then the head, and a strip narrower than
# the panel repeated across it. Copying the tail and then the rest
# of the frame from the head assumed the head was that wide, and
# raised at every position for a strip narrower than the panel
# (Vegas composes one, with no lead-in, when its content is
# narrower than the chain).
np.take(self.cached_array, np.arange(start_x, end_x), axis=1,
mode='wrap', out=self._frame_buffer)
return Image.frombytes('RGB', _size, self._frame_buffer.tobytes())
def calculate_dynamic_duration(self) -> int:
"""
Calculate display duration based on content width and scroll settings.
Returns:
Duration in seconds
"""
if not self.dynamic_duration_enabled:
return self.min_duration
# Validate total_scroll_width is set and valid
if not self.total_scroll_width or self.total_scroll_width <= 0:
if self.total_scroll_width == 0:
self.logger.warning(
"Dynamic duration calculation skipped: total_scroll_width is 0. "
"Ensure create_scrolling_image() or set_scrolling_image() has been called. "
"Using minimum duration: %ds",
self.min_duration
)
else:
self.logger.warning(
"Dynamic duration calculation skipped: total_scroll_width is invalid (%s). "
"Using minimum duration: %ds",
self.total_scroll_width,
self.min_duration
)
return self.min_duration
try:
# The strip's width plus one more screen, so the duration covers
# the last item leaving the panel even when the strip has less
# than display_width of lead-in blank (lead_gap).
total_scroll_distance = self.total_scroll_width + self.display_width
# Calculate effective pixels per second based on scrolling mode
if self.frame_based_scrolling:
# Frame-based mode: scroll_speed is pixels per scroll_delay
# seconds, and set_scroll_delay keeps scroll_delay >= 0.001.
pixels_per_second = self.scroll_speed / self.scroll_delay
scroll_mode_str = "frame-based"
else:
# Time-based mode: scroll_speed is already pixels per second
pixels_per_second = self.scroll_speed
scroll_mode_str = "time-based"
# Calculate time based on effective pixels per second
total_time = total_scroll_distance / pixels_per_second
# Add buffer time for smooth cycling
buffer_time = total_time * self.duration_buffer
calculated_duration = int(total_time + buffer_time)
# Apply min/max limits
if calculated_duration < self.min_duration:
self.calculated_duration = self.min_duration
elif calculated_duration > self.max_duration:
self.calculated_duration = self.max_duration
else:
self.calculated_duration = calculated_duration
self.logger.debug("Dynamic duration calculation (%s mode):", scroll_mode_str)
self.logger.debug(" Display width: %dpx", self.display_width)
self.logger.debug(" Content width: %dpx", self.total_scroll_width)
self.logger.debug(" Total scroll distance: %dpx", total_scroll_distance)
if self.frame_based_scrolling:
self.logger.debug(" Scroll speed: %.2f px/frame, delay: %.3fs", self.scroll_speed, self.scroll_delay)
self.logger.debug(" Effective speed: %.1f px/second", pixels_per_second)
else:
self.logger.debug(" Scroll speed: %.1f px/second", pixels_per_second)
self.logger.debug(" Base time: %.2fs", total_time)
self.logger.debug(" Buffer time: %.2fs", buffer_time)
self.logger.debug(" Final duration: %ds", self.calculated_duration)
return self.calculated_duration
except (ValueError, ZeroDivisionError, TypeError) as e:
self.logger.error("Error calculating dynamic duration: %s", e)
return self.min_duration
def is_scroll_complete(self) -> bool:
"""
Check if the current scroll cycle is complete.
Returns:
True if scroll has wrapped around to the beginning
"""
return self.scroll_complete
def append_content(self, content_items: list,
item_gap: int = 32,
element_gap: int = 0) -> bool:
"""
Append items to the right of the existing strip, preserving scroll state.
Lets a caller keep one continuous strip instead of replacing it. Vegas
mode uses this so the next group of plugins scrolls in from the right
rather than the strip being swapped out underneath the viewer — a swap
shows as a flash and a hard cut to already-full-screen content.
``scroll_position`` and ``total_distance_scrolled`` are untouched, so
motion continues uninterrupted; only the strip gets longer. Because
completion is measured against ``total_scroll_width``, extending the
strip also defers completion, which is the intent.
Args:
content_items: Images to append, in order. An item may instead be
its pixels already as an RGB array (``np.asarray`` of an RGB
image), so a caller can do that conversion off the render
thread (Vegas prepares its blocks with the group).
item_gap: Gap between appended items, and between the existing
content and the first appended item
element_gap: Extra gap after each item, mirroring
create_scrolling_image
Returns:
True if content was appended
"""
if not content_items:
return False
if self.cached_array is None or not self.has_strip():
# Nothing to extend yet — this is just the first build.
self.create_scrolling_image(
[Image.fromarray(item) if isinstance(item, np.ndarray) else item
for item in content_items],
item_gap=item_gap, element_gap=element_gap, lead_gap=0)
return True
gap = max(0, item_gap)
pieces = []
x = 0
for item in content_items:
x += gap # separate from whatever precedes
pixels = _rgb_pixels(item)
pieces.append((x, pixels))
x += pixels.shape[1] + element_gap
addition_width = x
# Each item is written straight into the spare room after the strip,
# when there is some: the strip can be tens of thousands of columns
# wide and this runs on the render thread, where copying all of it
# (2-3 ms at 512x64 on a Pi 4) -- or even laying the items out in an
# image of their own first (another 4-5 ms) -- cost the frame after
# every extension. The PIL image is built from the array only if
# something reads it (see cached_image).
self.cached_array = self._extended_strip(pieces, addition_width)
self._defer_image()
self.total_scroll_width = self.cached_array.shape[1]
self.scroll_complete = False
# Debug: this runs on the render thread, and the caller (Vegas) logs
# each extension itself.
self.logger.debug(
"Appended %d item(s) (%dpx) to scroll strip: now %dpx, position %.0f",
len(content_items), addition_width, self.total_scroll_width,
self.scroll_position
)
return True
#: Room an extended strip's buffer is given, as a multiple of what it
#: holds when (re)allocated. Trims free columns at the front and appends
#: use them at the back, so with 3x the buffer is reallocated -- the one
#: full copy -- about once every two strip-lengths scrolled.
STRIP_SPARE_FACTOR = 3.0
def _extended_strip(self, pieces: List[Tuple[int, np.ndarray]], added: int) -> np.ndarray:
"""The strip with ``added`` black columns after it, ``pieces`` drawn in.
Each piece is ``(x, pixels)``, x counted from the old strip's end;
written in place when the buffer has the room.
"""
live = self.cached_array
if live is None:
# append_content builds a first strip itself and never comes here.
raise RuntimeError("no strip to extend")
width = live.shape[1]
buffer = self._strip_buffer
if (live is self._strip_view and buffer is not None
and self._strip_start + width + added <= buffer.shape[1]):
end = self._strip_start + width
self.last_copy_bytes = 0
else:
total = width + added
buffer = np.empty((live.shape[0], max(total + 1, int(total * self.STRIP_SPARE_FACTOR)))
+ live.shape[2:], dtype=live.dtype)
buffer[:, :width] = live
self._strip_buffer = buffer
self._strip_start = 0
end = width
self.last_copy_bytes = live.nbytes
rows = buffer.shape[0]
region = buffer[:, end:end + added]
# Black only where no piece lands -- the gaps, and below a short
# piece: blanking the whole region first cost as much again as
# writing the pieces (1.8 ms at 512x64 on a Pi 4).
covered = 0
for x, pixels in pieces:
pixels = pixels[:rows]
cols = pixels.shape[1]
if x > covered:
region[:, covered:x] = 0
region[:pixels.shape[0], x:x + cols] = pixels
if pixels.shape[0] < rows:
region[pixels.shape[0]:, x:x + cols] = 0
covered = max(covered, x + cols)
if covered < added:
region[:, covered:] = 0
self.last_copy_bytes += region.nbytes
self._strip_view = buffer[:, self._strip_start:self._strip_start + width + added]
return self._strip_view
def _forget_strip_buffer(self) -> None:
"""A new strip replaces the extended one: let its buffer go."""
self._strip_buffer = None
self._strip_view = None
self._strip_start = 0
def drop_scrolled_prefix(self, keep_before: int = 0) -> int:
"""
Discard columns that have already scrolled past, to bound memory.
A continuously extended strip would otherwise grow without limit. All
the positional state is shifted by the amount removed so the visible
frame and the completion arithmetic are unchanged:
``total_distance_scrolled`` and ``total_scroll_width`` both shrink by the
same amount, preserving their difference.
Args:
keep_before: Columns to retain behind the current position, as a
safety margin against a caller reading slightly behind it
Returns:
Number of columns actually removed
"""
if self.cached_array is None or not self.has_strip():
return 0
strip_width = self.cached_array.shape[1]
# While the viewport wraps, get_visible_portion fills its right-hand side
# from the *head* of the strip, so trimming the head would change what
# is on screen. Continuous mode extends before ever reaching that state;
# refusing here keeps "trimming is invisible" true unconditionally.
if self.scroll_position + self.display_width > strip_width:
return 0
cut = int(self.scroll_position) - max(0, keep_before)
if cut <= 0:
return 0
# Never trim so far that the remaining strip is narrower than the
# viewport, or get_visible_portion has nothing to slice.
cut = min(cut, max(0, strip_width - self.display_width))
if cut <= 0:
return 0
if self.cached_array is self._strip_view:
# Only the view's start moves; the columns behind it are reused
# when the buffer is next reallocated (append_content).
self._strip_view = self.cached_array[:, cut:]
self._strip_start += cut
self.cached_array = self._strip_view
self.last_copy_bytes = 0
else:
# Not a strip this helper extended: .copy() so the original
# buffer is released rather than kept alive by a view.
self.cached_array = self.cached_array[:, cut:].copy()
self.last_copy_bytes = self.cached_array.nbytes
self._defer_image()
self.total_scroll_width = self.cached_array.shape[1]
self.scroll_position -= cut
self.total_distance_scrolled = max(0.0, self.total_distance_scrolled - cut)
self.logger.debug(
"Dropped %dpx of scrolled strip: now %dpx, position %.0f",
cut, self.total_scroll_width, self.scroll_position
)
return cut
def patch_columns(self, x: int, pixels: np.ndarray) -> int:
"""Overwrite the strip's columns from ``x`` with ``pixels``, in place.
What a live Vegas element update is (src/vegas_mode/elements.py): the
strip keeps its width, the scroll keeps its position, and only these
columns change. Call it between frames on the thread that draws them;
every frame copies its slice out of the strip (get_visible_portion),
so no frame already handed on can see half a patch.
Clipped to the strip at both ends. Refused (0) for an array this
helper may not write -- the multi-display follower adopts a read-only
one -- or for pixels of another height. The PIL image is deferred, so
a later read of cached_image shows the patch.
Args:
x: Strip column of the first column of ``pixels``
pixels: uint8 array (height, width, 3)
Returns:
Bytes written.
"""
strip = self.cached_array
if strip is None or not strip.flags.writeable:
return 0
if pixels.ndim != 3 or pixels.shape[0] != strip.shape[0] \
or pixels.shape[2] != strip.shape[2]:
return 0
width = pixels.shape[1]
lo, hi = max(0, int(x)), min(strip.shape[1], int(x) + width)
if hi <= lo:
return 0
strip[:, lo:hi] = pixels[:, lo - int(x):hi - int(x)]
if self.__dict__.get('_cached_image') is not None \
or self.__dict__.get('_image_source') is not None:
self._defer_image()
return (hi - lo) * strip.shape[0] * strip.shape[2]
def remaining_unscrolled(self) -> int:
"""Columns of strip still to the right of the viewport."""
if not self.has_strip():
return 0
return max(0, self.total_scroll_width - int(self.scroll_position)
- self.display_width)
def reset_scroll(self) -> None:
"""
Reset scroll position to beginning.
"""
self.scroll_position = 0.0
self.total_distance_scrolled = 0.0
self.scroll_complete = False
now = time.time()
self.scroll_start_time = now
self.last_progress_log_time = now
self.last_step_time = now # Reset step timer
# Reset last_update_time to prevent large delta_time on next update
# This ensures smooth scrolling after reset without jumping ahead
self.last_update_time = now
# Same reasoning for the frame-rate clock: the first frame after a
# reset has no predecessor in this scroll, and timing it against the
# last frame of the previous one measures the idle gap between them.
self.last_frame_time = None
self.logger.debug("Scroll position reset")
def reset(self) -> None:
"""Alias for reset_scroll() for convenience."""
self.reset_scroll()
def set_scrolling_image(self, image: Image.Image) -> None:
"""
Set a pre-rendered scrolling image and initialize all required state.
This method should be used when plugins create their own scrolling image
instead of using create_scrolling_image(). It properly initializes both
cached_image and cached_array, and updates all related state.
Args:
image: PIL Image containing the scrolling content
"""
if image is None:
self.logger.warning("Attempted to set None as scrolling image, clearing cache instead")
self.clear_cache()
return
# Every frame is cut from cached_array with Image.frombytes('RGB', ...),
# which reads a 4-channel (RGBA) array as garbage and raises on a
# 1-channel (L) one. Transparent pixels go to black, the panel's
# background, rather than to whatever colour hides under the alpha.
if image.mode != 'RGB':
if 'A' in image.mode or 'transparency' in image.info:
rgba = image.convert('RGBA')