Read from a BLE address.

        @param: mac - MAC address in format XX:XX:XX:XX:XX:XX
        @param: handle - BLE characteristics handle in format 0xXX
        @param: value - value to write to the given handle
14159
def write_handle(self, handle: int, value: bytes):
        # noqa: C901
        # pylint: disable=arguments-differ

        

        if not self.is_connected():
            raise BluetoothBackendException('Not connected to any device.')

        attempt = 0
        delay = 10
        _LOGGER.debug("Enter write_ble (%s)", current_thread())

        while attempt <= self.retries:
            cmd = "gatttool --device={} --addr-type={} --char-write-req -a {} -n {} --adapter={}".format(
                self._mac, self.address_type, self.byte_to_handle(handle), self.bytes_to_string(value), self.adapter)
            _LOGGER.debug("Running gatttool with a timeout of %d: %s",
                          self.timeout, cmd)

            with Popen(cmd,
                       shell=True,
                       stdout=PIPE,
                       stderr=PIPE,
                       preexec_fn=os.setsid) as process:
                try:
                    result = process.communicate(timeout=self.timeout)[0]
                    _LOGGER.debug("Finished gatttool")
                except TimeoutExpired:
                    # send signal to the process group
                    os.killpg(process.pid, signal.SIGINT)
                    result = process.communicate()[0]
                    _LOGGER.debug("Killed hanging gatttool")

            result = result.decode("utf-8").strip(' \n\t')
            if "Write Request failed" in result:
                raise BluetoothBackendException('Error writing handle to sensor: {}'.format(result))
            _LOGGER.debug("Got %s from gatttool", result)
            # Parse the output
            if "successfully" in result:
                _LOGGER.debug(
                    "Exit write_ble with result (%s)", current_thread())
                return True

            attempt += 1
            _LOGGER.debug("Waiting for %s seconds before retrying", delay)
            if attempt < self.retries:
                time.sleep(delay)
                delay *= 2

        raise BluetoothBackendException("Exit write_ble, no data ({})".format(current_thread()))
1032006
@Override
    public void setReadAhead(String value) throws JMSException {
        if (TraceComponent.isAnyTracingEnabled() && tc.isEntryEnabled())
            SibTr.entry(this, tc, "setReadAhead", value);

        // Check for null then ensure that the value is one of the permitted constants.
        if ((value != null)
            && ((ApiJmsConstants.READ_AHEAD_DEFAULT.equals(value))
                || (ApiJmsConstants.READ_AHEAD_ON.equals(value))
                || (ApiJmsConstants.READ_AHEAD_OFF.equals(value))
            )) {
            jcaManagedConnectionFactory.setReadAhead(value);
        }
        else {
            throw (JMSException) JmsErrorUtils.newThrowable(
                                                            JMSException.class,
                                                            "INVALID_VALUE_CWSIA0261",
                                                            new Object[] { "readAhead", value },
                                                            tc
                            );
        }

        if (TraceComponent.isAnyTracingEnabled() && tc.isEntryEnabled())
            SibTr.exit(this, tc, "setReadAhead");
    }
732069
def read_packet(sock, timeout=None):
    
    sock.settimeout(timeout)
    dlen, data = None, None
    try:
        if os.name == 'nt':
            #  Windows implementation
            datalen = sock.recv(SZ)
            dlen, = struct.unpack("l", datalen)
            data = b''
            while len(data) < dlen:
                data += sock.recv(dlen)
        else:
            #  Linux/MacOSX implementation
            #  Thanks to eborisch:
            #  See issue 1106
            datalen = temp_fail_retry(socket.error, sock.recv,
                                      SZ, socket.MSG_WAITALL)
            if len(datalen) == SZ:
                dlen, = struct.unpack("l", datalen)
                data = temp_fail_retry(socket.error, sock.recv,
                                       dlen, socket.MSG_WAITALL)
    except socket.timeout:
        raise
    except socket.error:
        data = None
    finally:
        sock.settimeout(None)
    if data is not None:
        try:
            return pickle.loads(data)
        except Exception:
            # Catch all exceptions to avoid locking spyder
            if DEBUG_EDITOR:
                traceback.print_exc(file=STDERR)
            return
865692
def write_handle(self, handle: int, value: bytes):
        """"""
        if not self.is_connected():
            raise BluetoothBackendException('Not connected to device!')
        self._device.char_write_handle(handle, value, True)
        return True
172098
def write(self, value):
        
        if not isinstance(value, bool):
            raise TypeError("Invalid value type, should be bool.")

        # Write value
        try:
            if value:
                os.write(self._fd, b"1\n")
            else:
                os.write(self._fd, b"0\n")
        except OSError as e:
            raise GPIOError(e.errno, "Writing GPIO: " + e.strerror)

        # Rewind
        try:
            os.lseek(self._fd, 0, os.SEEK_SET)
        except OSError as e:
            raise GPIOError(e.errno, "Rewinding GPIO: " + e.strerror)
786709
def read_value(self, offset=0):
        
        try:
            val = self._object.ReadValue(
                {'offset': dbus.UInt16(offset, variant_level=1)},
                dbus_interface='org.bluez.GattDescriptor1')
            return val
        except dbus.exceptions.DBusException as e:
            error = _error_from_dbus_error(e)
            self.service.device.descriptor_read_value_failed(self, error=error)
1278642
def range_mac(mac_start, mac_end, step=1):
    """"""
    start = int(EUI(mac_start))
    end = int(EUI(mac_end))
    for i_mac in range(start, end, step):
        mac = EUI(int(EUI(i_mac)) + 1)
        ip = ['10'] + [str(int(i, 2)) for i in mac.bits().split('-')[-3:]]
        yield str(mac).replace('-', ':'), '.'.join(ip)
33014
def _send_byte(self,value):
        

        # Coerce to int. This will throw a ValueError if the value can't
        # actually be converted.
        if type(value) != int:
            new_value = int(value)

            if self.give_warnings:
                w = "Coercing {} into int ({})".format(value,new_value)
                warnings.warn(w,Warning)
                value = new_value

        # Range check
        if value > 255 or value < 0:
            err = "Value {} exceeds the size of the board's byte.".format(value)
            raise OverflowError(err)

        return struct.pack("B",value)
429718
def device_characteristics_str(self, indent):
        
        s = "{}\n".format(self.label)
        s += indent + "MAC Address: {}\n".format(self.mac_addr)
        s += indent + "IP Address: {}\n".format(self.ip_addr)
        s += indent + "Port: {}\n".format(self.port)
        s += indent + "Power: {}\n".format(str_map(self.power_level))
        s += indent + "Location: {}\n".format(self.location)
        s += indent + "Group: {}\n".format(self.group)
        return s
1079721
def _read_byte(self, address):
        
        # Initialize memory location with a random value.
        if address not in self._memory:
            self._memory[address] = random.randint(0x00, 0xff)

        return self._memory[address]
859712
function readSingleAddress(conn, handleAddress){
	readSingleWallet(conn, function(wallet_id){
		conn.query("SELECT address FROM my_addresses WHERE wallet=?", [wallet_id], function(rows){
			if (rows.length === 0)
				throw Error("no addresses");
			if (rows.length > 1)
				throw Error("more than 1 address");
			handleAddress(rows[0].address);
		});
	})
}
499966
def mac_addr
      case RUBY_PLATFORM
      when /darwin/
        if (m = `ifconfig en0`.match(/ether (\S{17})/))
          m[1].upcase
        end
      when /linux/
        devices = Dir['/sys/class/net/*/address'].reject { |a| a =~ %r{/lo/} }
        dev = devices.first
        File.read(dev).chomp.upcase
      end
    end
165757
def write_byte(self, address, value):
        
        LOGGER.debug("Writing byte %s to device %s!", bin(value), hex(address))
        return self.driver.write_byte(address, value)
618163
public static byte[] parseMAC(String value) {
        final byte[] machineId;
        final char separator;
        switch (value.length()) {
            case 17:
                separator = value.charAt(2);
                validateMacSeparator(separator);
                machineId = new byte[EUI48_MAC_ADDRESS_LENGTH];
                break;
            case 23:
                separator = value.charAt(2);
                validateMacSeparator(separator);
                machineId = new byte[EUI64_MAC_ADDRESS_LENGTH];
                break;
            default:
                throw new IllegalArgumentException("value is not supported [MAC-48, EUI-48, EUI-64]");
        }

        final int end = machineId.length - 1;
        int j = 0;
        for (int i = 0; i < end; ++i, j += 3) {
            final int sIndex = j + 2;
            machineId[i] = StringUtil.decodeHexByte(value, j);
            if (value.charAt(sIndex) != separator) {
                throw new IllegalArgumentException("expected separator '" + separator + " but got '" +
                        value.charAt(sIndex) + "' at index: " + sIndex);
            }
        }

        machineId[end] = StringUtil.decodeHexByte(value, j);

        return machineId;
    }
364818
def write(self, offset, value):
		
		if(not self.mode & 0b10):
			raise ReadOnlyError("Device is Read-Only")
		if(offset >= self.size):
			raise AddressError("Offset({}) not in address space({})".format(offset, self.size))
		self.repr_[offset].setvalue(value)
1282802
def mac_address(self):
        
        if self._mac_address is None:
            self._mac_address = self._get_mac_address()

        return self._mac_address
391884
def read_instruction(fio, start_pos):
    
    op = fio.read(1)

    if not op:
        return None

    op = ord(op)

    ins = opcode_table[op]
    operands = ins['operands']
    name = ins['mnemonic']

    final_operands = []
    # Most opcodes have simple operands.
    if operands:
        for fmt, type_ in operands:
            final_operands.append(
                Operand(
                    type_,
                    fmt.value.unpack(fio.read(fmt.value.size))[0]
                )
            )
    # Special case for lookupswitch.
    elif op == 0xAB:
        # Get rid of the alignment padding.
        padding = 4 - (start_pos + 1) % 4
        padding = padding if padding != 4 else 0
        fio.read(padding)

        # Default branch address and branch count.
        default, npairs = unpack('>ii', fio.read(8))

        pairs = {}
        for _ in repeat(None, npairs):
            match, offset = unpack('>ii', fio.read(8))
            pairs[match] = offset

        final_operands.append(pairs)
        final_operands.append(Operand(OperandTypes.BRANCH, default))
    # Special case for tableswitch
    elif op == 0xAA:
        # Get rid of the alignment padding.
        padding = 4 - (start_pos + 1) % 4
        padding = padding if padding != 4 else 0
        fio.read(padding)

        default, low, high = unpack('>iii', fio.read(12))
        final_operands.append(Operand(OperandTypes.BRANCH, default))
        final_operands.append(Operand(OperandTypes.LITERAL, low))
        final_operands.append(Operand(OperandTypes.LITERAL, high))

        for _ in repeat(None, high - low + 1):
            offset = unpack('>i', fio.read(4))[0]
            final_operands.append(Operand(OperandTypes.BRANCH, offset))
    # Special case for the wide prefix
    elif op == 0xC4:
        real_op = unpack('>B', fio.read(1))[0]
        ins = opcode_table[real_op]
        name = ins['mnemonic']

        final_operands.append(Operand(
            OperandTypes.LOCAL_INDEX,
            unpack('>H', fio.read(2))[0]
        ))
        # Further special case for iinc.
        if real_op == 0x84:
            final_operands.append(Operand(
                OperandTypes.LITERAL,
                unpack('>H', fio.read(2))[0]
            ))

    return Instruction(name, op, final_operands, start_pos)