Trials

Author: KB-perByte

Notebooks/MyCheatsheet.ipynb

@@ -33,7 +33,70 @@
    "outputs": [],
    "source": [
     "li = [1, 10 ,2 ,22 ,454 ,64 ,778 ,845 ,32]\n",
-    "la = [\"Ab\", \"a1\" , \"1A\", \"b!\", \"B1\"]"
+    "la = [\"Ab\", \"a1\" , \"1A\", \"b!\", \"B1\"]\n",
+    "lx = {1, 3, 5 ,2 ,1 , 8, 4}"
+   ]
+  },
+  {
+   "cell_type": "code",
+   "execution_count": 5,
+   "metadata": {},
+   "outputs": [
+    {
+     "output_type": "execute_result",
+     "data": {
+      "text/plain": [
+       "{1, 2, 3, 4, 5, 8}"
+      ]
+     },
+     "metadata": {},
+     "execution_count": 5
+    }
+   ],
+   "source": [
+    "lx"
+   ]
+  },
+  {
+   "cell_type": "code",
+   "execution_count": 2,
+   "metadata": {},
+   "outputs": [
+    {
+     "output_type": "stream",
+     "name": "stdout",
+     "text": [
+      "[1, 845]\n"
+     ]
+    }
+   ],
+   "source": [
+    "final_list = list(filter(lambda x: (x%2 != 0) , li)) \n",
+    "print(final_list) "
+   ]
+  },
+  {
+   "cell_type": "code",
+   "execution_count": 9,
+   "metadata": {},
+   "outputs": [
+    {
+     "output_type": "execute_result",
+     "data": {
+      "text/plain": [
+       "'187'"
+      ]
+     },
+     "metadata": {},
+     "execution_count": 9
+    }
+   ],
+   "source": [
+    "from collections import namedtuple\n",
+    "Person = namedtuple('Person', 'name height')\n",
+    "person = Person('Jean-Luc', 187)\n",
+    "f'{person.height}'\n",
+    "'{p.height}'.format(p=person)"
    ]
   },
   {

local/DataModel.py

@@ -0,0 +1,22 @@
+#Better say python data models than dunder methords
+
+class Polynomail:
+    def __init__(self, *coeffs):
+        self.coeffs = coeffs
+
+    def __repr__(self):
+        return 'Polynomial(*{!r})'.format(self.coeffs)
+
+    def __add__(self, other):
+        return Polynomail(*(x + y for x, y in zip (self.coeffs, other.coeffs)))
+
+    def __len__(self):
+        return len(self.coeffs)
+
+p1 =Polynomail(1,2,3)
+p2 =Polynomail(2,3,4)
+
+print(p1)
+
+k = p1 + p2
+print(k)

local/RH/pickling.py

@@ -0,0 +1,55 @@
+import pickle
+import tempfile
+
+#pickling and unpickling  should be done using binary file since they support bytestream
+#to store structured data in a file
+ 
+# pickle.dump(object, file) <- converts object to binary file
+
+def storePickle():
+    sagar = {'key': 'Sagar', 'name': 'Sagar Paul', 'age': 24, 'iq' : 10}
+    nandini = {'key': 'Nandini', 'name': 'Nandini Chaudhary', 'age': 24, 'iq' : 60}
+
+    dbase = {}
+    dbase['Sagar'] = sagar
+    dbase['Nandini'] = nandini
+
+    try:
+        dbFile = open('tryPickle', 'ab')
+        pickle.dump(dbase, dbFile)
+    finally:
+        dbFile.close()
+
+def storePickleTemp():
+    sagar = {'key': 'Sagar', 'name': 'Sagar Paul', 'age': 24, 'iq' : 10}
+    nandini = {'key': 'Nandini', 'name': 'Nandini Chaudhary', 'age': 24, 'iq' : 60}
+
+    temp = tempfile.TemporaryFile()
+   
+    dbase = {}
+    dbase['Sagar'] = sagar
+    dbase['Nandini'] = nandini
+
+    try:
+        temp.write(pickle.dumps(dbase))
+        temp.seek(0)
+        print(temp.read())
+    finally:
+        temp.close()
+
+
+def loadPickle():
+    dbfile = open('tryPickle', 'rb')
+
+    dbase = pickle.load(dbfile)
+    for k in dbase:
+        print(k, dbase.get(k))
+
+    dbfile.close()
+
+if __name__ == "__main__":
+    storePickle()
+    loadPickle()
+    
+
+

local/RH/unixSocket/client.py

@@ -0,0 +1,27 @@
+# -*- coding: utf-8 -*-
+import socket
+import os
+
+print("Connecting...")
+if os.path.exists("/tmp/python_unix_sockets_example"):
+    client = socket.socket(socket.AF_UNIX, socket.SOCK_DGRAM)
+    client.connect("/tmp/python_unix_sockets_example")
+    print("Ready.")
+    print("Ctrl-C to quit.")
+    print("Sending 'DONE' shuts down the server and quits.")
+    while True:
+        try:
+            x = input("> ")
+            if "" != x:
+                print("SEND:", x)
+                client.send(x.encode('utf-8'))
+                if "DONE" == x:
+                    print("Shutting down.")
+                    break
+        except KeyboardInterrupt as k:
+            print("Shutting down.")
+            client.close()
+            break
+else:
+    print("Couldn't Connect!")
+    print("Done")

local/RH/unixSocket/interSocketComm.py

@@ -0,0 +1,22 @@
+import socket
+import os
+
+parent, child = socket.socketpair()
+
+pid = os.fork()
+
+if pid:
+    print ('in parent, sending message')
+    child.close()
+    parent.sendall('ping')
+    response = parent.recv(1024)
+    print ('response from child:', response)
+    parent.close()
+
+else:
+    print ('in child, waiting for message')
+    parent.close()
+    message = child.recv(1024)
+    print ('message from parent:', message)
+    child.sendall('pong')
+    child.close()

local/RH/unixSocket/server.py

@@ -0,0 +1,22 @@
+# -*- coding: utf-8 -*-
+import socket
+import os
+
+print("Opening socket...")
+server = socket.socket(socket.AF_UNIX, socket.SOCK_DGRAM)
+server.bind("/tmp/python_unix_sockets_example")
+
+print("Listening...")
+while True:
+    datagram = server.recv(1024)
+    if not datagram:
+        break
+    else:
+        print("-" * 20)
+        print(datagram.decode('utf-8'))
+        if "DONE" == datagram.decode('utf-8'):
+            break
+print("-" * 20)
+print("Shutting down...")
+server.close()
+print("Done")

local/something.py

@@ -0,0 +1,49 @@
+import asyncio, collections, curses, enum, random
+
+P = collections.namedtuple('P', 'x y')         # Position
+D = enum.Enum('D', 'n e s w')                  # Direction
+
+def main(screen):
+    curses.curs_set(0)                         # Makes cursor invisible.
+    screen.nodelay(True)                       # Makes getch() non-blocking.
+    asyncio.run(main_coroutine(screen))        # Starts running asyncio code.
+
+async def main_coroutine(screen):
+    state = {'*': P(0, 0), **{id_: P(30, 10) for id_ in range(10)}}
+    moves = asyncio.Queue()
+    coros = (*(random_controller(id_, moves) for id_ in range(10)),
+             human_controller(screen, moves),
+             model(moves, state, *screen.getmaxyx()),
+             view(state, screen))
+    await asyncio.wait(coros, return_when=asyncio.FIRST_COMPLETED)
+
+async def random_controller(id_, moves):
+    while True:
+        moves.put_nowait((id_, random.choice(list(D))))
+        await asyncio.sleep(random.random() / 2)
+
+async def human_controller(screen, moves):
+    while True:
+        ch = screen.getch()
+        key_mappings = {259: D.n, 261: D.e, 258: D.s, 260: D.w}
+        if ch in key_mappings:
+            moves.put_nowait(('*', key_mappings[ch]))
+        await asyncio.sleep(0.01)  
+
+async def model(moves, state, height, width):
+    while state['*'] not in {p for id_, p in state.items() if id_ != '*'}:
+        id_, d = await moves.get()
+        p      = state[id_]
+        deltas = {D.n: P(0, -1), D.e: P(1, 0), D.s: P(0, 1), D.w: P(-1, 0)}
+        new_p  = P(*[sum(a) for a in zip(p, deltas[d])])
+        if 0 <= new_p.x < width-1 and 0 <= new_p.y < height:
+            state[id_] = new_p
+
+async def view(state, screen):
+    while True:
+        screen.clear()
+        for id_, p in state.items():
+            screen.addstr(p.y, p.x, str(id_))
+        await asyncio.sleep(0.01)  
+
+curses.wrapper(main)