866 lines
31 KiB
Python
866 lines
31 KiB
Python
# This file is part of ssiv.
|
|
#
|
|
# ssiv is free software: you can redistribute it and/or modify
|
|
# it under the terms of the GNU Affero General Public License as published by
|
|
# the Free Software Foundation, either version 3 of the License, or
|
|
# (at your option) any later version.
|
|
#
|
|
# ssiv is distributed in the hope that it will be useful,
|
|
# but WITHOUT ANY WARRANTY; without even the implied warranty of
|
|
# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
|
|
# GNU Affero General Public License for more details.
|
|
#
|
|
# You should have received a copy of the GNU Affero General Public License
|
|
# along with ssiv. If not, see <https://www.gnu.org/licenses/>.
|
|
|
|
from _thread import allocate_lock,start_new_thread
|
|
from collections import deque
|
|
from ctypes import byref,sizeof
|
|
from os import urandom
|
|
from os.path import join
|
|
|
|
from .container import *
|
|
from .environ_draw import *
|
|
from .hci import *
|
|
from .modules import *
|
|
from ..co import *
|
|
from ..mt import start_thread
|
|
|
|
__all__=['ENV']
|
|
|
|
class ENV(NonWidget):
|
|
|
|
__slots__=(
|
|
'_stop','_tasklock','_containerlock','_hci','_listener','_window',
|
|
'_inrequire','_inrequirelock','_inrequirerun','_tasks',
|
|
'_canvases','_thumbnails','_containers','_actions',
|
|
'_anti_alias','_scale_ratio','_use_page','_canvas_offset','_canvas_offset_old',
|
|
'_thumbsize','_maxthumbsize','_thumbgap','_checker','_fgcolor','_cachesize',
|
|
'_viewmode','_tiling_width','_double_canvas','_double_right',
|
|
'_statusbar_size','_statusbar_pos','_orientation','_blklmt',
|
|
'_scene','_scene_id','_scene_pos','_animes','_motion_id','_motion_status',
|
|
'_motion_indicator',
|
|
)
|
|
|
|
def __init__(self,config):
|
|
self._stop=False
|
|
self._tasklock=allocate_lock()
|
|
self._containerlock=allocate_lock()
|
|
self._hci=HCI(self,config)
|
|
self._listener=None
|
|
self._window=None
|
|
# {('canvas' or 'thumbnail', key), ...}
|
|
self._inrequire=set()
|
|
# semaphore lock for _inrequire
|
|
self._inrequirerun=allocate_lock()
|
|
# thread-safe lock for _inrequire
|
|
self._inrequirelock=allocate_lock()
|
|
# {taskid:(args, kwds), ...}
|
|
# only for sdl user event
|
|
# any method on _tasks should be locked by _tasklock
|
|
self._tasks={}
|
|
# {key:canvas, ...}
|
|
# any method on _canvases should be in mainthread
|
|
self._canvases=OrderedDict()
|
|
# {key:thumbnail, ...}
|
|
# any method on _thumbnails should be in mainthread
|
|
self._thumbnails={}
|
|
# {container:[pathname, ...], ...}
|
|
# any method on _containers should be locked by _containerlock
|
|
self._containers=Containers(
|
|
key=sorter_numeric if config.user_interface.numsort else None)
|
|
# [(target,args,kwds), ...]
|
|
# any method on _actions should be in mainthread
|
|
self._actions=[]
|
|
# True for 'linear', False for 'nearest'
|
|
self._anti_alias=config.user_interface.antialias
|
|
# integer, 0 for autofit, positive for upscale, negative for downscale
|
|
self._scale_ratio=0
|
|
# True to use 'page' size instead of 'pixels' size
|
|
self._use_page=config.user_interface.usepage
|
|
# offset of canvas to center
|
|
self._canvas_offset=(0,0)
|
|
# previous value of canvas offset
|
|
self._canvas_offset_old=(0,0)
|
|
# size of thumbnail in tiling view
|
|
self._thumbsize=config.thumbnail.size
|
|
# max size of thumbnail in tiling view
|
|
self._maxthumbsize=config.thumbnail.maxsize
|
|
# gap between thumbnails in tiling view
|
|
self._thumbgap=config.thumbnail.gap
|
|
# whether to draw checker
|
|
self._checker=config.style.checker
|
|
# color of selector in tiling view
|
|
self._fgcolor=(*config.style.fgcolor,)
|
|
# amount of canvas to keep as cache
|
|
self._cachesize=config.user_interface.cache
|
|
# current view mode
|
|
self._viewmode=SDL_SDLUI_VIEWTILING \
|
|
if config.user_interface.tiling else SDL_SDLUI_VIEWCANVAS
|
|
# thumbnail amounts in each row in tiling view
|
|
self._tiling_width=0
|
|
# True to draw two canvases in canvas view
|
|
self._double_canvas=config.user_interface.double
|
|
# True if 1st page is in right in double canvas mode
|
|
self._double_right=config.user_interface.rtl
|
|
# size of statusbar, AttrDict(size,disabled)
|
|
self._statusbar_size=AttrDict(size=config.statusbar.size,
|
|
disabled=config.statusbar.size==0)
|
|
if self._statusbar_size.size:
|
|
self._statusbar_size.size=max(12,self._statusbar_size.size)
|
|
# position of statusbar, {'land':pos,'port':pos}
|
|
self._statusbar_pos={
|
|
s:{'top':0,'right':1,'bottom':2,'left':3,}[config.statusbar[f'pos{s}']]
|
|
for s in ('land','port')}
|
|
# cache value of orientation
|
|
self._orientation='port'
|
|
# max amount of blocks in statusbar for container and filename
|
|
self._blklmt=5
|
|
# keys of the canvas that is on scene currently. key is that in '_canvases'
|
|
# any method on _scene should be in mainthread
|
|
self._scene=set()
|
|
# unique id of scene, or None if no key in scene
|
|
# any method on _scene_id should be in mainthread
|
|
self._scene_id=None
|
|
# position of current scene
|
|
self._scene_pos=None
|
|
# keys of the canvas that is already started with anime.
|
|
# any method on _animes should be in mainthread
|
|
self._animes=set()
|
|
# unique id of motion, or no motion is in use
|
|
# any method on _motion_id should be in mainthread
|
|
self._motion_id=None
|
|
# current motion status
|
|
self._motion_status=(0,0)
|
|
# stype of motion indicator
|
|
self._motion_indicator=AttrDict(
|
|
size=config.motionbar.size,
|
|
**{s:{'disabled':None,'top':0,'left':0,'bottom':1,'right':1}[
|
|
config.motionbar[f'pos{s}axis']
|
|
] for s in 'xy'}
|
|
)
|
|
|
|
@property
|
|
def stop(self):
|
|
return self._stop
|
|
|
|
@stop.setter
|
|
def stop(self,value):
|
|
self._stop=not not value
|
|
|
|
@property
|
|
def listener(self):
|
|
# the listener instance
|
|
return self._listener
|
|
|
|
def set_listener(self,listener):
|
|
assert is_mainthread()
|
|
self._listener=listener
|
|
|
|
@property
|
|
def window(self):
|
|
# the window instance
|
|
return self._window
|
|
|
|
def set_window(self,window):
|
|
assert is_mainthread()
|
|
self._window=window
|
|
|
|
@property
|
|
def renderer(self):
|
|
# the renderer instance
|
|
return self._window.renderer
|
|
|
|
@property
|
|
def containers(self):
|
|
return self._containers
|
|
|
|
@property
|
|
def anti_alias(self):
|
|
assert is_mainthread()
|
|
return self._anti_alias
|
|
|
|
@anti_alias.setter
|
|
def anti_alias(self,value):
|
|
assert is_mainthread()
|
|
self._anti_alias=not not value
|
|
assert trace(__name__,f'{self}: anti_alias:',
|
|
'on' if self._anti_alias else 'off')
|
|
self.runactions()
|
|
|
|
@property
|
|
def scale_ratio(self):
|
|
assert is_mainthread()
|
|
return self._scale_ratio
|
|
|
|
@scale_ratio.setter
|
|
def scale_ratio(self,value):
|
|
assert is_mainthread()
|
|
if self._scale_ratio==value:return
|
|
self._scale_ratio=value
|
|
assert trace(__name__,f'{self}: scale ratio: {self._scale_ratio}')
|
|
self.runactions()
|
|
|
|
def scale_up(self):
|
|
assert is_mainthread()
|
|
scale_ratio=self.scale_ratio
|
|
while (scale_ratio:=scale_ratio+1) in (-1,0):pass
|
|
self.scale_ratio=scale_ratio
|
|
|
|
def scale_down(self):
|
|
assert is_mainthread()
|
|
scale_ratio=self.scale_ratio
|
|
while (scale_ratio:=scale_ratio-1) in (-1,0):pass
|
|
self.scale_ratio=scale_ratio
|
|
|
|
@property
|
|
def use_page(self):
|
|
assert is_mainthread()
|
|
return self._use_page
|
|
|
|
@use_page.setter
|
|
def use_page(self,value):
|
|
assert is_mainthread()
|
|
self._use_page=not not value
|
|
assert trace(__name__,f'{self}: use_page:',
|
|
'on' if self._use_page else 'off')
|
|
self.runactions()
|
|
|
|
@property
|
|
def canvas_offset(self):
|
|
assert is_mainthread()
|
|
return self._canvas_offset
|
|
|
|
@canvas_offset.setter
|
|
def canvas_offset(self,value):
|
|
assert is_mainthread()
|
|
x,y=value
|
|
if self._canvas_offset==(x,y):return
|
|
self._canvas_offset_old=self._canvas_offset
|
|
self._canvas_offset=x,y
|
|
self.runactions()
|
|
|
|
@property
|
|
def canvas_offset_old(self):
|
|
assert is_mainthread()
|
|
return self._canvas_offset_old
|
|
|
|
def canvas_move(self,x,y):
|
|
assert is_mainthread()
|
|
self.motion_status=x,y
|
|
if not (x or y):return False
|
|
self.canvas_offset=map(sum,zip(self.canvas_offset,(x,y)))
|
|
return self.canvas_offset!=self.canvas_offset_old
|
|
|
|
@property
|
|
def thumbsize(self):
|
|
return self._thumbsize
|
|
|
|
@thumbsize.setter
|
|
def thumbsize(self,value):
|
|
assert is_mainthread()
|
|
self._thumbsize=value
|
|
self.runactions()
|
|
|
|
@property
|
|
def maxthumbsize(self):
|
|
return self._maxthumbsize
|
|
|
|
@property
|
|
def thumbgap(self):
|
|
return self._thumbgap
|
|
|
|
@property
|
|
def checker(self):
|
|
return self._checker
|
|
|
|
def toggle_checker(self):
|
|
assert is_mainthread()
|
|
self._checker=not self._checker
|
|
self.runactions()
|
|
|
|
@property
|
|
def fgcolor(self):
|
|
return self._fgcolor
|
|
|
|
@property
|
|
def cachesize(self):
|
|
return self._cachesize
|
|
|
|
@property
|
|
def viewmode(self):
|
|
return self._viewmode
|
|
|
|
@viewmode.setter
|
|
def viewmode(self,value):
|
|
assert is_mainthread()
|
|
if value==self._viewmode:return
|
|
self._viewmode=value
|
|
self.listener.speak('purge')
|
|
assert note(__name__,f'{self}: view',
|
|
'tiling' if value==SDL_SDLUI_VIEWTILING else 'canvas')
|
|
|
|
@property
|
|
def tiling_width(self):
|
|
return self._tiling_width
|
|
|
|
@tiling_width.setter
|
|
def tiling_width(self,value):
|
|
self._tiling_width=value
|
|
|
|
@property
|
|
def double_canvas(self):
|
|
return self._double_canvas
|
|
|
|
@double_canvas.setter
|
|
def double_canvas(self,value):
|
|
assert is_mainthread()
|
|
self._double_canvas=not not value
|
|
self.runactions()
|
|
|
|
@property
|
|
def double_right(self):
|
|
return self._double_right
|
|
|
|
@double_right.setter
|
|
def double_right(self,value):
|
|
assert is_mainthread()
|
|
self._double_right=not not value
|
|
self.runactions()
|
|
|
|
@property
|
|
def statusbar_size(self):
|
|
if self._statusbar_size.disabled:return 0
|
|
return self._statusbar_size.size
|
|
|
|
def toggle_statusbar(self):
|
|
assert is_mainthread()
|
|
if not self._statusbar_size.size:
|
|
# permanently disabled in config
|
|
return
|
|
self._statusbar_size.disabled=not self._statusbar_size.disabled
|
|
self.runactions()
|
|
|
|
@property
|
|
def orientation(self):
|
|
return self._orientation
|
|
|
|
@orientation.setter
|
|
def orientation(self,value):
|
|
self._orientation=value
|
|
|
|
@property
|
|
def blklmt(self):
|
|
return self._blklmt
|
|
|
|
@property
|
|
def statusbar_pos(self):
|
|
return self._statusbar_pos[self.orientation]
|
|
|
|
def rotate_statsbar(self,reverse=False):
|
|
assert is_mainthread()
|
|
n=-1 if reverse else 1
|
|
self._statusbar_pos[self.orientation]=(self.statusbar_pos+n)%4
|
|
self.runactions()
|
|
|
|
@property
|
|
def motion_status(self):
|
|
assert is_mainthread()
|
|
if self._motion_id is None:return None
|
|
if not any(self._motion_status):return None
|
|
return self._motion_status
|
|
|
|
@motion_status.setter
|
|
def motion_status(self,value):
|
|
assert is_mainthread()
|
|
x,y=value
|
|
self._motion_status=(x,y)
|
|
|
|
@property
|
|
def motion_indicator(self):
|
|
return self._motion_indicator.size,\
|
|
self._motion_indicator.x,self._motion_indicator.y
|
|
|
|
def action_quit(self):
|
|
event=SDL_Event()
|
|
SDL_memset(byref(event),0,sizeof(event))
|
|
event.type=SDL_QUIT
|
|
SDL_PushEvent(byref(event))
|
|
assert trace(__name__,f'{self}: send quit event')
|
|
|
|
def action_view(self):
|
|
self.puttask(SDL_SDLUI,self.viewmode)
|
|
|
|
def action_view_tiling(self):
|
|
assert is_mainthread()
|
|
if self.containers.current_page is None:
|
|
self.make_scenario_empty()
|
|
elif not (self.viewmode==SDL_SDLUI_VIEWTILING and self.scene_len):
|
|
self.viewmode=SDL_SDLUI_VIEWTILING
|
|
self.make_scenario_tiling()
|
|
self.puttask(SDL_SDLUI,SDL_SDLUI_CANVASDRAW)
|
|
|
|
def action_view_canvas(self):
|
|
assert is_mainthread()
|
|
if self.containers.current_page is None:
|
|
self.make_scenario_empty()
|
|
elif not (self.viewmode==SDL_SDLUI_VIEWCANVAS and self.scene_len):
|
|
self.viewmode=SDL_SDLUI_VIEWCANVAS
|
|
self.make_scenario_canvas()
|
|
self.puttask(SDL_SDLUI,SDL_SDLUI_CANVASDRAW)
|
|
|
|
def action_cross(self,value):
|
|
assert is_mainthread()
|
|
if not self.containers.cross(value):return
|
|
self.listener.speak('purge')
|
|
self.action_view()
|
|
assert note(__name__,f'{self}: cross to {self.containers.current_page[0]}')
|
|
|
|
def action_motion_start(self,*,x=0,y=0):
|
|
assert is_mainthread()
|
|
if not (x or y):return
|
|
motion_id=urandom(256).__hash__()
|
|
self._motion_id=motion_id
|
|
assert trace(__name__,f'{self}: motion start')
|
|
self.puttask(SDL_SDLUI,SDL_SDLUI_MOTIONCONTINUE,(x,y,motion_id))
|
|
|
|
def action_motion_stop(self):
|
|
assert is_mainthread()
|
|
if self._motion_id is None:return
|
|
self._motion_id=None
|
|
self.runactions()
|
|
|
|
def match_event(self,event):
|
|
# match and process event in mainthread, return True to breakout looper
|
|
assert is_mainthread()
|
|
match event.type:
|
|
case SDLE_EvEmu.SDL_QUIT:
|
|
assert trace(__name__,f'{self}: SDL_QUIT')
|
|
self.stop=1
|
|
self.listener.speak('over')
|
|
return True
|
|
case SDLE_EvEmu.SDL_KEYDOWN|SDLE_EvEmu.SDL_KEYUP:
|
|
self._hci.handle_key_event(event.key)
|
|
case SDLE_EvEmu.SDL_MOUSEBUTTONDOWN|SDLE_EvEmu.SDL_MOUSEBUTTONUP:
|
|
self._hci.handle_mousebtn_event(event.button)
|
|
case SDLE_EvEmu.SDL_MOUSEMOTION:
|
|
self._hci.handle_mousemtn_event(event.motion)
|
|
case SDLE_EvEmu.SDL_MOUSEWHEEL:
|
|
self._hci.handle_mousewhl_event(event.wheel)
|
|
case SDLE_EvEmu.SDL_FINGERDOWN|SDLE_EvEmu.SDL_FINGERUP:
|
|
self._hci.handle_touchact_event(event.tfinger)
|
|
case SDLE_EvEmu.SDL_WINDOWEVENT:
|
|
if event.window.event==SDL_WINDOWEVENT_EXPOSED:
|
|
self.window.update()
|
|
self.runactions()
|
|
case SDLE_EvEmu.SDL_RENDER_TARGETS_RESET:
|
|
self.runactions()
|
|
case SDLE_EvEmu.SDL_RENDER_DEVICE_RESET:
|
|
self.purgecanvases()
|
|
self.purgethumbnails()
|
|
self.runactions()
|
|
case SDLE_EvEmu.SDL_WAND:
|
|
evtype=event.type
|
|
evcode=event.user.code
|
|
data,taskid=SDLE_GetUserEventData(event)
|
|
args,kwds=self.gettask(taskid)
|
|
SDLE_DelUserEventData(event)
|
|
match evcode:
|
|
case SDLE_EvEmu.SDL_WAND_LOADINGSTARTED:
|
|
path=unpackstr(data)
|
|
n_loop,n_frames=args
|
|
canvas=self.renderer.new_canvas(n_loop=n_loop,n_frames=n_frames,
|
|
name=prettypath(*path))
|
|
self.putcanvas(path,canvas)
|
|
case SDLE_EvEmu.SDL_WAND_THUMBNAILSTARTED:
|
|
path=unpackstr(data)
|
|
thumbnail=self.renderer.new_thumbnail(name=prettypath(*path))
|
|
self.putthumbnail(path,thumbnail)
|
|
case SDLE_EvEmu.SDL_WAND_FRAMEDECODED:
|
|
path=unpackstr(data)
|
|
if (canvas:=self.getcanvas(path)) is None:return
|
|
n,pixfmt,frameprop,databuffer=args
|
|
canvas.add_frame(n,pixfmt,frameprop,databuffer)
|
|
case SDLE_EvEmu.SDL_WAND_THUMBNAILDECODED:
|
|
path=unpackstr(data)
|
|
if (thumbnail:=self.getthumbnail(path)) is None:return
|
|
if thumbnail.ready:return
|
|
n,pixfmt,frameprop,databuffer=args
|
|
thumbnail.add_frame(n,pixfmt,frameprop,databuffer)
|
|
case SDLE_EvEmu.SDL_WAND_LOADINGFINISHED:
|
|
path=unpackstr(data)
|
|
if self.viewmode==SDL_SDLUI_VIEWCANVAS and \
|
|
self.is_in_scene(path):
|
|
self.puttask(SDL_SDLUI,SDL_SDLUI_CANVASDRAW)
|
|
case SDLE_EvEmu.SDL_WAND_THUMBNAILFINISHED:
|
|
path=unpackstr(data)
|
|
if self.getthumbnail(path) is None and \
|
|
(canvas:=self.getcanvas(path)) is not None:
|
|
assert note(__name__,f'{self}: preview {prettypath(*path)}')
|
|
self.putthumbnail(path,canvas.create_thumbnail())
|
|
if self.viewmode==SDL_SDLUI_VIEWTILING and \
|
|
self.is_in_scene(path):
|
|
self.puttask(SDL_SDLUI,SDL_SDLUI_CANVASDRAW,data=data)
|
|
case SDLE_EvEmu.SDL_SDLUI:
|
|
evtype=event.type
|
|
evcode=event.user.code
|
|
data,taskid=SDLE_GetUserEventData(event)
|
|
args,kwds=self.gettask(taskid)
|
|
SDLE_DelUserEventData(event)
|
|
match evcode:
|
|
case SDLE_EvEmu.SDL_SDLUI_VIEWCANVAS:
|
|
self.action_view_canvas()
|
|
case SDLE_EvEmu.SDL_SDLUI_VIEWTILING:
|
|
self.action_view_tiling()
|
|
case SDLE_EvEmu.SDL_SDLUI_CANVASDRAW:
|
|
self.runactions()
|
|
case SDLE_EvEmu.SDL_SDLUI_ANIMECONTINUE:
|
|
path=unpackstr(data)
|
|
scene_id,loop_count=args
|
|
if scene_id!=self.get_scene_id() or not self.is_in_scene(path):
|
|
assert note(__name__,f'{self}: anime stop',
|
|
prettypath(*path))
|
|
self._animes.discard(path)
|
|
return
|
|
if (canvas:=self.getcanvas(path)) is None:
|
|
self._animes.discard(path)
|
|
return
|
|
if not canvas.next_frame() and canvas.n_loop:
|
|
loop_count+=1
|
|
if loop_count>=canvas.n_loop:
|
|
assert note(__name__,f'{self}: anime stop',
|
|
prettypath(*path))
|
|
self._animes.discard(path)
|
|
return
|
|
self.runactions()
|
|
start_new_thread(
|
|
self.update_anime,
|
|
(path,scene_id,loop_count,canvas.current_delay()))
|
|
case SDLE_EvEmu.SDL_SDLUI_MOTIONCONTINUE:
|
|
x,y,motion_id=args
|
|
if motion_id!=self._motion_id:
|
|
assert verb(__name__,f'{self}: motion stop')
|
|
return
|
|
if not self.canvas_move(x,y):
|
|
assert verb(__name__,f'{self}: motion interrupted at edge')
|
|
self._motion_id=None
|
|
else:
|
|
self.puttask(SDL_SDLUI,SDL_SDLUI_MOTIONCONTINUE,
|
|
(x,y,motion_id))
|
|
case SDLE_EvEmu.SDL_TEXTINPUT|SDLE_EvEmu.SDL_TEXTEDITING:
|
|
pass # ignore
|
|
case _ as evtype:
|
|
assert warn(__name__,f'{self}: unknown event:',
|
|
SDLE_GetEventTypeName(evtype))
|
|
|
|
def looper(self,window):
|
|
assert is_mainthread()
|
|
event=SDL_Event()
|
|
SDL_memset(byref(event),0,sizeof(event))
|
|
assert trace(__name__,f'{self}: loop in')
|
|
self.set_window(window)
|
|
while True:
|
|
runsdl(SDL_WaitEvent,event,errval=false1)
|
|
try:
|
|
if self.match_event(event):
|
|
break
|
|
except Exception as e:
|
|
assert error(__name__,f'{self}: error',exc=e)
|
|
self.set_window(None)
|
|
assert trace(__name__,f'{self}: loop out')
|
|
|
|
def anime_starter(self):
|
|
if self.viewmode==SDL_SDLUI_VIEWTILING:return
|
|
for path in self.iter_scene():
|
|
if path in self._animes:continue
|
|
if (canvas:=self.getcanvas(path)) is None:continue
|
|
if not canvas.ready:continue
|
|
self._animes.add(path)
|
|
if canvas.n_frames<2:continue
|
|
canvas.first_frame()
|
|
assert note(__name__,f'{self}: anime start',prettypath(*path))
|
|
start_new_thread(self.update_anime,
|
|
(path,self.get_scene_id(),0,canvas.current_delay()))
|
|
return True
|
|
|
|
def update_anime(self,path,scene_id,loop_count,delay=0):
|
|
# delay is in millisecond
|
|
assert trace(__name__,f'{self}: anime delay {delay}',path)
|
|
if delay:
|
|
SDL_Delay(delay)
|
|
self.puttask(SDL_SDLUI,SDL_SDLUI_ANIMECONTINUE,
|
|
args=(scene_id,loop_count),data=packstr(*path))
|
|
|
|
def make_scenario_empty(self):
|
|
self.clearactions()
|
|
self.start_scene()
|
|
self.addaction(self.renderer.clear)
|
|
self.addaction(self.clear_scene)
|
|
self.addaction(draw_empty,(self,))
|
|
self.addaction(self.renderer.present)
|
|
assert trace(__name__,f'{self}: make canvas empty')
|
|
|
|
def make_scenario_canvas(self):
|
|
self.clearactions()
|
|
self.start_scene()
|
|
self.addaction(self.renderer.clear)
|
|
self.addaction(self.clear_scene)
|
|
self.addaction(draw_canvas,(self,))
|
|
self.addaction(draw_status,(self,))
|
|
self.addaction(self.renderer.present)
|
|
assert trace(__name__,f'{self}: make canvas scenario')
|
|
|
|
def make_scenario_tiling(self):
|
|
self.clearactions()
|
|
self.start_scene()
|
|
self.addaction(self.renderer.clear)
|
|
self.addaction(self.clear_scene)
|
|
self.addaction(draw_tiling,(self,))
|
|
self.addaction(draw_status,(self,))
|
|
self.addaction(self.renderer.present)
|
|
assert trace(__name__,f'{self}: make tiling scenario')
|
|
|
|
def get_renderer_area_for_canvas(self):
|
|
# get width,height,offset_x,offset_y of renderer area for canvas
|
|
w,h=self.renderer.output_size
|
|
self.renderer.size=w,h
|
|
self.orientation='land' if w>h else 'port'
|
|
x=y=0
|
|
if size:=self.statusbar_size:
|
|
match self.statusbar_pos:
|
|
case 0: # top
|
|
y=size
|
|
h-=size
|
|
case 1: # right
|
|
w-=size
|
|
case 2: # bottom
|
|
h-=size
|
|
case 3: # left
|
|
x=size
|
|
w-=size
|
|
return (x,y,w,h)
|
|
|
|
def get_renderer_area_for_statusbar(self):
|
|
# get width,height,offset_x,offset_y,is_landscape of renderer area for statusbar
|
|
w,h=self.renderer.output_size
|
|
self.renderer.size=w,h
|
|
self.orientation='land' if w>h else 'port'
|
|
if not (size:=self.statusbar_size):
|
|
return 0,0,0,0,False
|
|
x=y=0
|
|
match self.statusbar_pos:
|
|
case 0: # top
|
|
h=size
|
|
case 1: # right
|
|
x=w-size-1
|
|
w=size
|
|
case 2: # bottom
|
|
y=h-size-1
|
|
h=size
|
|
case 3: # left
|
|
w=size
|
|
return x,y,w,h,self.statusbar_pos in (0,2)
|
|
|
|
def set_anti_alias(self,canvas):
|
|
canvas.scale_mode='linear' if self.anti_alias else 'nearest'
|
|
|
|
def clear(self):
|
|
self.clearactions()
|
|
self.purgetasks()
|
|
self.purgecanvases()
|
|
self.purgethumbnails()
|
|
assert trace(__name__,f'{self}: clear')
|
|
|
|
def push_user_event(self,evtype,code,data=None,data2=None):
|
|
if data is None:data=b''
|
|
assert isinstance(data,bytes),f'unsupported type of data: {type(data)}'
|
|
event=SDL_Event()
|
|
SDL_memset(byref(event),0,sizeof(event))
|
|
event.type=evtype
|
|
event.user.code=code
|
|
SDLE_SetUserEventData(event,data,data2)
|
|
SDL_PushEvent(byref(event))
|
|
|
|
def gettask(self,taskid):
|
|
with self._tasklock:
|
|
try:
|
|
return self._tasks.pop(taskid)
|
|
except (KeyError,IndexError) as e:
|
|
assert error(__name__,f'{self}: no task for {prettybytes(taskid)}',
|
|
exc=e)
|
|
return (),{}
|
|
def puttask(self,evtype,evev,args=(),kwds={},data=None):
|
|
with self._tasklock:
|
|
if self.stop==1:return
|
|
self._tasks[taskid:=urandom(64)]=(args,kwds)
|
|
self.push_user_event(evtype,evev,data,taskid)
|
|
def purgetask(self,taskid):
|
|
with self._tasklock:
|
|
del self._tasks[taskid]
|
|
def purgetasks(self):
|
|
with self._tasklock:
|
|
self._tasks.clear()
|
|
|
|
def hasrequire(self,key,canvas=True):
|
|
with self._inrequirelock:
|
|
return ('canvas' if canvas else 'thumbnail',key) in self._inrequire
|
|
def addrequire(self,key,canvas=True,cutin=False):
|
|
reqkey=('canvas' if canvas else 'thumbnail',key)
|
|
with self._inrequirelock:
|
|
if ('canvas',key) in self._inrequire:
|
|
return
|
|
if not canvas and ('thumbnail',key) in self._inrequire:
|
|
return
|
|
if not self._inrequire:
|
|
self._inrequirerun.acquire()
|
|
self._inrequire.add(('thumbnail',key))
|
|
if canvas:
|
|
self._inrequire.add(('canvas',key))
|
|
self.listener.speak('load' if canvas else 'thum',*key,
|
|
b'T' if cutin else b'')
|
|
def cancelrequire(self,key,is_canvas=True,/):
|
|
with self._inrequirelock:
|
|
if not self._inrequire:
|
|
return
|
|
self._inrequire.discard(('thumbnail',key))
|
|
if is_canvas:
|
|
self._inrequire.discard(('canvas',key))
|
|
if not self._inrequire:
|
|
self._inrequirerun.release()
|
|
|
|
def _purge_first_canvas(self):
|
|
assert is_mainthread()
|
|
key,value=self._canvases.popitem(0)
|
|
if value is not None:
|
|
value.close()
|
|
def hascanvas(self,key):
|
|
assert is_mainthread()
|
|
return key in self._canvases or self.hasrequire(key)
|
|
def getcanvas(self,key,cutin=False):
|
|
assert is_mainthread()
|
|
if key in self._canvases:
|
|
self._canvases.move_to_end(key,last=True)
|
|
return self._canvases.get(key,None)
|
|
else:
|
|
self.addrequire(key,cutin=cutin)
|
|
return
|
|
def putcanvas(self,key,value):
|
|
assert is_mainthread()
|
|
assert self._canvases.get(key,None) is None,f'canvas already exists: {key}'
|
|
self._canvases[key]=value
|
|
while len(self._canvases)>self.cachesize+(2 if self.double_canvas else 1):
|
|
self._purge_first_canvas()
|
|
def holdcanvas(self,key):
|
|
self._canvases.setdefault(key,None)
|
|
def purgecanvas(self,key):
|
|
assert is_mainthread()
|
|
try:
|
|
if (value:=self._canvases.pop(key)) is not None:
|
|
value.close()
|
|
except KeyError:
|
|
pass
|
|
def purgecanvases(self):
|
|
assert is_mainthread()
|
|
while self._canvases:
|
|
self._purge_first_canvas()
|
|
|
|
def hasthumbnail(self,key):
|
|
assert is_mainthread()
|
|
return key in self._thumbnails or self.hasrequire(key,canvas=False)
|
|
def getthumbnail(self,key,cutin=False):
|
|
assert is_mainthread()
|
|
if key not in self._thumbnails:
|
|
self.addrequire(key,canvas=False,cutin=cutin)
|
|
return self._thumbnails.get(key,None)
|
|
def putthumbnail(self,key,value):
|
|
assert is_mainthread()
|
|
assert self._thumbnails.get(key,None) is None,f'thumbnail already exists: {key}'
|
|
self._thumbnails[key]=value
|
|
def holdthumbnail(self,key):
|
|
self._thumbnails.setdefault(key,None)
|
|
def purgethumbnail(self,key):
|
|
assert is_mainthread()
|
|
try:
|
|
if (value:=self._thumbnails.pop(key)) is not None:
|
|
value.close()
|
|
except KeyError:
|
|
pass
|
|
def purgethumbnails(self):
|
|
assert is_mainthread()
|
|
for key in set(self._thumbnails.keys()):
|
|
self.purgethumbnail(key)
|
|
|
|
def hascontainer(self,key):
|
|
with self._containerlock:
|
|
return key in self.containers
|
|
def getcontainer(self,key):
|
|
with self._containerlock:
|
|
try:
|
|
return self.containers[key]
|
|
except KeyError:
|
|
pass
|
|
def delpath(self,container,filename=None,/):
|
|
with self._containerlock:
|
|
if filename is not None:
|
|
return self.containers.remove_filename(container,filename)
|
|
try:
|
|
self.containers[container].clear()
|
|
del self.containers[container]
|
|
except KeyError:
|
|
pass
|
|
def putcontainer(self,key,value):
|
|
with self._containerlock:
|
|
self.containers[key]=value
|
|
def purgecontainers(self):
|
|
with self._containerlock:
|
|
self.containers.clear()
|
|
|
|
def addaction(self,target,args=(),kwds={}):
|
|
assert is_mainthread()
|
|
self._actions.append((target,args,kwds))
|
|
def hasactions(self):
|
|
assert is_mainthread()
|
|
return not not self._actions
|
|
def runactions(self):
|
|
assert is_mainthread()
|
|
for target,args,kwds in self._actions:
|
|
try:
|
|
target(*args,**kwds)
|
|
except Exception as e:
|
|
assert warn(__name__,f'{self}: action failed: {target.__name__}',exc=e)
|
|
def clearactions(self):
|
|
assert is_mainthread()
|
|
self._actions.clear()
|
|
def start_scene(self):
|
|
assert is_mainthread()
|
|
self._scene_id=urandom(256).__hash__()
|
|
def get_scene_id(self):
|
|
assert is_mainthread()
|
|
return self._scene_id
|
|
def add_to_scene(self,*keys):
|
|
assert is_mainthread()
|
|
self._scene.update(keys)
|
|
def remove_from_scene(self,key):
|
|
assert is_mainthread()
|
|
self._scene.remove(key)
|
|
def iter_scene(self):
|
|
return iter(self._scene)
|
|
def is_in_scene(self,key):
|
|
assert is_mainthread()
|
|
return key in self._scene
|
|
@property
|
|
def scene_len(self):
|
|
assert is_mainthread()
|
|
return len(self._scene)
|
|
def clear_scene(self):
|
|
assert is_mainthread()
|
|
self._scene.clear()
|
|
|
|
|
|
# Local Variables:
|
|
# coding: utf-8
|
|
# mode: python
|
|
# python-indent-offset: 4
|
|
# indent-tabs-mode: nil
|
|
# End:
|