You can not select more than 25 topics Topics must start with a letter or number, can include dashes ('-') and can be up to 35 characters long.

tasks_loop.py 8.6KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305306307308309310311312313314315316317318319320321322323324325326327328329330331332333334335336337338339340341342343344345346347348349350351352353354355356357358359360361362363364365366367368369370371372373374375376377378379380381382383384385386387388389390391392393394395396397398399400401402403404405406407
  1. import asyncio
  2. import time
  3. import redis
  4. import os
  5. import copy
  6. import json
  7. import traceback
  8. import inspect
  9. import atexit
  10. '''
  11. umod
  12. # '''
  13. p=print
  14. async def anop(*a,**kw):pass
  15. def nop(*a,**kw):pass
  16. class Edict(dict):pass
  17. class Eobj():pass
  18. def connect_redis():
  19. return redis.Redis(host='localhost', port=6379, db=0,charset="utf-8", decode_responses=True, encoding_errors="backslashreplace")
  20. def eclass_factory(n):
  21. ret = []
  22. for k in range(n):
  23. class nx:pass
  24. # _rcls_name
  25. ret.append(nx)
  26. return ret
  27. '''
  28. # '''
  29. def callable_helper(fn):
  30. ret = {
  31. "is_callable":0,
  32. "is_coroutine":0,
  33. }
  34. ret["is_callable"]= callable(fn)
  35. ret["is_coroutine"]= inspect.iscoroutinefunction(fn)
  36. # if (ca)
  37. return ret
  38. verbose = print
  39. class DbgMixin:
  40. async def ws_rec(self,text_data,*a,**kw):
  41. # if "dbg" in
  42. text_data_json = json.loads(text_data)
  43. if not 'dbg' in text_data_json:
  44. return
  45. p("DBG!")
  46. exec(text_data_json["dbg"])
  47. class AconMixin(DbgMixin):
  48. # class AconMixin:
  49. async def connect(self):
  50. # print("-----------CONNECTING async def connect")
  51. p("ACON!!!!!!!!!!!!!!!!!!!!!!!!!!!")
  52. p(flush=True)
  53. await self.accept()
  54. verbose("<MRO")
  55. verbose(type(self))
  56. await self.call_all_mro("ws_conn0")
  57. await self.call_all_mro("ws_conn")
  58. await self.call_all_mro("ws_conn2")
  59. verbose("/MRO>")
  60. # await self.ws_conn_once(event)
  61. await self.ws_conn_once()
  62. async def ws_conn_once(self, *a,**kw):pass
  63. async def ws_disconn_once(self, *a,**kw):pass
  64. async def ws_rec_once(self, *a,**kw):pass
  65. async def receive(self, *a,**kw):
  66. # print("REC")
  67. # print("-----------REC")
  68. await self.call_all_mro("ws_rec",*a,**kw)
  69. await self.ws_rec_once(*a,**kw)
  70. async def call_all_mro(self,mthd_name,*args,**kwargs):
  71. called = set([None])
  72. for cls_obj in self.__class__.__mro__:
  73. mthd = getattr(cls_obj,mthd_name,None)
  74. # print("~",cls_obj,mthd)
  75. if not mthd in called:
  76. # print(cls_obj,mthd)
  77. called.add(mthd)
  78. await mthd(self,*args,**kwargs)
  79. async def disconnect(self, close_code):
  80. await self.call_all_mro("ws_disconn",close_code)
  81. await self.ws_disconn_once(close_code)
  82. async def websocket_connect_x(self, event):
  83. await super().websocket_connect(event)
  84. print("<MRO")
  85. print(type(self))
  86. await self.call_all_mro("ws_conn",event)
  87. await self.call_all_mro("ws_conn2",event)
  88. print("/MRO>")
  89. await self.ws_conn_once(event)
  90. class AclTaskMixin:
  91. async def ws_conn0(self):
  92. glob_cl.run_persistent_loop(self)
  93. print("++++++++++++++++++++++++++")
  94. async def ws_conn_once(self,*a,**kw):
  95. print("\t\tws_conn_once:","AclTaskMixin")
  96. class DevMixin(*eclass_factory(10)):pass
  97. class DevMroMixin(*eclass_factory(10)):pass
  98. class WSStoMixin(*eclass_factory(10)):pass
  99. class GCLMixin(*eclass_factory(10)):pass
  100. class RldManMixin(*eclass_factory(10)):pass
  101. Mixins = Eobj()
  102. Mixins.DevMixin = DevMixin
  103. Mixins.DevMroMixin = DevMroMixin
  104. Mixins.WSStoMixin = WSStoMixin
  105. Mixins.GCLMixin = GCLMixin
  106. Mixins.RldManMixin = RldManMixin
  107. Mixins.DbgMixin = DbgMixin
  108. Mixins.AconMixin = AconMixin
  109. Mixins.AclTaskMixin = AclTaskMixin
  110. # GCLMixin
  111. class globCL():
  112. def __init__(self,*a,**kw):
  113. # for now we are not going to do any weird class stuff
  114. self.rconn = connect_redis()
  115. self.rman = RldMan()
  116. self.cbs_once = []
  117. self.cbs_once_set = set()
  118. self.cbs = {"fn1":fn1,"fn2x":nop}
  119. self.timeout = 1
  120. pass
  121. async def persistent_loop(self,*a,**kw):
  122. while 1:
  123. # p("PL:",time.time(),flush=1)
  124. # await asyncio.sleep(5)
  125. # for k,v in cls.cbs.items():
  126. for k,v in [*self.cbs.items()]:
  127. try:
  128. call_info = callable_helper(v)
  129. if (call_info["is_callable"]):
  130. if call_info["is_coroutine"]:
  131. pass
  132. await v({k,self})
  133. else:
  134. v({k,self})
  135. else:
  136. pass
  137. # p(k,v)
  138. pass
  139. except Exception as e:
  140. p("persistent_loop Exception",e)
  141. p(flush=True)
  142. # raise
  143. else:
  144. pass
  145. finally:
  146. pass
  147. # cbs_once
  148. cbs_once = [*self.cbs_once,*self.cbs_once_set]
  149. self.cbs_once = []
  150. self.cbs_once_set = set()
  151. k=0
  152. for v in cbs_once:
  153. try:
  154. call_info = callable_helper(v)
  155. if (call_info["is_callable"]):
  156. if call_info["is_coroutine"]:
  157. pass
  158. await v({k,self})
  159. else:
  160. v({k,self})
  161. else:
  162. pass
  163. # p(k,v)
  164. pass
  165. except Exception as e:
  166. p("persistent_loop Exception cbs_once:",e)
  167. p(flush=True)
  168. # raise
  169. else:
  170. pass
  171. finally:
  172. pass
  173. k += 1
  174. await asyncio.sleep(self.timeout)
  175. def _run_persistent_loop(self,*a,**kw):
  176. p("_run_persistent_loop:")
  177. loop = asyncio.get_event_loop()
  178. self.run_persistent_loop = nop
  179. self.rman.persistent_loop_cb()
  180. loop.create_task(self.persistent_loop())
  181. self.loop = loop
  182. run_persistent_loop = _run_persistent_loop
  183. # RldManMixin
  184. class RldMan():
  185. def __init__(self,*a,**kw):
  186. self.files = {}
  187. z="ABC"
  188. self.scopes = {
  189. "dflt_scope":{"globals":globals(),"locals":locals()},
  190. "dflt_scope_zloc":{"globals":globals(),"locals":{}},
  191. "dflt_scope_gscope":{"globals":globals(),"locals":{}},
  192. "dflt_scope_copy": {"globals":copy.copy(globals()),"locals":copy.copy(locals())},
  193. # "dflt_scope":{"globals":globals(),"locals":{}}
  194. }
  195. self.scopes['current_scope'] = self.scopes['dflt_scope_zloc']
  196. self.scope_opt = ""
  197. self.print_tb = 0
  198. # self.dflt_scope = {"globals":globals(),"locals":locals()}
  199. # self.dflt_scope2 = {"globals":copy.copy(globals()),"locals":copy.copy(locals())}
  200. base_path = "/usr/games/repos/ign4/py_rld/djc/rdir1/"
  201. # [base_path+"r0.py",{"ftxt":"zz2"}],
  202. file_list = [
  203. base_path+"i0.py",
  204. base_path+"i1.py",
  205. base_path+"i2.py",
  206. base_path+"i3.py",
  207. base_path+"i4.py",
  208. base_path+"i5.py",
  209. ]
  210. self.add_files(file_list)
  211. # self.add_files(file_list,{"run"})
  212. def persistent_loop_cb(self,*a,**kw):
  213. base_path = "/usr/games/repos/ign4/py_rld/djc/rdir1/"
  214. file_list = [
  215. base_path+"r0.py",
  216. base_path+"r1.py",
  217. base_path+"r2.py",
  218. base_path+"r3.py",
  219. base_path+"r4.py",
  220. base_path+"r5.py",
  221. ]
  222. self.add_files(file_list)
  223. def add_files(self,files):
  224. for file in files:
  225. if type(file)==str:
  226. self.add_file(file)
  227. elif type(file)==list:
  228. self.add_file(*file)
  229. else:
  230. p("add files???",file)
  231. def add_file(self,file_name,fnx={}):
  232. self.files[file_name] = {"ftxt":"",**fnx}
  233. def get_scope(self,rfile_obj,file_name):
  234. return self.scopes["current_scope"]
  235. return {
  236. "scope":self.scopes["current_scope"],
  237. }
  238. # rfile_obj.get("scope")
  239. # if
  240. def rld_file(self,file_name,ret= {"errs":{},"all":{},"alle":{},},**kw):
  241. if file_name in self.files:
  242. rfile_obj = self.files[file_name]
  243. st = os.stat(file_name)
  244. st_tuple = (st.st_mtime,st.st_size)
  245. # if rfile_obj["ftxt"] == "":
  246. eflag ="nd"
  247. # rfile_obj["ftxt"] = st_tuple
  248. if rfile_obj["ftxt"] != st_tuple:
  249. # p(rfile_obj["ftxt"])
  250. rfile_obj["ftxt"] = st_tuple
  251. try:
  252. f = open(file_name,"r")
  253. ftxt = f.read()
  254. # scope_key = rfile_obj.get("scope")
  255. # p(scope_key,rfile_obj)
  256. f.close()
  257. scope_obj = self.get_scope(file_name,rfile_obj)
  258. if self.scope_opt == "locals":
  259. eflag ="locals"
  260. exec(ftxt,scope_obj["globals"],scope_obj["locals"])
  261. elif self.scope_opt == "globals":
  262. eflag ="globals"
  263. exec(ftxt,scope_obj["globals"])
  264. else:
  265. eflag ="[]"
  266. exec(ftxt)
  267. except Exception as e:
  268. print("EXCEPT",eflag,e)
  269. if self.print_tb:
  270. traceback.print_tb(e.__traceback__,file=sys.stdout)
  271. else:
  272. pass
  273. # print("ELSE")
  274. finally:
  275. # print("FINALLY")
  276. pass
  277. print(end="",flush=True)
  278. return ret
  279. def rld_files(self):
  280. # p()
  281. ret = {
  282. "errs":{},
  283. "all":{},
  284. "alle":{},
  285. }
  286. # for k in self.files:
  287. for k in [*self.files]:
  288. self.rld_file(k,ret)
  289. # p("rld_files!")
  290. p("",end="",flush=True)
  291. return ret
  292. def fn1(*a,**kw):
  293. # rld_files
  294. glob_cl.rman.rld_files()
  295. glob_cl = globCL()
  296. glob_cl.rman.rld_files()
  297. # print("..",end="\n",flush=True)
  298. def redis_info():
  299. rconn = glob_cl.rconn
  300. keys = rconn.keys()
  301. for k in keys:
  302. p(":",k)
  303. p("len(keys)",len(keys))
  304. print("",end="",flush=True)
  305. # rconn.flushall()
  306. redis_info()
  307. r"""
  308. def atexit_fn():
  309. pass
  310. # rconn.flushall()
  311. print("ATEXIT FN")
  312. rinfo_len()
  313. keys = rconn.keys()
  314. print(keys)
  315. print(len(keys))
  316. print("t 000")
  317. # time.sleep(10)
  318. print("t 010")
  319. # time.sleep(20)
  320. print("t 020")
  321. print("ATEXIT FN...")
  322. rconn.flushall()
  323. atexit.register(atexit_fn)
  324. # """
  325. def atexit_fn():
  326. print("atexit_fn!\n\n")
  327. print("",end="",flush=True)
  328. atexit.register(atexit_fn)