master
yumoqing 2022-03-07 11:53:31 +08:00
parent 70505676e7
commit e412a15e25
2 changed files with 1361 additions and 0 deletions

View File

@ -0,0 +1,742 @@
import os
import sys
import codecs
import ujson as json
from traceback import print_exc
from functools import partial
from appPublic.dictExt import dictExtend
from appPublic.folderUtils import ProgramPath
from appPublic.dictObject import DictObject
from appPublic.Singleton import SingletonDecorator, GlobalEnv
from appPublic.datamapping import keyMapping
from appPublic.registerfunction import RegisterFunction
from kivy.logger import Logger
from kivy.config import Config
from kivy.metrics import sp,dp,mm
from kivy.core.window import WindowBase, Window
from kivy.properties import BooleanProperty
from kivy.uix.widget import Widget
from kivy.uix.modalview import ModalView
from kivy.app import App
from kivy.factory import Factory
from kivy.uix.video import Video
from .utils import *
from .newvideo import Video
from .orientationlayout import OrientationLayout
from .threadcall import HttpClient
from .register import *
class WidgetNotFoundById(Exception):
def __init__(self, id):
super().__init__()
self.idstr = id
def __str__(self):
return "Widget not found by id:" + self.idstr + ':'
def __expr__(self):
return str(self)
class ClassMethodNotFound(Exception):
def __init__(self,k,m):
super().__init__()
self.kname = k
self.mname = m
def __str__(self):
s = 'Method(%s) not found in class(%s)' % (self.mname,
str(self.kname.__classname__))
return s
def __expr__(self):
return self.__str__()
class NotExistsObject(Exception):
def __init__(self,name):
super().__init__()
self.name = name
def __str__(self):
s = 'not exists widget(%s)' % self.name
return s
def __expr__(self):
return self.__str__()
class ArgumentError(Exception):
def __init__(self,argument,desc):
super().__init__()
self.argument = argument
self.desc = desc
def __str__(self):
s = 'argument(%s) missed:%s' % (self.argument,self.desc)
return s
def __expr__(self):
return self.__str__()
class NotRegistedWidget(Exception):
def __init__(self, name:str):
super().__init__()
self.widget_name = name
def __str__(self):
s = 'not reigsted widget(%s)' % self.name
return s
def __expr__(self):
return self.__str__()
def registerWidget(name:str,widget):
globals()[name] = widget
class Blocks(EventDispatcher):
def __init__(self):
EventDispatcher.__init__(self)
self.action_id = 0
self.register_event_type('on_built')
self.register_event_type('on_failed')
self.env = GlobalEnv()
def set(self, k:str, v):
self.env[k] = v
def register_widget(self, name:str, widget:Widget):
globals()[name] = widget
def buildAction(self, widget:Widget, desc):
conform_desc = desc.get('conform')
blocks = Blocks()
if not conform_desc:
return partial(blocks.uniaction, widget, desc)
func =partial(blocks.conform_action,widget, desc)
return func
def eval(self, s:str, l:dict):
g = {}
forbidens = [
"os",
"sys",
"codecs",
"json",
]
for k,v in globals().copy().items():
if k not in forbidens:
g[k] = v
"""
g['__builtins__'] = globals()['__builtins__']
g['__builtins__']['__import__'] = None
g['__builtins__']['__loader__'] = None
g['__builtins__']['open'] = None
"""
g.update(self.env)
return eval(s,g,l)
def getUrlData(self, url:str, method:str='GET',
params:dict={}, files:dict={},
callback=None,
errback=None,**kw):
if url is None:
if errback:
errback(None,Exception('url is None'))
else:
return None
if url.startswith('file://'):
filename = url[7:]
with codecs.open(filename,'r','utf-8') as f:
b = f.read()
dic = json.loads(b)
return dic
elif url.startswith('http://') or url.startswith('https://'):
try:
hc = HttpClient()
resp=hc(url,method=method,params=params,files=files)
# print('Blocks.py :resp=',resp)
return resp
except Exception as e:
print_exc()
if errback:
return errback(None,e)
return None
else:
config = getConfig()
url = config.uihome + url
return self.getUrlData(url,method=method,
params=params,
files=files,
**kw)
def strValueExpr(self,s:str,localnamespace:dict={}):
if not s.startswith('py::'):
return s
s = s[4:]
try:
v = self.eval(s,localnamespace)
return v
except Exception as e:
print('Exception .... ',e,'script=',s)
print_exc()
return s
def arrayValueExpr(self,arr:list,localnamespace:dict={}) -> list:
d = []
for v in arr:
if type(v) == type(''):
d.append(self.strValueExpr(v,localnamespace))
continue
if type(v) == type([]):
d.append(self.arrayValueExpr(v,localnamespace))
continue
if type(v) == type({}):
d.append(self.dictValueExpr(v,localnamespace))
continue
if type(v) == type(DictObject):
d.append(self.dictValueExpr(v,localnamespace))
continue
d.append(v)
return d
def dictValueExpr(self,dic:dict,localnamespace:dict={}) -> dict:
d = {}
for k,v in dic.items():
if type(v) == type(''):
d[k] = self.strValueExpr(v,localnamespace)
continue
if type(v) == type([]):
d[k] = self.arrayValueExpr(v,localnamespace)
continue
if type(v) == type({}):
d[k] = self.dictValueExpr(v,localnamespace)
continue
if type(v) == type(DictObject):
d[k] = self.dictValueExpr(v,localnamespace)
continue
d[k] = v
return d
def valueExpr(self,obj,localnamespace:dict={}):
if type(obj) == type(''):
return self.strValueExpr(obj,localnamespace)
if type(obj) == type([]):
return self.arrayValueExpr(obj,localnamespace)
if type(obj) == type({}):
return self.dictValueExpr(obj,localnamespace)
if isinstance(obj,DictObject):
return self.dictValueExpr(obj,localnamespace)
return obj
def w_build(self,desc) -> Widget:
# print('w_build(),desc=',desc)
widgetClass = desc.get('widgettype',None)
if not widgetClass:
Logger.info("Block: w_build(), desc invalid", desc)
raise Exception(desc)
widgetClass = desc['widgettype']
opts = self.valueExpr(desc.get('options',{}).copy())
widget = None
try:
klass = Factory.get(widgetClass)
widget = klass(**opts)
except Exception as e:
print('Error:',widgetClass,'contructon error')
print_exc()
raise NotExistsObject(widgetClass)
if desc.get('id'):
id = desc.get('id')
if id.startswith('app.'):
app = App.get_running_app()
id = id[4:]
setattr(app, id, widget)
if id.startswith('root.'):
app = App.get_running_app()
id = id[5:]
setattr(app.root, id, widget)
if '.' in id:
Logger.info('widget id(%s) can not contain "."', id)
else:
widget.widget_id = id
widget.build_desc = desc
self.build_attributes(widget,desc)
self.build_rest(widget,desc)
return widget
def build_attributes(self, widget:Widget,desc,t=None):
excludes = ['widgettype','options','subwidgets','binds']
for k,v in [(k,v) for k,v in desc.items() if k not in excludes]:
if isinstance(v,dict) and v.get('widgettype'):
b = Blocks()
v = self.valueExpr(v, localnamespace={'self':widget})
w = b.w_build(v)
if hasattr(widget,k):
aw = getattr(widget,k)
if isinstance(aw,Layout):
aw.add_widget(w)
continue
setattr(widget,k,w)
continue
setattr(widget,k,self.valueExpr(v,\
localnamespace={'self':widget}))
def build_rest(self, widget:Widget,desc,t=None):
self.subwidget_total = len(desc.get('subwidgets',[]))
self.subwidgets = [ None for i in range(self.subwidget_total)]
pos = 0
for pos,sw in enumerate(desc.get('subwidgets',[])):
b = Blocks()
kw = self.valueExpr(sw.copy(),
localnamespace={'self':widget})
w = b.widgetBuild(kw)
widget.add_widget(w)
for b in desc.get('binds',[]):
kw = self.valueExpr(b.copy(), \
localnamespace={'self':widget})
self.buildBind(widget,kw)
def buildBind(self, widget:Widget, desc):
wid = desc.get('wid','self')
w = Blocks.getWidgetById(desc.get('wid','self'),from_widget=widget)
if not w:
Logger.info('Block: %s %s',desc.get('wid','self'),
'not found via Blocks.getWidgetById()')
return
event = desc.get('event')
if event is None:
Logger.info('Block: binds desc miss event, desc=%s',str(desc))
return
f = self.buildAction(widget,desc)
if f is None:
Logger.info('Block: get a null function,%s',str(desc))
return
w.bind(**{event:f})
def multipleAction(self, widget:Widget, desc, *args):
desc1 = {k:v for k, v in desc.items() if k != 'actions'}
mydesc = desc1.copy()
for a in desc['actions']:
new_desc = mydesc.copy()
new_desc.update(a)
self.uniaction(widget,new_desc, *args)
def conform_action(self, widget:Widget, desc, *args):
conform_desc = desc.get('conform')
blocks = Blocks()
if not conform_desc:
blocks.uniaction(widget, desc,*args, **kw)
return
w = blocks.widgetBuild({
"widgettype":"Conform",
"options":conform_desc
})
w.bind(on_conform=partial(blocks.uniaction, widget, desc))
w.open()
def uniaction(self, widget:Widget, desc, *args):
acttype = desc.get('actiontype')
if acttype=='blocks':
return self.blocksAction(widget,desc, *args)
if acttype=='urlwidget':
return self.urlwidgetAction(widget,desc, *args)
if acttype == 'registedfunction':
return self.registedfunctionAction(widget,desc, *args)
if acttype == 'script':
return self.scriptAction(widget, desc, *args)
if acttype == 'method':
return self.methodAction(widget, desc, *args)
if acttype == 'event':
return self.eventAction(widget, desc, *args)
if acttype == 'multiple':
return self.multipleAction(widget, desc, *args)
alert("actiontype(%s) invalid" % acttype,title='error')
def eventAction(self, widget:Widget, desc, *args):
target = self.get_target(widget, desc)
event = desc.get('dispatch_event')
if not event:
Logger.info('Block: eventAction():desc(%s) miss dispatch_event',
str(desc))
return
params = desc.get('params',{})
d = self.getActionData(widget,desc, *args)
if d:
params.update(d)
try:
target.dispatch(event, params)
except Exception as e:
Logger.info(f'Block: eventAction():dispatch {event} error')
print_exc()
return
def get_target(self, widget:Widget, desc):
if not desc.get('target'):
return None
return Blocks.getWidgetById(desc.get('target'),from_widget=widget)
def blocksAction(self, widget:Widget, desc, *args):
target = self.get_target(widget, desc)
add_mode = desc.get('mode','replace')
opts = desc.get('options').copy()
d = self.getActionData(widget,desc, *args)
p = opts.get('options',{}).copy()
if d:
p.update(d)
opts['options'] = p
def doit(target:Widget, add_mode:str, o, w:Widget):
if isinstance(w, ModalView):
return
if target and not w.parent:
if add_mode == 'replace':
target.clear_widgets()
target.add_widget(w)
def doerr(o,e):
Logger.info('Block: blocksAction(): desc=%s widgetBuild error'
,str(desc))
raise e
b = Blocks()
b.bind(on_built=partial(doit,target,add_mode))
b.bind(on_failed=doerr)
b.widgetBuild(opts)
def urlwidgetAction(self, widget:Widget, desc, *args):
target = self.get_target(widget, desc)
add_mode = desc.get('mode','replace')
opts = desc.get('options').copy()
p = opts.get('params',{}).copy()
if len(args) >= 1 and isinstance(args[0],dict):
p.update(args[0])
d = self.getActionData(widget, desc, *args)
if d:
p.update(d)
opts['params'] = p
d = {
'widgettype' : 'urlwidget',
'options': opts
}
def doit(target:Widget, add_mode:str, o, w:Widget):
if isinstance(w, ModalView):
return
if target and not w.parent:
if add_mode == 'replace':
target.clear_widgets()
target.add_widget(w)
def doerr(o,e):
Logger.info('Block: urlwidgetAction(): desc=%s widgetBuild error'
,str(desc))
b = Blocks()
b.bind(on_built=partial(doit,target,add_mode))
b.bind(on_failed=doerr)
b.widgetBuild(d)
def getActionData(self, widget:Widget, desc, *args):
Logger.info('Block: getActionData(): desc=%s args=%s'
,str(desc), args)
data = {}
rtdesc = self.build_rtdesc(desc)
if rtdesc:
rt = self.get_rtdata(widget, rtdesc, *args)
if rt:
data.update(rt)
if desc.get('keymapping'):
data = keyMapping(data, desc.get('keymapping'))
Logger.info('getActionData():rtdesc=%s, data=%s', rtdesc, data)
return data
def registedfunctionAction(self, widget:Widget, desc, *args):
target = self.get_target(widget, desc)
rf = RegisterFunction()
name = desc.get('rfname')
func = rf.get(name)
if func is None:
Logger.info('Block: desc=%s rfname(%s) not found',
str(desc), name)
raise Exception('rfname(%s) not found' % name)
params = desc.get('params',{}).copy()
d = self.getActionData(widget,desc, *args)
if d:
params.update(d)
func(target, *args, **params)
def scriptAction(self, widget:Widget, desc, *args):
script = desc.get('script')
if not script:
Logger.info('Block: scriptAction():desc(%s) target not found',
str(desc))
return
target = self.get_target(widget, desc)
d = self.getActionData(widget,desc, *args)
ns = {
"self":target,
"args":args,
"kwargs":d
}
if d:
ns.update(d)
try:
self.eval(script, ns)
except Exception as e:
print_exc()
print(e)
def build_rtdesc(self, desc):
rtdesc = desc.get('rtdata')
if not rtdesc:
if desc.get('datawidget'):
rtdesc = {}
rtdesc['target'] = desc['datawidget']
if desc.get('datascript'):
rtdesc['script'] = desc.get('datacript')
else:
rtdesc['method'] = desc.get('datamethod', 'getValue')
rtdesc['kwargs'] = desc.get('datakwargs', {})
return rtdesc
def get_rtdata(self, widget:Widget, desc, *args):
"""
desc descript follow attributes
{
"target", a target name from widget
"method", a method name, default is 'getValue',
"script", script to handle the data
"kwargs":" a dict arguments, default is {}
}
"""
if desc is None or desc == {}:
print('get_rtdata():desc is None')
return {}
kwargs = desc.get('kwargs', {})
if not isinstance(kwargs, dict):
kwargs = {}
w = None
if len(args) > 0:
w = args[0]
target = desc.get('target')
if target:
w = self.getWidgetById(target, from_widget=widget)
if w is None:
print('get_rtdata():w is None', desc)
return {}
script = desc.get('script')
if script:
ns = {
"self":w,
"args":args,
"kwargs":kwargs
}
d = self.eval(script, ns)
return d
method = desc.get('method', 'getValue')
if not hasattr(w, method):
print('get_rtdata():method is None', desc, type(w))
return {}
f = getattr(w, method)
try:
r = f(**kwargs)
if isinstance(r, dict):
return r
if isinstance(r, DictObject):
return r.to_dict()
print('get_rtdata():method return is not dict', desc, 'ret=', r, type(r))
return {}
except:
print_exc()
return {}
def methodAction(self, widget:Widget, desc, *args):
method = desc.get('method')
target = self.get_target(widget, desc)
if target is None:
Logger.info('Block: methodAction():desc(%s) target not found',
str(desc))
return
if method is None:
Logger.info('Block: methodAction():desc(%s) method not found',
str(desc))
return
if hasattr(target,method):
f = getattr(target, method)
kwargs = desc.get('options',{}).copy()
d = self.getActionData(widget,desc, *args)
if d:
kwargs.update(d)
f(*args, **kwargs)
else:
alert('%s method not found' % method)
def widgetBuild(self, desc):
"""
desc format:
{
widgettype:<registered widget>,
id:widget id,
options:{}
subwidgets:[
]
binds:[
]
}
"""
def doit(desc):
if isinstance(desc,DictObject):
desc = desc.to_dict()
if not isinstance(desc,dict):
Logger.info('Block: desc(%s) must be a dict object(%s)',
desc,type(desc))
return None
# desc = self.valueExpr(desc)
try:
widget = self.w_build(desc)
self.dispatch('on_built',widget)
if hasattr(widget,'ready'):
widget.ready = True
return widget
except Exception as e:
print_exc()
self.dispatch('on_failed',e)
return None
if isinstance(desc,DictObject):
desc = desc.to_dict()
if not isinstance(desc, dict):
print('widgetBuild1: desc must be a dict object',
desc,type(desc))
self.dispatch('on_failed',Exception('miss url'))
return
widgettype = desc.get('widgettype')
while widgettype == "urlwidget":
opts = desc.get('options',{}).copy()
extend = desc.get('extend')
addon = None
if desc.get('extend'):
addon = desc.get('extend').copy()
url = opts.get('url')
if url is None:
self.dispatch('on_failed',Exception('miss url'))
return
if opts.get('url'):
del opts['url']
rtdesc = self.build_rtdesc(opts)
if rtdesc:
rtdata = self.get_rtdata(None, rtdesc)
params = opts.get('params', {})
params.update(rtdata)
opts['params'] = params
desc = self.getUrlData(url,**opts)
if not (isinstance(desc, DictObject) or \
isinstance(desc, dict)):
print('Block2: desc must be a dict object',
desc,type(desc))
self.dispatch('on_failed',Exception('miss url'))
return
if addon:
desc = dictExtend(desc,addon)
widgettype = desc.get('widgettype')
if widgettype is None:
print('Block3: desc must be a dict object not None')
return None
return doit(desc)
@classmethod
def getWidgetById(self, id:str, from_widget:Widget=None) -> Widget:
def find_widget_by_id(id, from_widget):
if id=='self':
return from_widget
if hasattr(from_widget,'widget_id'):
if from_widget.widget_id == id:
return from_widget
if id[0] == '-' and from_widget.widget_id == id[1:]:
return from_widget
if hasattr(from_widget, id):
w = getattr(from_widget,id)
return w
if id[0] == '-':
print('find_widget_by_id(), id=', id,
'widget=', from_widget)
if isinstance(from_widget, WindowBase):
return None
if from_widget.parent:
return find_widget_by_id(id,
from_widget=from_widget.parent)
return None
else:
children = [i for i in from_widget.children]
if hasattr(from_widget, 'get_subwidgets'):
children = from_widget.get_subwidgets()
for c in children:
ret = find_widget_by_id(id,from_widget=c)
if ret:
return ret
return None
ids = id.split('.')
app = App.get_running_app()
fid = ids[0]
if fid == '/self' or fid == 'root':
from_widget = app.root
if len(ids) == 1:
return from_widget
ids = ids[1:]
if fid == 'Window':
from_widget == Window
if len(ids) == 1:
return from_widget
ids = ids[1:]
if fid == 'app':
return app
if from_widget is None:
from_widget = app.root
for id in ids:
w = find_widget_by_id(id,from_widget=from_widget)
if w is None \
and id == ids[0] \
and Window.fullscreen == True \
and app.root != app.fs_widget:
w = find_widget_by_id(id, app.fs_widget)
if w is None:
print(id, 'not found ....')
return None
from_widget = w
return w
def on_built(self,v=None):
return
def on_failed(self,e=None):
return
Factory.register('Blocks',Blocks)
Factory.register('Video',Video)
Factory.register('OrientationLayout', OrientationLayout)

619
kivyblocks/dg.py 100755
View File

@ -0,0 +1,619 @@
import time
import ujson as json
from functools import partial
from kivy.logger import Logger
from kivy.uix.gridlayout import GridLayout
from kivy.uix.boxlayout import BoxLayout
from kivy.uix.scrollview import ScrollView
from kivy.uix.label import Label
from kivy.graphics import Fbo
from kivy.uix.button import ButtonBehavior
from kivy.clock import Clock
from kivy.properties import BooleanProperty, StringProperty, \
OptionProperty, DictProperty, NumericProperty, ListProperty
from kivy.properties import ListProperty
from kivy.graphics import Color, Rectangle
from kivy.app import App
from kivy.factory import Factory
from appPublic.dictObject import DictObject
from appPublic.uniqueID import getID
from appPublic.myTE import string_template_render
from .utils import *
from .baseWidget import Text, HBox, VBox
from .scrollpanel import ScrollPanel
from .paging import Paging, RelatedLoader
from .ready import WidgetReady
from .toolbar import Toolbar
from .bgcolorbehavior import BGColorBehavior
from .widget_css import WidgetCSS
from .uitype.factory import UiFactory
class BLabel(ButtonBehavior, Text):
def __init__(self, **kw):
ButtonBehavior.__init__(self)
Text.__init__(self,**kw)
self.csscls = 'dummy'
class Cell(ButtonBehavior, WidgetCSS, BoxLayout):
colume_name = StringProperty(None)
cell_type = OptionProperty('data', \
options=['data', 'header'])
def __init__(self,row,desc, **kw):
"""
desc:{
width,
datatype:fff
value:
format:
on_press:
}
"""
self.desc = desc
self.row = row
super().__init__(size_hint=(None,None),
width = self.desc['width'],
height = self.row.part.datagrid.rowHeight()
)
self.csscls=self.row.part.datagrid.row_normal_css
if self.row.header:
self.csscls=self.row.part.datagrid.header_css
if desc['header']:
bl = Text(i18n=True, text=str(desc['value']),
font_size=CSize(1),wrap=True,
halign='left', valign='middle'
)
self.cell_type = 'header'
else:
self.cell_type = 'data'
bl = UiFactory.build_view_widget(desc,self.row.row_data)
self.colume_name = desc['name']
if bl:
self.add_widget(bl)
class Row(BoxLayout):
def __init__(self,part, rowdesc,header=False,data=None, **kw):
"""
rowdesc=[
{
width,
name
value
on_press
}
"""
self.part = part
self.header = header
self.row_data = data
self.row_id = None
self.linewidth = self.part.datagrid.linewidth
self.rowdesc = rowdesc
opts = kw.copy()
opts.update({
"spacing":self.linewidth,
"orientation":"horizontal"
})
super(Row, self).__init__(**opts)
self.height = self.part.datagrid.rowHeight()
self.init(0)
def on_row_press(self, *args):
pass
def init(self,t):
w = 0
h = 0
for c in self.rowdesc:
c['header'] = self.header
cell = Cell(self,c)
self.add_widget(cell)
cell.bind(on_press=self.part.datagrid.cell_pressed)
w += cell.width
self.size_hint = None,None
self.width = w + self.linewidth * (len(self.rowdesc)+1)
def unselected(self):
self.select(False)
def selected(self):
self.select(True)
def select(self, flag):
for c in self.children:
if flag:
c.csscls = self.part.datagrid.row_selected_css
else:
c.csscls = self.part.datagrid.row_normal_css
class Header(WidgetReady, ScrollPanel):
def __init__(self,part,**kw):
SUPER(Header, self, kw)
self.part = part
self.init(1)
self.bind(on_scroll_stop=self.part.datagrid.scrollstop)
if self.part.freeze_flag:
self.bar_width = 0
self.bar_width = 0
def init(self,t):
rd = [ f.copy() for f in self.part.rowdesc ]
[ f.update({'value':self.part.fields[i].get('label', \
self.part.fields[i].get('name'))}) \
for i,f in enumerate(rd) ]
self.header = Row(self.part,rd,header=True)
self.add_widget(self.header)
self.height = self.header.height
class Body(WidgetReady, ScrollPanel):
def __init__(self,part,**kw):
self.part = part
SUPER(Body, self, kw)
self._inner.spacing = self.part.datagrid.linewidth
self.size_hint=(1,1)
self.idRow = {}
self.bind(on_scroll_stop=self.part.datagrid.scrollstop)
if self.part.freeze_flag:
self.bar_width = 0
def addRow(self,id, data,index=0):
rd = [ f.copy() for f in self.part.rowdesc ]
[ f.update({'value':data.get(f['name'])}) for f in rd ]
row = Row(self.part,rd,data=data)
row.row_id = id
self.add_widget(row,index=index)
self.idRow[id] = row
return row
def clearRows(self):
self.idRow = {}
self.clear_widgets()
def delRowById(self,id):
row = self.idRow.get(id)
if row:
self.remove_widget(row)
if self.idRow.get(id):
del self.idRow[id]
def getRowData(self,rowid):
return self.idRow[rowid].row_data
def getRowHeight(self):
return self.part.datagrid.rowHeight()
def get_row_by_id(self, rowid):
return self.idRow.get(rowid)
class DataGridPart(WidgetReady, BoxLayout):
def __init__(self,dg, freeze_flag, fields):
self.datagrid = dg
self.fields = fields
self.freeze_flag = freeze_flag
self.fields_width = None
BoxLayout.__init__(self, orientation='vertical')
WidgetReady.__init__(self)
self.init(0)
def setWidth(self):
if self.freeze_flag:
self.size_hint_x = None
self.width = self.getFieldsWidth()
def getFieldsWidth(self):
if not self.fields_width:
width = 0
for f in self.rowdesc:
width += f['width']
self.fields_width = width
return self.fields_width
def init(self,t):
rd = []
for f in self.fields:
r = f.copy()
r['width'] = CSize(f.get('width',10))
rd.append(r)
self.rowdesc = rd
self.setWidth()
kw = {}
if self.freeze_flag:
kw['width'] = self.fields_width
kw['size_hint'] = (None,None)
else:
kw['size_hint'] = (1,None)
kw['height'] = self.datagrid.rowHeight()
self.header = None
if not self.datagrid.noheader:
self.header = Header(self,**kw)
self.add_widget(self.header)
inner = {
"widgettype":"VBox",
"options":{
}
}
self.body = Body(self)
self.add_widget(self.body)
if not self.freeze_flag:
self.body.bind(pos=self.datagrid.bodyOnSize,
size=self.datagrid.bodyOnSize)
def clearRows(self):
return self.body.clearRows()
def addRow(self,id, data):
return self.body.addRow(id, data)
def on_size(self, o, s=None):
if not hasattr(self, 'body'):
return
if hasattr(self, '_toolbar'):
if self._toolbar is not None:
self._toolbar.width = self.width
self.body.size_hint_y = None
if self.header:
self.body.height = self.height - self.header.height
else:
self.body.height = self.height
class DataGrid(VBox):
"""
DataGrid data format:
{
"widgettype":"DataGrid",
"dataloader":{
"page":1,
"rows":60,
"dataurl":"eeee",
"params":{
}
},
"tailer":{
"options":{
}
"info":[
"total_cnt",
"total_page",
"page_rows",
"curpage"
],
"others":{
}
},
"row_height": CSize,
"header_css":"default",
"body_css":"default",
"spacing":1,
"fields":[
{
"name":"field1",
"label":"Field1",
"datatype":"str",
"uitype":"code",
"value_field":,
"text_field":
},
{
"name":"field2",
"label":"Field2",
"viewer":{
block dic
}
}
...
]
"binds":[
]
}
"""
row_selected = BooleanProperty(False)
row_normal_css = StringProperty('default')
row_selected_css = StringProperty('default')
header_css = StringProperty('default')
body_css = StringProperty('default')
row_height = NumericProperty(2)
noheader = BooleanProperty(False)
linewidth = NumericProperty(1)
toolbar = DictProperty(None)
dataloader = DictProperty(None)
fields = ListProperty(None)
tailer = DictProperty(None)
def __init__(self,**options):
self.select_rowid = None
self.rowheight = None
self.on_sizeTask = None
self.selected_rowid = None
self.show_rows = 0
self._toolbar = None
self.freeze_part = None
self.normal_part = None
SUPER(DataGrid, self, options)
self.freeze_fields = self.getPartFields(freeze_flag=True)
self.normal_fields = self.getPartFields(freeze_flag=False)
if not self.dataloader:
raise Exception('DataGrid need a DataLoader')
self._dataloader = RelatedLoader(target=self, **self.dataloader)
self._dataloader.bind(on_deletepage=self.delete_page)
self._dataloader.bind(on_pageloaded=self.add_page)
self._dataloader.bind(on_pageloaded=self.update_tailer_info)
self._dataloader.bind(on_newbegin=self.clearRows)
self.register_event_type('on_selected')
self.register_event_type('on_rowpress')
self.register_event_type('on_cellpress')
self.register_event_type('on_headerpress')
self.createDataGridPart()
self.createToolbar()
if self._toolbar:
self.add_widget(self._toolbar)
b = BoxLayout(orientation='horizontal')
if self.freeze_part:
b.add_widget(self.freeze_part)
if self.normal_part:
b.add_widget(self.normal_part)
self.add_widget(b)
if self.tailer:
self.tailer_widgets = {}
self.build_tailer(self.tailer)
def on_rowpress(self, *args):
print('on_rowpress fire, args=', args)
def on_cellpress(self, *args):
print('on_cesspress fire, args=', args)
def on_headerpress(self, *args):
print('on_headerpress fire, args=', args)
def cell_pressed(self, o):
if o.cell_type == 'header':
self.dispatch('on_headerpress', o.colume_name)
return
row = o.row
if self.selected_rowid:
self.unselect_row(self.selected_rowid)
self.selected_rowid = row.row_id
self.select_row(row.row_id)
self.dispatch('on_cellpress', o)
self.dispatch('on_rowpress', row)
self.dispatch('on_selected', row)
def unselect_row(self, row_id):
if self.freeze_part:
row = self.freeze_part.body.get_row_by_id(row_id)
if not row:
return
row.unselected()
row = self.normal_part.body.get_row_by_id(row_id)
if not row:
return
row.unselected()
def select_row(self, row_id):
if self.freeze_part:
row = self.freeze_part.body.get_row_by_id(row_id)
row.selected()
row = self.normal_part.body.get_row_by_id(row_id)
row.selected()
def on_ready(self, *args):
self.loadData()
def build_tailer(self, tailer_desc):
kw = tailer_desc.get('options', {})
kw.update({
'size_hint_y':None,
'height':self.rowheight
})
w = HBox(**kw)
self.add_widget(w)
self.show_infos(w, tailer_desc.get('info'))
if tailer_desc.get('others'):
w1 = self.build_tailer_other(tailer_desc.get('others'))
if w1:
w.add_widget(w1)
def update_tailer_info(self, *args):
if not hasattr(self, 'tailer_widgets'):
return
for n,w in self.tailer_widgets.items():
w.text = self.loader_info(n)
def show_infos(self, tailer_widget, info_names):
for n in info_names:
desc = {
"widgettype":"Text",
"options":{
"text":n,
"i18n":True,
}
}
w = Factory.Blocks().widgetBuild(desc)
tailer_widget.add_widget(w)
tailer_widget.add_widget(Label(text=':'))
self.tailer_widgets[n] = Label(text=self.loader_info(n))
tailer_widget.add_widget(self.tailer_widgets[n])
def build_tailer_others(desc):
return Factory.Blocks().widgetBuild(desc)
def loader_info(self, n):
if hasattr(self._dataloader, n):
txt=getattr(self._dataloader, n, 0)
if txt is None:
txt = '0'
txt = str(txt)
return txt
def locater(self,pos):
self.normal_part.body.scroll_y = pos
if self.freeze_part:
self.freeze_part.body.scroll_y = pos
def scrollstop(self,o,v=None):
if not self.noheader and o == self.normal_part.header:
self.normal_part.body.scroll_x = o.scroll_x
return
if o == self.normal_part.body:
if not self.noheader:
self.normal_part.header.scroll_x = o.scroll_x
if self.freeze_part:
self.freeze_part.body.scroll_y = o.scroll_y
if self.freeze_part and o == self.freeze_part.body:
self.normal_part.body.scroll_y = o.scroll_y
if o.scroll_y <= 0.01:
self._dataloader.loadNextPage()
if o.scroll_y >= 0.99:
self._dataloader.loadPreviousPage()
def getValue(self):
if not self.select_rowid:
return None
return self._getRowData(self.select_rowid)
def _getRowData(self, rowid):
d = {}
if self.freeze_part:
d.update(self.freeze_part.body.getRowData(rowid))
d.update(self.normal_part.body.getRowData(rowid))
return DictObject(**d)
def bodyOnSize(self,o,s):
if self.on_sizeTask is not None:
self.on_sizeTask.cancel()
self.on_sizeTask = Clock.schedule_once(self.calculateShowRows,0.3)
def rowHeight(self):
if not self.rowheight:
self.rowheight = CSize(self.row_height)
return self.rowheight
def calculateShowRows(self,t):
self.getShowRows()
self._dataloader.setPageRows(self.show_rows * 2)
def getShowRows(self):
if self.show_rows == 0:
return 60
self.show_rows = int(self.rowHeight() / self.normal_part.body.height)
return self.show_rows * 2
def clearRows(self, *args):
if self.freeze_part:
self.freeze_part.body.clearRows()
self.normal_part.body.clearRows()
def add_page(self,o,data):
dir = data['dir']
if not self.show_rows:
self.getShowRows()
ids = []
recs = data['data']
idx = 0
if dir == 'up':
recs.reverse()
idx = -1
recs1 = recs[:self.show_rows]
recs2 = recs[self.show_rows:]
self._fbo = Fbo(size=self.size)
with self._fbo:
self._background_color = Color(0,0,0,1)
self._background_rect = Rectangle(size=self.size)
for r in recs1:
id = self.addRow(r,index=idx)
ids.append(id)
with self.canvas:
self._fbo_rect = Rectangle(size=self.size,
texture=self._fbo.texture)
data['idx'] = idx
data['ids'] = ids
data['data'] = recs2
f = partial(self.add_page_delay,data)
Clock.schedule_once(f, 0)
def add_page_delay(self, data, *args):
recs = data['data']
page = data['page']
idx = data['idx']
ids = data['ids']
self._fbo = Fbo(size=self.size)
with self._fbo:
self._background_color = Color(0,0,0,1)
self._background_rect = Rectangle(size=self.size)
for r in recs:
id = self.addRow(r,index=idx)
ids.append(id)
with self.canvas:
self._fbo_rect = Rectangle(size=self.size,
texture=self._fbo.texture)
self._dataloader.bufferObjects(page,ids)
x = self._dataloader.getLocater()
self.locater(x)
def delete_page(self,o,data):
for id in data:
self.delRow(id)
def addRow(self,data, **kw):
id = getID()
f_row = None
if self.freeze_part:
self.freeze_part.body.addRow(id, data, **kw)
self.normal_part.body.addRow(id, data, **kw)
return id
def delRow(self,id,**kw):
if self.freeze_part:
self.freeze_part.body.delRowById(id)
self.normal_part.body.delRowById(id)
def createToolbar(self):
if self.toolbar:
self._toolbar = Toolbar(**self.toolbar)
def on_selected(self,row):
print("DataGrid():on_selected fire")
def loadData(self,*args, **kwargs):
kwargs['page'] = 1
self.selected_rowid = None
self._dataloader.do_search(None,kwargs)
def createDataGridPart(self):
self.freeze_part = None
self.normal_part = None
if self.freeze_fields:
self.freeze_part = DataGridPart(self,True, self.freeze_fields)
if self.normal_fields:
self.normal_part = DataGridPart(self, False, self.normal_fields)
def getPartFields(self,freeze_flag:bool=False) -> list:
fs = []
for f in self.fields:
if freeze_flag:
if f.get('freeze',False):
fs.append(f)
else:
if not f.get('freeze',False):
fs.append(f)
return fs
def get_selected_data(self):
if not self.selected_rowid:
return {}
data = {}
row_id = self.selected_rowid
if self.freeze_part:
row = self.freeze_part.body.get_row_by_id(row_id)
data.update(row.row_data)
row = self.normal_part.body.get_row_by_id(row_id)
data.update(row.row_data)
return data