| #!/usr/bin/env python |
| |
| # Copyright (c) 2013 The Chromium OS Authors. All rights reserved. |
| # Use of this source code is governed by a BSD-style license that can be |
| # found in the LICENSE file. |
| |
| import cmd |
| import dbus |
| import dbus.exceptions |
| import dbus.mainloop.glib |
| import gobject |
| import threading |
| |
| from functools import wraps |
| |
| |
| DBUS_ERROR = 'org.freedesktop.DBus.Error' |
| NEARD_PATH = '/org/neard/' |
| PROMPT = 'NFC> ' |
| |
| class NfcClientException(Exception): |
| """Exception class for exceptions thrown by NfcClient.""" |
| |
| |
| def print_message(message, newlines=2): |
| """ |
| Prints the given message with extra wrapping newline characters. |
| |
| @param message: Message to print. |
| @param newlines: Integer, specifying the number of '\n' characters that |
| should be padded at the beginning and end of |message| before |
| being passed to "print". |
| |
| """ |
| padding = newlines * '\n' |
| message = padding + message + padding |
| print message |
| |
| |
| def handle_errors(func): |
| """ |
| Decorator for handling exceptions that are commonly raised by many of the |
| methods in NfcClient. |
| |
| @param func: The function this decorator is wrapping. |
| |
| """ |
| @wraps(func) |
| def _error_handler(*args): |
| try: |
| return func(*args) |
| except dbus.exceptions.DBusException as e: |
| if e.get_dbus_name() == DBUS_ERROR + '.ServiceUnknown': |
| print_message('neard may have crashed or disappeared. ' |
| 'Check if neard is running and run "initialize" ' |
| 'from this shell.') |
| return |
| if e.get_dbus_name() == DBUS_ERROR + '.UnknownObject': |
| print_message('Could not find object.') |
| return |
| print_message(str(e)) |
| except Exception as e: |
| print_message(str(e)) |
| return _error_handler |
| |
| |
| class NfcClient(object): |
| """ |
| neard D-Bus client |
| |
| """ |
| NEARD_SERVICE_NAME = 'org.neard' |
| IMANAGER = NEARD_SERVICE_NAME + '.Manager' |
| IADAPTER = NEARD_SERVICE_NAME + '.Adapter' |
| ITAG = NEARD_SERVICE_NAME + '.Tag' |
| IRECORD = NEARD_SERVICE_NAME + '.Record' |
| IDEVICE = NEARD_SERVICE_NAME + '.Device' |
| |
| def __init__(self): |
| self._mainloop = None |
| self._mainloop_thread = None |
| self._adapters = {} |
| self._adapter_property_handler_matches = {} |
| |
| def begin(self): |
| """ |
| Starts the D-Bus client. |
| |
| """ |
| # Here we run a GLib MainLoop in its own thread, so that the client can |
| # listen to D-Bus signals while keeping the console interactive. |
| self._dbusmainloop = dbus.mainloop.glib.DBusGMainLoop( |
| set_as_default=True) |
| dbus.mainloop.glib.threads_init() |
| gobject.threads_init() |
| |
| def _mainloop_thread_func(): |
| self._mainloop = gobject.MainLoop() |
| context = self._mainloop.get_context() |
| self._run_loop = True |
| while self._run_loop: |
| context.iteration(True) |
| self._mainloop_thread = threading.Thread(None, _mainloop_thread_func) |
| self._mainloop_thread.start() |
| |
| self._bus = dbus.SystemBus() |
| self.setup_manager() |
| |
| def end(self): |
| """ |
| Stops the D-Bus client. |
| |
| """ |
| self._run_loop = False |
| self._mainloop.quit() |
| self._mainloop_thread.join() |
| |
| def restart(self): |
| """Reinitializes the NFC client.""" |
| self.setup_manager() |
| |
| @handle_errors |
| def _get_manager_proxy(self): |
| return dbus.Interface( |
| self._bus.get_object(self.NEARD_SERVICE_NAME, '/'), |
| self.IMANAGER) |
| |
| @handle_errors |
| def _get_adapter_proxy(self, adapter): |
| return dbus.Interface( |
| self._bus.get_object(self.NEARD_SERVICE_NAME, adapter), |
| self.IADAPTER) |
| |
| def _get_cached_adapter_proxy(self, adapter): |
| adapter_proxy = self._adapters.get(adapter, None) |
| if not adapter_proxy: |
| raise NfcClientException('Adapter "' + adapter + '" not found.') |
| return adapter_proxy |
| |
| |
| @handle_errors |
| def _get_tag_proxy(self, tag): |
| return dbus.Interface( |
| self._bus.get_object(self.NEARD_SERVICE_NAME, tag), |
| self.ITAG) |
| |
| @handle_errors |
| def _get_device_proxy(self, device): |
| return dbus.Interface( |
| self._bus.get_object(self.NEARD_SERVICE_NAME, device), |
| self.IDEVICE) |
| |
| @handle_errors |
| def _get_record_proxy(self, record): |
| return dbus.Interface( |
| self._bus.get_object(self.NEARD_SERVICE_NAME, record), |
| self.IRECORD) |
| |
| @handle_errors |
| def _get_adapter_properties(self, adapter): |
| adapter_proxy = self._get_cached_adapter_proxy(adapter) |
| return adapter_proxy.GetProperties() |
| |
| def _get_adapters(self): |
| props = self._manager.GetProperties() |
| return props.get('Adapters', None) |
| |
| def setup_manager(self): |
| """ |
| Creates a manager proxy and subscribes to adapter signals. This method |
| will also initialize proxies for adapters if any are available. |
| |
| """ |
| # Create the manager proxy. |
| self._adapters.clear() |
| self._manager = self._get_manager_proxy() |
| if not self._manager: |
| print_message('Failed to create a proxy to the Manager interface.') |
| return |
| |
| # Listen to the adapter added and removed signals. |
| self._manager.connect_to_signal( |
| 'AdapterAdded', |
| lambda adapter: self.register_adapter(str(adapter))) |
| self._manager.connect_to_signal( |
| 'AdapterRemoved', |
| lambda adapter: self.unregister_adapter(str(adapter))) |
| |
| # See if there are any adapters and create proxies for each. |
| adapters = self._get_adapters() |
| if adapters: |
| for adapter in adapters: |
| self.register_adapter(adapter) |
| |
| def register_adapter(self, adapter): |
| """ |
| Registers an adapter proxy with the given object path and subscribes to |
| adapter signals. |
| |
| @param adapter: string, containing the adapter's D-Bus object path. |
| |
| """ |
| print_message('Added adapter: ' + adapter) |
| adapter_proxy = self._get_adapter_proxy(adapter) |
| self._adapters[adapter] = adapter_proxy |
| |
| # Tag found/lost currently don't get fired. Monitor property changes |
| # instead. |
| if self._adapter_property_handler_matches.get(adapter, None) is None: |
| self._adapter_property_handler_matches[adapter] = ( |
| adapter_proxy.connect_to_signal( |
| 'PropertyChanged', |
| (lambda name, value: |
| self._adapter_property_changed_signal( |
| adapter, name, value)))) |
| |
| def unregister_adapter(self, adapter): |
| """ |
| Removes the adapter proxy for the given object path from the internal |
| cache of adapters. |
| |
| @param adapter: string, containing the adapter's D-Bus object path. |
| |
| """ |
| print_message('Removed adapter: ' + adapter) |
| match = self._adapter_property_handler_matches.get(adapter, None) |
| if match is not None: |
| match.remove() |
| self._adapter_property_handler_matches.pop(adapter) |
| self._adapters.pop(adapter) |
| |
| def _adapter_property_changed_signal(self, adapter, name, value): |
| if name == 'Tags' or name == 'Devices': |
| print_message('Found ' + name + ': ' + |
| self._dbus_array_to_string(value)) |
| |
| @handle_errors |
| def show_adapters(self): |
| """ |
| Prints the D-Bus object paths of all adapters that are available. |
| |
| """ |
| adapters = self._get_adapters() |
| if not adapters: |
| print_message('No adapters found.') |
| return |
| for adapter in adapters: |
| print_message(' ' + str(adapter), newlines=0) |
| print |
| |
| def _dbus_array_to_string(self, array): |
| string = '[ ' |
| for value in array: |
| string += ' ' + str(value) + ', ' |
| string += ' ]' |
| return string |
| |
| def print_adapter_status(self, adapter): |
| """ |
| Prints the properties of the given adapter. |
| |
| @param adapter: string, containing the adapter's D-Bus object path. |
| |
| """ |
| props = self._get_adapter_properties(adapter) |
| if not props: |
| return |
| print_message('Status ' + adapter + ': ', newlines=0) |
| for key, value in props.iteritems(): |
| if type(value) == dbus.Array: |
| value = self._dbus_array_to_string(value) |
| else: |
| value = str(value) |
| print_message(' ' + key + ' = ' + value, newlines=0) |
| print |
| |
| @handle_errors |
| def set_powered(self, adapter, powered): |
| """ |
| Enables or disables the adapter. |
| |
| @param adapter: string, containing the adapter's D-Bus object path. |
| @param powered: boolean that dictates whether the adapter will be |
| enabled or disabled. |
| |
| """ |
| adapter_proxy = self._get_cached_adapter_proxy(adapter) |
| if not adapter_proxy: |
| return |
| adapter_proxy.SetProperty('Powered', powered) |
| |
| @handle_errors |
| def start_polling(self, adapter): |
| """ |
| Starts polling for nearby tags and devices in "Initiator" mode. |
| |
| @param adapter: string, containing the adapter's D-Bus object path. |
| |
| """ |
| adapter_proxy = self._get_cached_adapter_proxy(adapter) |
| adapter_proxy.StartPollLoop('Initiator') |
| print_message('Started polling.') |
| |
| @handle_errors |
| def stop_polling(self, adapter): |
| """ |
| Stops polling for nearby tags and devices. |
| |
| @param adapter: string, containing the adapter's D-Bus object path. |
| |
| """ |
| adapter_proxy = self._get_cached_adapter_proxy(adapter) |
| adapter_proxy.StopPollLoop() |
| self._polling_stopped = True |
| print_message('Stopped polling.') |
| |
| @handle_errors |
| def show_tag_data(self, tag): |
| """ |
| Prints the properties of the given tag, as well as the contents of any |
| records associated with it. |
| |
| @param tag: string, containing the tag's D-Bus object path. |
| |
| """ |
| tag_proxy = self._get_tag_proxy(tag) |
| if not tag_proxy: |
| print_message('Tag "' + tag + '" not found.') |
| return |
| props = tag_proxy.GetProperties() |
| print_message('Tag ' + tag + ': ', newlines=1) |
| for key, value in props.iteritems(): |
| if key != 'Records': |
| print_message(' ' + key + ' = ' + str(value), newlines=0) |
| records = props['Records'] |
| if not records: |
| return |
| print_message('Records: ', newlines=1) |
| for record in records: |
| self.show_record_data(str(record)) |
| print |
| |
| @handle_errors |
| def show_device_data(self, device): |
| """ |
| Prints the properties of the given device, as well as the contents of |
| any records associated with it. |
| |
| @param device: string, containing the device's D-Bus object path. |
| |
| """ |
| device_proxy = self._get_device_proxy(device) |
| if not device_proxy: |
| print_message('Device "' + device + '" not found.') |
| return |
| records = device_proxy.GetProperties()['Records'] |
| if not records: |
| print_message('No records on device.') |
| return |
| print_message('Records: ', newlines=1) |
| for record in records: |
| self.show_record_data(str(record)) |
| print |
| |
| @handle_errors |
| def show_record_data(self, record): |
| """ |
| Prints the contents of the given record. |
| |
| @param record: string, containing the record's D-Bus object path. |
| |
| """ |
| record_proxy = self._get_record_proxy(record) |
| if not record_proxy: |
| print_message('Record "' + record + '" not found.') |
| return |
| props = record_proxy.GetProperties() |
| print_message('Record ' + record + ': ', newlines=1) |
| for key, value in props.iteritems(): |
| print ' ' + key + ' = ' + value |
| print |
| |
| def _create_record_data(self, record_type, params): |
| if record_type == 'Text': |
| possible_keys = [ 'Encoding', 'Language', 'Representation' ] |
| tag_data = { 'Type': 'Text' } |
| elif record_type == 'URI': |
| possible_keys = [ 'URI' ] |
| tag_data = { 'Type': 'URI' } |
| else: |
| print_message('Writing record type "' + record_type + |
| '" currently not supported.') |
| return None |
| for key, value in params.iteritems(): |
| if key in possible_keys: |
| tag_data[key] = value |
| return tag_data |
| |
| @handle_errors |
| def write_tag(self, tag, record_type, params): |
| """ |
| Writes an NDEF record to the given tag. |
| |
| @param tag: string, containing the tag's D-Bus object path. |
| @param record_type: The type of the record, e.g. Text or URI. |
| @param params: dictionary, containing the parameters of the NDEF. |
| |
| """ |
| tag_data = self._create_record_data(record_type, params) |
| if not tag_data: |
| return |
| tag_proxy = self._get_tag_proxy(tag) |
| if not tag_proxy: |
| print_message('Tag "' + tag + '" not found.') |
| return |
| tag_proxy.Write(tag_data) |
| print_message('Tag written!') |
| |
| @handle_errors |
| def push_to_device(self, device, record_type, params): |
| """ |
| Pushes an NDEF record to the given device. |
| |
| @param device: string, containing the device's D-Bus object path. |
| @param record_type: The type of the record, e.g. Text or URI. |
| @param params: dictionary, containing the parameters of the NDEF. |
| |
| """ |
| record_data = self._create_record_data(record_type, params) |
| if not record_data: |
| return |
| device_proxy = self._get_device_proxy(device) |
| if not device_proxy: |
| print_message('Device "' + device + '" not found.') |
| return |
| device_proxy.Push(record_data) |
| print_message('NDEF pushed to device!') |
| |
| |
| class NfcConsole(cmd.Cmd): |
| """ |
| Interactive console to interact with the NFC daemon. |
| |
| """ |
| def __init__(self): |
| cmd.Cmd.__init__(self) |
| self.prompt = PROMPT |
| |
| def begin(self): |
| """ |
| Starts the interactive shell. |
| |
| """ |
| print_message('NFC console! Run "help" for a list of commands.', |
| newlines=1) |
| self._nfc_client = NfcClient() |
| self._nfc_client.begin() |
| self.cmdloop() |
| |
| def can_exit(self): |
| """Override""" |
| return True |
| |
| def do_initialize(self, args): |
| """Handles "initialize".""" |
| if args: |
| print_message('Command "initialize" expects no arguments.') |
| return |
| self._nfc_client.restart() |
| |
| def help_initialize(self): |
| """Prints the help message for "initialize".""" |
| print_message('Initializes the neard D-Bus client. This can be ' |
| 'run many times to restart the client in case of ' |
| 'neard failures or crashes.') |
| |
| def do_adapters(self, args): |
| """Handles "adapters".""" |
| if args: |
| print_message('Command "adapters" expects no arguments.') |
| return |
| self._nfc_client.show_adapters() |
| |
| def help_adapters(self): |
| """Prints the help message for "adapters".""" |
| print_message('Displays the D-Bus object paths of the available ' |
| 'adapter objects.') |
| |
| def do_adapter_status(self, args): |
| """Handles "adapter_status".""" |
| args = args.strip().split(' ') |
| if len(args) != 1 or not args[0]: |
| print_message('Usage: adapter_status <adapter>') |
| return |
| self._nfc_client.print_adapter_status(NEARD_PATH + args[0]) |
| |
| def help_adapter_status(self): |
| """Prints the help message for "adapter_status".""" |
| print_message('Returns the properties of the given NFC adapter.\n\n' |
| ' Ex: "adapter_status nfc0"') |
| |
| def do_enable_adapter(self, args): |
| """Handles "enable_adapter".""" |
| args = args.strip().split(' ') |
| if len(args) != 1 or not args[0]: |
| print_message('Usage: enable_adapter <adapter>') |
| return |
| self._nfc_client.set_powered(NEARD_PATH + args[0], True) |
| |
| def help_enable_adapter(self): |
| """Prints the help message for "enable_adapter".""" |
| print_message('Powers up the adapter. Ex: "enable_adapter nfc0"') |
| |
| def do_disable_adapter(self, args): |
| """Handles "disable_adapter".""" |
| args = args.strip().split(' ') |
| if len(args) != 1 or not args[0]: |
| print_message('Usage: disable_adapter <adapter>') |
| return |
| self._nfc_client.set_powered(NEARD_PATH + args[0], False) |
| |
| def help_disable_adapter(self): |
| """Prints the help message for "disable_adapter".""" |
| print_message('Powers down the adapter. Ex: "disable_adapter nfc0"') |
| |
| def do_start_poll(self, args): |
| """Handles "start_poll".""" |
| args = args.strip().split(' ') |
| if len(args) != 1 or not args[0]: |
| print_message('Usage: start_poll <adapter>') |
| return |
| self._nfc_client.start_polling(NEARD_PATH + args[0]) |
| |
| def help_start_poll(self): |
| """Prints the help message for "start_poll".""" |
| print_message('Initiates a poll loop.\n\n Ex: "start_poll nfc0"') |
| |
| def do_stop_poll(self, args): |
| """Handles "stop_poll".""" |
| args = args.split(' ') |
| if len(args) != 1 or not args[0]: |
| print_message('Usage: stop_poll <adapter>') |
| return |
| self._nfc_client.stop_polling(NEARD_PATH + args[0]) |
| |
| def help_stop_poll(self): |
| """Prints the help message for "stop_poll".""" |
| print_message('Stops a poll loop.\n\n Ex: "stop_poll nfc0"') |
| |
| def do_read_tag(self, args): |
| """Handles "read_tag".""" |
| args = args.strip().split(' ') |
| if len(args) != 1 or not args[0]: |
| print_message('Usage read_tag <tag>') |
| return |
| self._nfc_client.show_tag_data(NEARD_PATH + args[0]) |
| |
| def help_read_tag(self): |
| """Prints the help message for "read_tag".""" |
| print_message('Reads the contents of a tag. Ex: read_tag nfc0/tag0') |
| |
| def _parse_record_args(self, record_type, args): |
| if record_type == 'Text': |
| if len(args) < 5: |
| print_message('Usage: write_tag <tag> Text <encoding> ' |
| '<language> <representation>') |
| return None |
| if args[2] not in [ 'UTF-8', 'UTF-16' ]: |
| print_message('Encoding must be one of "UTF-8" or "UTF-16".') |
| return None |
| return { |
| 'Encoding': args[2], |
| 'Language': args[3], |
| 'Representation': ' '.join(args[4:]) |
| } |
| if record_type == 'URI': |
| if len(args) != 3: |
| print_message('Usage: write_tag <tag> URI <uri>') |
| return None |
| return { |
| 'URI': args[2] |
| } |
| print_message('Only types "Text" and "URI" are supported by this ' |
| 'script.') |
| return None |
| |
| def do_write_tag(self, args): |
| """Handles "write_tag".""" |
| args = args.strip().split(' ') |
| if len(args) < 3: |
| print_message('Usage: write_tag <tag> [params]') |
| return |
| record_type = args[1] |
| params = self._parse_record_args(record_type, args) |
| if not params: |
| return |
| self._nfc_client.write_tag(NEARD_PATH + args[0], |
| record_type, params) |
| |
| def help_write_tag(self): |
| """Prints the help message for "write_tag".""" |
| print_message('Writes the given data to a tag. Usage:\n' |
| ' write_tag <tag> Text <encoding> <language> ' |
| '<representation>\n write_tag <tag> URI <uri>') |
| |
| def do_read_device(self, args): |
| """Handles "read_device".""" |
| args = args.strip().split(' ') |
| if len(args) != 1 or not args[0]: |
| print_message('Usage read_device <device>') |
| return |
| self._nfc_client.show_device_data(NEARD_PATH + args[0]) |
| |
| def help_read_device(self): |
| """Prints the help message for "read_device".""" |
| print_message('Reads the contents of a device. Ex: read_device ' |
| 'nfc0/device0') |
| |
| def do_push_to_device(self, args): |
| """Handles "push_to_device".""" |
| args = args.strip().split(' ') |
| if len(args) < 3: |
| print_message('Usage: push_to_device <device> [params]') |
| return |
| record_type = args[1] |
| params = self._parse_record_args(record_type, args) |
| if not params: |
| return |
| self._nfc_client.push_to_device(NEARD_PATH + args[0], |
| record_type, params) |
| |
| def help_push_to_device(self): |
| """Prints the help message for "push_to_device".""" |
| print_message('Pushes the given data to a device. Usage:\n' |
| ' push_to_device <device> Text <encoding> <language> ' |
| '<representation>\n push_to_device <device> URI <uri>') |
| |
| def do_exit(self, args): |
| """ |
| Handles the 'exit' command. |
| |
| @param args: Arguments to the command. Unused. |
| |
| """ |
| if args: |
| print_message('Command "exit" expects no arguments.') |
| return |
| resp = raw_input('Are you sure? (yes/no): ') |
| if resp == 'yes': |
| print_message('Goodbye!') |
| self._nfc_client.end() |
| return True |
| if resp != 'no': |
| print_message('Did not understand: ' + resp) |
| return False |
| |
| def help_exit(self): |
| """Handles the 'help exit' command.""" |
| print_message('Exits the console.') |
| |
| do_EOF = do_exit |
| help_EOF = help_exit |
| |
| |
| def main(): |
| """Main function.""" |
| NfcConsole().begin() |
| |
| |
| if __name__ == '__main__': |
| main() |
