Nov 15-17, 2017 Vikrant Patil
These notes are available online at http://notes.pipal.in/2017/arcesium-oct-advpython/day3.html
© Pipal Academy LLP
Syntactically classes in python look like this
class NameOfClass:
<statement-1>
<statement-2>
.
.import math
class Circle:
def __init__(self,radius):
self.radius = radius
def area(self):
return math.pi * self.radius**2
class Square:
def __init__(self, s):
self.side = s
def area(self):
return self.side**2
class Triangle:
def __init__(self, base, height):
self.base = base
self.height = height
shapes = [Circle(1), Square(1), Circle(2), Square(2)]
areas = [s.area() for s in shapes]
areas
shapes.append(Triangle(1,1))
areas = [s.area() for s in shapes]
filhandle = open("data.csv")
def grep(filehandle, pattern):
return [line for line in filehandle.readlines() if pattern in line]
grep(filhandle, "p")
class FakeReadlines:
def readlines(self):
return []
f = FakeReadlines()
grep(f, "p")
Q: is it possible to have nested modules?
!mkdir nested
!touch nested/__init__.py
!mkdir nested/inner
!touch nested/inner/__init__.py
!touch nested/inner/hello.py
import nested
type(nested)
from nested.inner import hello
!echo 'print("Hello world!")' > nested/inner/__main__.py
!python nested/inner/
class ClassA:
value = 42
def f(self):
return "Hello from classA"
ClassA
ClassA.value
ClassA.f
x = ClassA()
x
x.value
x.f
x.y = 10
x.y
def f2():
pass
x.method2 = f2
x.method2
x.f
method = x.f
method
method()
ClassA.f()
ClassA.f(x)
x.value
y = ClassA()
y.value
ClassA.value = 43
x.value
y.value
class A:
z = 0
def __init__(self, x, y):
self.x = x
self.y = y
a1 = A(2, 4)
a2 = A(4,5)
a1.z
a1.x
a1.y
a2.z
A.z = -1
a1.z
A.x = 3
a1.x
def outside_increment(self): #this is function
self.value +=1
class A:
def __init__(self):
self.value = 0
def increment(self): #this is method
self.value +=1
a = A()
a.increment()
a.value
outside_increment(a)
a.outside_increment = outside_increment
a.outside_increment()
dir(a)
a.value
class B:
name = ""
email = ""
b1 = B()
b2 = B()
b1.name = "alice"
b2.email = "hello@alice.in"
b2.name
B.name = "hello"
b1.name
class C:
name = ""
email = ""
z = 3
def __init__(self, name, email):
self.name = name
self.email = email
c1 = C("alice", "hello@alice.in")
c2 = C("xzy", "hello@xyz.com")
C.name = "abc"
c1.name
c1.z
C.z = 5
c1.z
c2.z
class Pair:
def __init__(self, x, y):
self.x = x
self.y = y
def __repr__(self):
return 'Pair({0.x!r},{0.y!r})'.format(self)
def __str__(self):
return '({0.x!s},{0.y!s})'.format(self)
p = Pair(2,3)
p
print(p)
"{0} {1}".format("hello", "python")
p1 =Pair(Pair(1,2), Pair(2,3))
p1
print(p1)
class Pair:
def __init__(self, x, y):
self.x = x
self.y = y
def __repr__(self):
return 'Pair({0.x!r},{0.y!r})'.format(self)
def __str__(self):
return '({0.x!s},{0.y!s})'.format(self)
def __add__(self, p):
return Pair(self.x+p.x, self.y + p.y)
def __sub__(self, p):
pass
def __rmul__(self, c):
pass
def __mul__(self, c):
pass
def __eq__(self, p):
pass
p = Pair(1,2)
p2 = Pair(3,4)
p + p2
2*"wew"
"**"*5
class Pair:
def __init__(self, x, y):
self.x = x
self.y = y
def __repr__(self):
return 'Pair({0.x!r},{0.y!r})'.format(self)
def __getitem__(self, name):
return self.__dict__[name]
def __setitem__(self, name , value):
if name in ["x", "y"]:
self.__dict__[name] = value
else:
raise Exception("Unsupported name!")
p = Pair(1,2)
p['x']
p['x'] = 3
p
p['z'] = 4
a1 + a2
%%file five.txt
one
two
three
four
five
class Formatter:
def format_text(self, text):
"""Formats given text.
This implememtation returns the same text,
but sub classes can override this method to provide
different way of formatting.
"""
return text
def format_file(self, filename):
text = open(filename).read()
return self.format_text(text)
class UpperCaseFormatter(Formatter):
def format_text(self, text):
return text.upper()
f = UpperCaseFormatter()
print(f.format_text("Hello"))
print(f.format_file("five.txt"))
class LineForematter(Formatter):
def format_line(self, line):
return line
def format_text(self, text):
lines = text.split("\n")
lines = [self.format_line(l) for l in lines]
return "\n".join(lines)
class PrefixFormatter(LineForematter):
def __init__(self, prefix):
self.prefix = prefix
def format_line(self, line):
return self.prefix + line
f = PrefixFormatter(prefix="[INFO]")
print(f.format_text("Hello\nWorld"))
print(f.format_file("five.txt"))
Q: is it possible to make class immutable?
class UpperCase:
def __getattr__(self, name):
return name.upper()
u = UpperCase()
u.hello
class Sealed:
def __setattr__(self, name, value):
raise Exception("No chance!")
s = Sealed()
s.x = 0
class Date:
__slots__ = ['year','month','day']
def __init__(self, year, month, day):
self.year = year
self.month = month
self.day = day
def method(self, x):
self.x = x
d = Date(2017, 11, 17)
d.method(60)
problem: Write a class Timer to measure time takes by task. The class should have start and stop methods and should be able to find time taken between call of start() and stop(). use: time.time()
t = Timer()
t.start()
do_some_stuff()
t.stop()
print("Time taken:", t.get_time_taken())import time
class Timer:
def __init__(self):
self._start = 0
self._end = 0
def start(self):
self._start = time.time()
def stop(self):
self._end = time.time()
def get_time_taken(self):
return self._end - self._start
t = Timer()
t.start()
time.sleep(5)
t.stop()
print("Time taken:", t.get_time_taken())
with open("five.txt") as f:
print(f.read())
with open("five.txt", "w") as f:
f.write("six\n")
f.write("seven\n")
from socket import socket, AF_INET, SOCK_STREAM
class LazyConnection:
def __init__(self, address, family=AF_INET, type_=SOCK_STREAM):
self.address = address
self.family = family
self.type_ = type_
self.sock = None
def __enter__(self):
print("__enter__ of LazyConnection")
if self.sock is not None:
raise RuntimeError("Already connected")
self.sock = socket(self.family, self.type_)
self.sock.connect(self.address)
return self.sock
def __exit__(self, exception_type, exeception_value, tracebak):
print("__exit__ of LazyConnection")
self.sock.close()
self.sock = None
from functools import partial
def fprint(s):
print("*"*5, s)
conn = LazyConnection(("www.python.org",80))
fprint("Before with")
with conn as s:
#conn.__enter__() executes
fprint("Inside with start")
s.send(b'GET /index.html HTTP/1.0\r\n')
s.send(b'Host: www.python.org\r\n')
s.send(b'\r\n')
resp = b''.join(iter(partial(s.recv, 8192), b''))
# the callable is called untill it returns b''
fprint("Inside with ends")
print(resp.decode("utf-8"))
problem: Write class ContextTimer which extends from Timer class. This new class implements context management protocol. where when you enter in with block your timer should start. and when you exit from with block your timer should stop and print time taken
with ContextTimer() as ct:
for i in range(10000):
for j in range(1000):
s = i*j*1.0
Time taken in with block: 2.121..class ContextTimer(Timer):
def __enter__(self):
self.start()
def __exit__(self, ect, ec, tb):
self.stop()
print("Time taken in with block:", self.get_time_taken())
with ContextTimer() as ct:
time.sleep(3)
class Person(object):
def __init__(self, firstname, lastname):
self.firstname = firstname
self.lastname = lastname
@property
def fullname(self):
print("Calling function fullname")
return " ".join([self.firstname, self.lastname])
p = Person("Alice", "Whoever")
p.fullname
p.firstname
p.lastname
p.fullname
class Person(object):
def __init__(self, name, email):
self._name = name
self._email = email
#getter
@property
def name(self):
## you might do some additonal processing or checks
return self._name
#setter
@name.setter
def name(self, value):
if not isinstance(value, str):
raise TypeError("Expected a string for name")
self._name = value
#deleter
@name.deleter
def name(self):
raise AttributeError("Can't delete name attribute")
p = Person("Alice", "alice@example.com")
p.name
p.name = "alice"
p.name
p.name = 42
del p.name
class Zero:
def __get__(self, obj, cls):
if obj is None:
return self
print("Zero.__get__")
return 0
def __repr__(self):
print("<Descriptor Zero>")
class Foo:
x = Zero()
f = Foo()
f.x
three = 3
type(three)
f.x + 3
class Integer:
def __init__(self, name):
self.name = name
def __get__(self, instance, cls):
print("__get__ from", self)
if instance is None:
return self
else:
return instance.__dict__[self.name]
def __set__(self, instance, value):
print("__set__ from", self)
if not isinstance(value, int):
raise TypeError("Expected an int")
instance.__dict__[self.name] = value
def __delete__(self, instance):
print("__delete__ from,", self)
del instance.__dict__[self.name]
def __str__(self):
return "Integer<{0.name!s}>".format(self)
class Point:
x = Integer('x')
y = Integer('y')
def __init__(self, x, y):
self.x = x
self.x = y
p = Point(2,3)
Point.__dict__
p.x
Point.x
p.y = 2
p.y = "3"
problem: Implement a my_property decorator that works like built in property
class my_property:
def __init__(self, func):
self.func = func
def __get__(self, instance, cls):
if instance is None:
return self
print("my_property __get__")
return self.func(instance)
class Person:
def __init__(self, name, email):
self._name = name
self._email = email
@my_property
def name(self):
return self._name
p = Person("name", "email@xyz.com")
p.name
my_property(func)
class A:
def __init__(self, x):
self.x = x
def __repr__(self):
return "{0}({1})".format(self.__class__.__name__, repr(self.x))
@staticmethod
def parse1(value):
return A(int(value))
@classmethod
def parse2(cls, value):
return cls(int(value))
class B(A):
pass
a = A.parse1("123")
a
b = B.parse1("123")
b
type(b)
b1 = B.parse2("123")
b1
class Integer:
def __init__(self, name):
self.name = name
def __get__(self, instance, cls):
print(instance, cls)
print("__get__ from", self)
if instance is None:
return self
else:
return instance.__dict__[self.name]
def __set__(self, instance, value):
print("__set__ from", self)
if not isinstance(value, int):
raise TypeError("Expected an int")
instance.__dict__[self.name] = value
def __delete__(self, instance):
print("__delete__ from,", self)
del instance.__dict__[self.name]
def __str__(self):
return "Integer<{0.name!s}>".format(self)
class Point:
x = Integer('x')
y = Integer('y')
def __init__(self, x, y):
self.x = x
self.x = y
Point.x
p = Point(2,3)
p.x
def debug(func):
prefix = "*"*5
msg = prefix + func.__qualname__
def wrapper(*args, **kwargs):
print(msg)
return func(*args, **kwargs)
return wrapper
def debugmethods(cls):
for name, val in vars(cls).items():
if callable(val):
setattr(cls, name, debug(val))
return cls
@debugmethods
class Foo:
x = "dummy"
def method1(self):
pass
def method2(self):
pass
def method3(self):
pass
f = Foo()
f.x
f.method1()
%%file t1.py
import threading
def task():
print("Hello ", threading.currentThread().getName())
def main():
t1 = threading.Thread(target=task)
t1.start()
t1.join()
if __name__ == "__main__":
main()
!python t1.py
%%file t2.py
import threading
def task():
print("Hello ", threading.currentThread().getName())
def main():
threads = [threading.Thread(target=task) for i in range(10)]
for t in threads:
t.start()
for t in threads:
t.join()
if __name__ == "__main__":
main()
!python t2.py
%%file counter.py
import threading
class Counter:
def __init__(self):
self.count = 0
def tick(self):
self.count +=1
def task(counter, n):
for i in range(n):
counter.tick()
def main():
counter = Counter()
n = 100000
nthreads = 10
threads = [threading.Thread(target=task, args=(counter, n)) for i in range(nthreads)]
for t in threads:
t.start()
for t in threads:
t.join()
print(counter.count)
if __name__ == "__main__":
main()
!time python counter.py
!python counter.py
%%file counter1.py
import threading
class Counter:
def __init__(self):
self.count = 0
self.lock = threading.Lock()
def tick(self):
with self.lock:
self.count +=1
def task(counter, n):
for i in range(n):
counter.tick()
def main():
counter = Counter()
n = 100000
nthreads = 10
threads = [threading.Thread(target=task, args=(counter, n)) for i in range(nthreads)]
for t in threads:
t.start()
for t in threads:
t.join()
print(counter.count)
if __name__ == "__main__":
main()
!time -p python counter.py
!time -p python counter1.py
problem: Download urls in parallel using threads.
!seq 10 | xargs printf "http://httpbin.org/get?x=%d\n"
!seq 10 | xargs printf "http://httpbin.org/get?x=%d\n" > urls.txt
%%file sget.py
import sys
from urllib.request import urlopen
def get_urls(filename):
return (line.strip() for line in open(filename))
def wget(url):
return urlopen(url).read()
def main():
filename = sys.argv[1]
for url in get_urls(filename):
wget(url)
if __name__ == "__main__":
main()
!time -p python sget.py urls.txt
%%file pget.py
import sys
from sget import get_urls, wget
from multiprocessing.pool import ThreadPool
def main():
filename = sys.argv[1]
concurrency = int(sys.argv[2])
urls = get_urls(filename)
pool = ThreadPool(concurrency)
pool.map(wget, urls)
if __name__ == "__main__":
main()
!time -p python pget.py urls.txt 1
!time -p python pget.py urls.txt 2
!time -p python pget.py urls.txt 4
Lets try same task with multiprocessing
%%file pcpu.py
import sys
from multiprocessing.pool import Pool
def totaltask():
task("")
task2("")
def task(dummy):
print("Started task")
sum = 0
for i in range(1000):
for j in range(10000):
sum += 1.0*i*j
print("Finished task")
return sum
def task2(x):
print("Started task2")
sum = 0
for i in range(1000):
for j in range(10000):
sum += 1.0*i*j
return sum
def main():
conc = int(sys.argv[1])
pool = Pool(conc)
pool.apply(totaltask, args=() )
if __name__ == "__main__":
main()
!time -p python pcpu.py 2
def coroutine(func):
def start(*args,**kwargs):
cr = func(*args,**kwargs)
next(cr)
return cr
return start
@coroutine
def grep(pattern):
print("Looking for %s" % pattern)
while True:
line = (yield)
if pattern in line:
print(line)
g = grep("python")
g.send("Hello python")
g.send("There is no pattern")
import ctypes
from ctypes.util import find_library
path = find_library("c")
print(path)
libc = ctypes.cdll.LoadLibrary(path)
libc.printf("Hello c world!")
libc.abs(-2)
libc.fabs
!mkdir cy
%%file cy/sum.pyx
def sum(values):
result = 0
for v in values:
result += v
return result
%%file cy/setup.py
from setuptools import setup
from Cython.Build import cythonize
setup(
ext_modules = cythonize("sum.pyx")
)
!cd cy && python setup.py build_ext --inplace
%%file cy/testcy.py
import sum
print(sum.sum(range(1000)))
!cd cy && python testcy.py
from numba import jit
from numpy import arange
# jit decorator tells Numba to compile this function.
# The argument types will be inferred by Numba when function is called.
@jit
def sum2d(arr):
M, N = arr.shape
result = 0.0
for i in range(M):
for j in range(N):
result += arr[i,j]
return result
def sum2d_(arr):
M, N = arr.shape
result = 0.0
for i in range(M):
for j in range(N):
result += arr[i,j]
return result
@jit()
def sum2d_i(arr):
M, N = arr.shape
result = 0.0
for i in range(M):
for j in range(N):
result += arr[i,j]
return result
a = arange(10000).reshape(100,100)
%timeit sum2d(a)
%timeit sum2d_(a)