python: added iio_writedev with Python bindings.

Implemented the iio_writedev module with Python bindings.

Signed-off-by: Cristi Iacob <[email protected]>

bindings/python/examples/iio_writedev.py

@@ -0,0 +1,223 @@
+#!/usr/bin/env python
+#
+# Copyright (C) 2020 Analog Devices, Inc.
+# Author: Cristian Iacob <[email protected]>
+#
+# This program is free software; you can redistribute it and/or
+# modify it under the terms of the GNU General Public License
+# as published by the Free Software Foundation; either version 2
+# of the License, or (at your option) any later version.
+#
+# This program is distributed in the hope that it will be useful,
+# but WITHOUT ANY WARRANTY; without even the implied warranty of
+# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
+# GNU General Public License for more details.
+#
+# You should have received a copy of the GNU General Public License
+# along with this program; if not, write to the Free Software
+# Foundation, Inc., 51 Franklin Street, Fifth Floor, Boston, MA  02110-1301, USA.
+import time
+
+import iio
+import getopt
+import sys
+import signal
+
+
+class Option:
+    def __init__(self, name, alias, description):
+        self.name = name
+        self.alias = alias
+        self.description = description
+
+
+options = [
+    Option('help', 'h', 'Show this help and quit.'),
+    Option('network', 'n', 'Use the network backend with the provided hostname.'),
+    Option('uri', 'u', 'Use the context with the provided URI.'),
+    Option('buffer-size', 'b', 'Size of the capture buffer. Default is 256.'),
+    Option('samples', 's', 'Number of samples to capture, 0 = infinite. Default is 0.'),
+    Option('timeout', 'T', 'Buffer timeout in milliseconds. 0 = no timeout'),
+    Option('auto', 'a', 'Scan for available contexts and if only one is available use it.'),
+    Option('cyclic', 'c', 'Use cyclic buffer mode.')
+]
+
+usage_message = 'Usage:\n\t' \
+                'iio_readdev [-n <hostname>] [-T <timeout-ms>] [-b <buffer-size>] ' \
+                '[-s <samples>] [-c <cyclic>] <iio_device> [<channel> ...]\n'
+
+unixOptions = 'hn:u:b:s:T:ac'
+gnuOptions = ['help', 'network', 'uri', 'buffer-size', 'samples', 'timeout', 'auto', 'cyclic']
+
+
+def get_device_and_channels(arguments):
+    arguments = iter(arguments)
+    arg = next(arguments, None)
+
+    while True:
+        if arg == '-h' or arg == '--help' or arg == '-a' or arg == '--auto' or arg == '-c' or arg == '--cyclic':
+            arg = next(arguments, None)
+        elif arg == '-n' or arg == '--network' or arg == '-u' or arg == '--uri' or \
+                arg == '-b' or arg == '--buffer-size' or arg == '-s' or arg == '--samples' or \
+                arg == '-T' or arg == '--timeout':
+            next(arguments)
+            arg = next(arguments, None)
+        else:
+            break
+
+    # arg should contain the device name or None if there is no device
+    ret_list = [arg]
+
+    while arg is not None:
+        arg = next(arguments, None)
+
+        if arg is not None:
+            ret_list += [arg]
+
+    return ret_list
+
+
+def usage():
+    print(usage_message)
+    print('Options:')
+    for option in options:
+        print('\t-' + option.alias + ', --' + option.name)
+        print('\t\t\t', option.description)
+
+
+def keyboard_interrupt_handler(signal, frame):
+    sys.exit(0)
+
+
+signal.signal(signal.SIGINT, keyboard_interrupt_handler)
+
+
+def main():
+    raw_arguments = sys.argv[1:]
+
+    try:
+        arguments, _ = getopt.getopt(raw_arguments, unixOptions, gnuOptions)
+    except getopt.error as err:
+        sys.stderr.write(str(err))
+        sys.exit(1)
+
+    arg_ip = ''
+    arg_uri = ''
+    scan_for_context = False
+    buffer_size = 256
+    num_samples = 0
+    timeout = 0
+    cyclic = False
+
+    for argument, value in arguments:
+        if argument in ('-h', '--help'):
+            usage()
+            exit(0)
+        elif argument in ('-n', '--network'):
+            arg_ip = str(value)
+        elif argument in ('-u', '--uri'):
+            arg_uri = str(value)
+        elif argument in ('-a', '--auto'):
+            scan_for_context = True
+        elif argument in ('-b', '--buffer-size'):
+            buffer_size = int(value)
+        elif argument in ('-s', '--samples'):
+            num_samples = int(value)
+        elif argument in ('-T', '--timeout'):
+            timeout = int(timeout)
+        elif argument in ('-c', '--cyclic'):
+            cyclic = True
+
+    device_name = get_device_and_channels(raw_arguments)[0]
+    channels = get_device_and_channels(raw_arguments)[1:]
+
+    if device_name is None:
+        sys.stderr.write('Incorrect number of arguments.')
+        usage()
+        exit(1)
+
+    ctx = None
+
+    try:
+        if scan_for_context:
+            contexts = iio.scan_contexts()
+            if len(contexts) == 0:
+                sys.stderr.write('No IIO context found.')
+                exit(1)
+            elif len(contexts) == 1:
+                uri, _ = contexts.popitem()
+                ctx = iio.Context(_context=uri)
+            else:
+                print('Multiple contexts found. Please select one using --uri!')
+
+                for uri, _ in contexts:
+                    print(uri)
+        elif arg_uri != '':
+            ctx = iio.Context(_context=arg_uri)
+        elif arg_ip != '':
+            ctx = iio.NetworkContext(arg_ip)
+        else:
+            ctx = iio.Context()
+    except FileNotFoundError:
+        sys.stderr.write('Unable to create IIO context')
+        exit(1)
+
+    if timeout >= 0:
+        ctx.set_timeout(timeout)
+
+    dev = ctx.find_device(device_name)
+
+    if dev is None:
+        sys.stderr.write('Device %s not found!' % device_name)
+        exit(1)
+
+    if len(channels) == 0:
+        for channel in dev.channels:
+            channel.enabled = True
+    else:
+        for channel_idx in channels:
+            dev.channels[int(channel_idx)].enabled = True
+
+    buffer = iio.Buffer(dev, buffer_size, cyclic=cyclic)
+
+    if buffer is None:
+        sys.stderr.write('Unable to create buffer!')
+        exit(1)
+
+    app_running = True
+    num_samples_set = num_samples > 0
+
+    while app_running:
+        bytes_to_read = buffer._length if not num_samples_set else min(buffer._length, num_samples * dev.sample_size)
+        write_len = bytes_to_read
+        data = []
+
+        while bytes_to_read > 0:
+            read_data = sys.stdin.buffer.read(bytes_to_read)
+            bytes_to_read -= len(read_data)
+            data.extend(read_data)
+
+            if len(read_data) == 0:
+                exit(0)
+
+        if num_samples_set:
+            num_samples -= int(write_len / dev.sample_size)
+
+            if num_samples == 0 and write_len < buffer_size * dev.sample_size:
+                exit(0)
+
+        ret = buffer.write(bytearray(data))
+        buffer.push()
+
+        if ret == 0:
+            if app_running:
+                sys.stderr.write('Unable to push buffer!')
+                exit(1)
+            break
+
+        while app_running and cyclic:
+            time.sleep(1)
+
+
+if __name__ == '__main__':
+    main()