Skip to content

Instantly share code, notes, and snippets.

@shihpeng
Last active August 29, 2015 14:07
Show Gist options
  • Save shihpeng/81dbc37a621579743cf3 to your computer and use it in GitHub Desktop.
Save shihpeng/81dbc37a621579743cf3 to your computer and use it in GitHub Desktop.
Frequently used Python code snippets
import os
import re
import subprocess
def available_cpu_count():
""" Number of available virtual or physical CPUs on this system, i.e.
user/real as output by time(1) when called with an optimally scaling
userspace-only program"""
# cpuset
# cpuset may restrict the number of *available* processors
try:
m = re.search(r'(?m)^Cpus_allowed:\s*(.*)$',
open('/proc/self/status').read())
if m:
res = bin(int(m.group(1).replace(',', ''), 16)).count('1')
if res > 0:
return res
except IOError:
pass
# Python 2.6+
try:
import multiprocessing
return multiprocessing.cpu_count()
except (ImportError, NotImplementedError):
pass
# http://code.google.com/p/psutil/
try:
import psutil
return psutil.NUM_CPUS
except (ImportError, AttributeError):
pass
# POSIX
try:
res = int(os.sysconf('SC_NPROCESSORS_ONLN'))
if res > 0:
return res
except (AttributeError, ValueError):
pass
# Windows
try:
res = int(os.environ['NUMBER_OF_PROCESSORS'])
if res > 0:
return res
except (KeyError, ValueError):
pass
# jython
try:
from java.lang import Runtime
runtime = Runtime.getRuntime()
res = runtime.availableProcessors()
if res > 0:
return res
except ImportError:
pass
# BSD
try:
sysctl = subprocess.Popen(['sysctl', '-n', 'hw.ncpu'],
stdout=subprocess.PIPE)
scStdout = sysctl.communicate()[0]
res = int(scStdout)
if res > 0:
return res
except (OSError, ValueError):
pass
# Linux
try:
res = open('/proc/cpuinfo').read().count('processor\t:')
if res > 0:
return res
except IOError:
pass
# Solaris
try:
pseudoDevices = os.listdir('/devices/pseudo/')
res = 0
for pd in pseudoDevices:
if re.match(r'^cpuid@[0-9]+$', pd):
res += 1
if res > 0:
return res
except OSError:
pass
# Other UNIXes (heuristic)
try:
try:
dmesg = open('/var/run/dmesg.boot').read()
except IOError:
dmesgProcess = subprocess.Popen(['dmesg'], stdout=subprocess.PIPE)
dmesg = dmesgProcess.communicate()[0]
res = 0
while '\ncpu' + str(res) + ':' in dmesg:
res += 1
if res > 0:
return res
except OSError:
pass
raise Exception('Can not determine number of CPUs on this system')
>>> import datetime
>>> # replace datetime.datetime.now() with your datetime object
>>> int(datetime.datetime.now().strftime("%s")) * 1000
1312908481000
>>> import datetime, time
>>> # replace datetime.datetime.now() with your datetime object
>>> time.mktime(datetime.datetime.now().timetuple()) * 1000
1312908681000.0
>>> dt = time.gmtime(timestamp_millis/1000)
>>> dt
import os
def get_size(start_path = '.'):
total_size = 0
for dirpath, dirnames, filenames in os.walk(start_path):
for f in filenames:
fp = os.path.join(dirpath, f)
total_size += os.path.getsize(fp)
return total_size
print get_size()
#
# Use __import__() and sys.modules
#
import sys
name = 'foo.bar.baz'
__import__(name)
#<module 'foo' from ...>
baz = sys.modules[name]
#<module 'foo.bar.baz' from ...>
#
# Use __import__()
#
__import__( name[, globals[, locals[, fromlist[, level]]]])
# method 1.
mod = __import__('foo.bar.baz', globals(), locals(), ['object'], -1)
# method 2.
def my_import(name):
mod = __import__(name)
components = name.split('.')
for comp in components[1:]:
mod = getattr(mod, comp)
return mod
Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment