unix-ffi/os.linux: Add Linux-specific system call wrappers.

New package providing ioctl helpers, mount/umount/reboot, execv/execvp,
setenv, realpath, and device access modules for MTD, SPI, I2C, evdev,
block devices, syslog and monotonic clocks via FFI.

Made-with: Cursor

Author: dbrignoli

unix-ffi/os.linux/manifest.py

@@ -0,0 +1,6 @@
+metadata(version="0.1.0", description="Linux-specific system call wrappers (ioctl, mount, reboot, MTD, SPI, I2C, evdev, syslog, clock_gettime).")
+
+require("ffilib")
+require("os")
+
+package("os/linux")

unix-ffi/os.linux/os/linux/__init__.py

@@ -0,0 +1,71 @@
+
+import ffilib
+import os
+
+
+libc = ffilib.libc()
+sleep = libc.func('I', 'sleep', 'I')
+_mount = libc.func('i', 'mount', 'sssLs')
+_umount = libc.func('i', 'umount', 's')
+_setenv = libc.func('i', 'setenv', 'ssi')
+_reboot_syscall = libc.func('l', 'syscall', 'liiis')
+_realpath = libc.func("s", "realpath", "ss")
+_free = libc.func("v", "free", "p")
+
+
+
+LINUX_REBOOT_MAGIC1         = 0xfee1dead
+LINUX_REBOOT_MAGIC2         = 672274793
+LINUX_REBOOT_CMD_RESTART	= 0x01234567
+LINUX_REBOOT_CMD_HALT		= 0xCDEF0123
+LINUX_REBOOT_CMD_CAD_ON		= 0x89ABCDEF
+LINUX_REBOOT_CMD_CAD_OFF	= 0x00000000
+LINUX_REBOOT_CMD_POWER_OFF	= 0x4321FEDC
+LINUX_REBOOT_CMD_RESTART2	= 0xA1B2C3D4
+LINUX_REBOOT_CMD_SW_SUSPEND	= 0xD000FCE2
+LINUX_REBOOT_CMD_KEXEC		= 0x45584543
+
+
+def reboot(cmd, arg_str):
+	SYS_reboot = 142
+	e = _reboot_syscall(SYS_reboot, LINUX_REBOOT_MAGIC1, LINUX_REBOOT_MAGIC2, cmd, arg_str)
+	os.check_error(e)
+
+
+def mount(source, target, fstype, flags = 0, opts = None):
+	e = _mount(source, target, fstype, flags, opts)
+	os.check_error(e)
+
+
+def umount(target):
+	e = _umount(target)
+	os.check_error(e)
+
+
+def execv(path, args = []):
+	assert args, '`args` argument cannot be empty'
+	_args = [path] + args + [None]
+	_execl = libc.func('i', 'execl', 's'*len(_args))
+	e = _execl(*_args)
+	os.check_error(e)
+
+
+def execvp(executable, args = []):
+	assert args, '`args` argument cannot be empty'
+	_args = [executable] + args + [None]
+	_execlp = libc.func('i', 'execlp', 's'*len(_args))
+	e = _execlp(*_args)
+	os.check_error(e)
+
+
+def setenv(name, value, overwrite = True):
+	e = _setenv(name, value, 1 if overwrite else 0)
+	os.check_error(e)
+
+def realpath(path):
+	p = _realpath(path, None)
+	# Do not free the string here, it is tracked by upy and freed by it.
+	# _free(p)
+	if p == None:
+		os.raise_error()
+	return p

unix-ffi/os.linux/os/linux/blkdev.py

@@ -0,0 +1,31 @@
+
+# buffer data from src_fobj (it could be the stdin pipe) before writing
+# it out in blocksz bytes chunks
+def block_copy(src_fobj, dst_fobj, byte_count = 0, blocksz=512, progress_func=None):
+	buf = bytearray(blocksz)
+	buf_view = memoryview(buf)
+	read_bytes = 0
+	left_to_read = byte_count
+	total_written_bytes = 0
+	while True:
+		sz = src_fobj.readinto(buf_view[read_bytes:])
+		#print('readinto() -> {}, {}'.format(read_bytes, sz), file=sys.stderr)
+		# in blocking mode sz will be zero only on EOF
+		left_to_read -= sz
+		read_bytes += sz
+		if not sz or (byte_count != 0 and left_to_read <= 0):
+			cnt = read_bytes
+			if byte_count != 0 and left_to_read <= 0:
+				cnt += left_to_read
+			dst_fobj.write(buf_view[:cnt])
+			total_written_bytes += cnt
+			if progress_func is not None:
+				progress_func(total_written_bytes)
+			break
+		if read_bytes == blocksz:
+			dst_fobj.write(buf_view)
+			#print('write({})'.format(read_bytes), file=sys.stderr)
+			total_written_bytes += read_bytes
+			read_bytes = 0
+			if progress_func is not None:
+				progress_func(total_written_bytes)

unix-ffi/os.linux/os/linux/evdev.py

@@ -0,0 +1,18 @@
+
+import builtins
+import os
+from os.linux import ioctl
+
+DEV_PATH_TPL = '/dev/input/event{}'
+
+KEY_SELECT = 0x161
+
+def EVIOCGKEY(len):
+    return ioctl._ioc(ioctl._IOC_READ, ord('E'), 0x18, len)
+
+def get_global_keystate(dev, byte_array):
+    r = ioctl.ioctl_p(dev.fileno(), EVIOCGKEY(len(byte_array)), byte_array)
+    os.check_error(r)
+
+def open(index=0):
+    return builtins.open(DEV_PATH_TPL.format(index), 'r+b')

unix-ffi/os.linux/os/linux/i2c.py

@@ -0,0 +1,13 @@
+
+import builtins
+from os.linux import ioctl
+
+DEV_PATH_TPL = '/dev/i2c-{}'
+
+I2C_SLAVE = 0x0703
+
+def set_slave_addr(dev_fd, addr):
+    ioctl.ioctl_l(dev_fd, I2C_SLAVE, addr)
+
+def open(index=0):
+    return builtins.open(DEV_PATH_TPL.format(index), 'r+b')

unix-ffi/os.linux/os/linux/ioctl.py

@@ -0,0 +1,27 @@
+
+import ffilib
+
+libc = ffilib.libc()
+
+TIOCCONS = 0x541D
+
+_IOC_WRITE = 1
+_IOC_READ = 2
+
+_IOC_NRBITS = 8
+_IOC_TYPEBITS = 8
+_IOC_SIZEBITS = 14
+_IOC_DIRBITS = 2
+
+_IOC_NRSHIFT = 0
+_IOC_TYPESHIFT = (_IOC_NRSHIFT+_IOC_NRBITS)
+_IOC_SIZESHIFT = (_IOC_TYPESHIFT+_IOC_TYPEBITS)
+_IOC_DIRSHIFT = (_IOC_SIZESHIFT+_IOC_SIZEBITS)
+
+def _ioc(dir, type, num, size):
+	return ((num << _IOC_NRSHIFT) | (type << _IOC_TYPESHIFT) |
+		(size << _IOC_SIZESHIFT) | (dir << _IOC_DIRSHIFT))
+
+ioctl_p = libc.func("i", "ioctl", "iip")
+ioctl_l = libc.func("i", "ioctl", "iil")
+del libc

unix-ffi/os.linux/os/linux/mtd.py

@@ -0,0 +1,67 @@
+
+import os
+from os.linux import ioctl
+try:
+	import ustruct as struct
+except:
+	import struct
+
+
+# struct erase_info_user {
+#     __u32 start;
+#     __u32 length;
+# };
+
+MTD_NORFLASH = 3
+
+def _mtd_ioc(dir, num, size):
+    return ioctl._ioc(dir, ord('M'), num, size)
+
+MEM_INFO_STRUCT = 'BIIIIIII'
+_MEM_INFO_LEN = struct.calcsize(MEM_INFO_STRUCT)
+ERASE_INFO_STRUCT = 'II'
+_ERASE_INFO_LEN = struct.calcsize(ERASE_INFO_STRUCT)
+MEMGETINFO = _mtd_ioc(ioctl._IOC_READ, 1, _MEM_INFO_LEN)
+MEMERASE = _mtd_ioc(ioctl._IOC_WRITE, 2, _ERASE_INFO_LEN)
+MEMUNLOCK = _mtd_ioc(ioctl._IOC_WRITE, 6, _ERASE_INFO_LEN)
+
+
+def _pack_erase_info(start, length):
+	return struct.pack(ERASE_INFO_STRUCT, start, length)
+
+
+def _check_offset(offset, eblock):
+	assert offset % eblock == 0, \
+		'offset {} is not a multiple of erase block {}'.format(offset, eblock)
+
+
+def info(mtd_dev):
+	mem_info = bytearray(_MEM_INFO_LEN)
+	e = ioctl.ioctl_p(mtd_dev.fileno(), MEMGETINFO, mem_info)
+	os.check_error(e)
+	# do not return padding words
+	return struct.unpack(MEM_INFO_STRUCT, mem_info)[:-2]
+
+
+def unlock(mtd_dev, offset, eblock_size):
+	_check_offset(offset, eblock_size)
+	erase_info_user = _pack_erase_info(offset, eblock_size)
+	e = ioctl.ioctl_p(mtd_dev.fileno(), MEMUNLOCK, erase_info_user)
+	os.check_error(e)
+
+
+def erase(mtd_dev, offset, eblock_size):
+	_check_offset(offset, eblock_size)
+	erase_info_user = _pack_erase_info(offset, eblock_size)
+	e = ioctl.ioctl_p(mtd_dev.fileno(), MEMERASE, erase_info_user)
+	os.check_error(e)
+
+
+def read(mtd_dev, offset, length):
+	assert mtd_dev.seek(offset) == offset
+	return mtd_dev.read(length)
+
+
+def write(mtd_dev, offset, data):
+	assert mtd_dev.seek(offset) == offset
+	return mtd_dev.write(data)

unix-ffi/os.linux/os/linux/spidev.py

@@ -0,0 +1,17 @@
+
+import builtins
+from os.linux import ioctl
+
+DEV_PATH_TPL = '/dev/spidev{}.{}'
+
+def _spidev_ioc(dir, num, size):
+    return ioctl._ioc(dir, ord('k'), num, size)
+
+SPI_IOC_RD_BITS_PER_WORD = _spidev_ioc(ioctl._IOC_READ, 3, 1)
+SPI_IOC_WR_BITS_PER_WORD = _spidev_ioc(ioctl._IOC_WRITE, 3, 1)
+
+SPI_IOC_RD_MAX_SPEED_HZ = _spidev_ioc(ioctl._IOC_READ, 4, 4)
+SPI_IOC_WR_MAX_SPEED_HZ = _spidev_ioc(ioctl._IOC_WRITE, 4, 4)
+
+def open(bus_idx=0, cs_idx=0):
+    return builtins.open(DEV_PATH_TPL.format(bus_idx, cs_idx), 'r+b')

unix-ffi/os.linux/os/linux/syslog.py

@@ -0,0 +1,99 @@
+
+import sys
+import os
+from os import libc
+
+openlog_ = libc.func("v", "openlog", "sii")
+setlogmask_ = libc.func("i", "setlogmask", "i")
+syslog_ = libc.func("v", "syslog", "is")
+isatty_ = libc.func("i", "isatty", "i")
+
+# Syslog priorities
+CRITICAL = 2
+ERROR    = 3
+WARNING  = 4
+NOTICE   = 5
+INFO     = 6
+DEBUG    = 7
+NOTSET   = 0
+
+# Facility codes
+LOG_USER    = (1<<3) # random user-level messages
+
+# Option flags for openlog
+LOG_PID     = 0x01 # log the pid with each message
+LOG_CONS    = 0x02 # log on the console if errors in sending
+LOG_ODELAY  = 0x04 # delay open until first syslog() (default)
+LOG_NDELAY  = 0x08 # don't delay open
+LOG_NOWAIT  = 0x10 # don't wait for console forks: DEPRECATED
+LOG_PERROR  = 0x20 # log to stderr as well
+
+def _logmask_upto(pri):
+	return ((1<<((pri)+1))-1)
+
+class Logger:
+
+    def __init__(self, name):
+        self.name = name
+
+    def log(self, level, msg, *args):
+        if self.name is not None:
+            s = ('%s:'+ msg) % ((self.name,) + args)
+        else:
+            s = msg % args
+        syslog_(level, s)
+
+    def debug(self, msg, *args):
+        self.log(DEBUG, msg, *args)
+
+    def info(self, msg, *args):
+        self.log(INFO, msg, *args)
+
+    def warning(self, msg, *args):
+        self.log(WARNING, msg, *args)
+
+    def error(self, msg, *args):
+        self.log(ERROR, msg, *args)
+
+    def critical(self, msg, *args):
+        self.log(CRITICAL, msg, *args)
+
+
+_level = ERROR
+_loggers = {}
+
+r = isatty_(sys.stdout.fileno())
+os.check_error(r)
+
+flags = LOG_CONS | LOG_PID
+# if we are outputting to a tty log also to stderr
+if r > 0:
+    flags |= LOG_PERROR
+ident = 'python'
+if len(sys.argv):
+    ident = sys.argv[0]
+openlog_(ident, flags, LOG_USER)
+
+r = setlogmask_(_logmask_upto(_level))
+os.check_error(r)
+
+def getLogger(name):
+    if name in _loggers:
+        return _loggers[name]
+    l = Logger(name)
+    _loggers[name] = l
+    return l
+
+def info(msg, *args):
+    getLogger(None).info(msg, *args)
+
+def debug(msg, *args):
+    getLogger(None).debug(msg, *args)
+
+def basicConfig(level=INFO, filename=None, format=None):
+    global _level
+    _level = level
+    if filename is not None:
+        print("logging.basicConfig: filename arg is not supported")
+    if format is not None:
+        print("logging.basicConfig: format arg is not supported")

unix-ffi/os.linux/os/linux/time.py

@@ -0,0 +1,40 @@
+
+import ustruct
+import ffilib
+import os
+
+
+TIMESPEC_FMT = 'll'
+TIMESPEC_SIZE = ustruct.calcsize(TIMESPEC_FMT)
+TIMEVAL_FMT = TIMESPEC_FMT
+TIMEVAL_SIZE = ustruct.calcsize(TIMEVAL_FMT)
+
+CLOCK_MONOTONIC = 1
+CLOCK_MONOTONIC_RAW = 4
+
+librt = ffilib.open('librt')
+_clock_gettime = librt.func('i', 'clock_gettime', 'ip')
+_gettimeofday = librt.func('i', 'gettimeofday', 'ip')
+_ts_buf = bytearray(TIMESPEC_SIZE)
+_tv_buf = bytearray(TIMEVAL_SIZE)
+
+def clock_gettimeofday(clk_id):
+	e1 = _clock_gettime(clk_id, _ts_buf)
+	e2 = _gettimeofday(_tv_buf, None)
+	os.check_error(e1)
+	os.check_error(e2)
+	s, ns = ustruct.unpack(TIMESPEC_FMT, _ts_buf)
+	utc_s, utc_us = ustruct.unpack(TIMEVAL_FMT, _tv_buf)
+	return s, ns, utc_s, utc_us
+
+def clock_gettime(clk_id):
+	e = _clock_gettime(clk_id, _ts_buf)
+	os.check_error(e)
+	s, ns = ustruct.unpack(TIMESPEC_FMT, _ts_buf)
+	return (s*10**9)+ns
+
+def monotime():
+	return clock_gettime(CLOCK_MONOTONIC)
+
+def monotime_raw():
+	return clock_gettime(CLOCK_MONOTONIC_RAW)