Python源码示例：binascii.crc32()
示例1
def cmh_autotune(self, module, message, additional_data, cm):
message = message[len(self.CONTROL_AUTOTUNE)+2:]
# get tune type, requested record type, length and encoding for crafting the answer
(query_type, RRtype, length, encode_class) = struct.unpack("<BHHH", message[0:7])
if self.DNS_proto.get_RR_type(RRtype)[0] == None:
return True
# extra parameters added to be able to response in the proper way
additional_data = additional_data + (True, self.download_encoding_list[encode_class], self.DNS_proto.get_RR_type(RRtype)[0])
if (query_type == 0) or (query_type == 3):
# record && downstream length discovery
message = ''.join(random.choice(string.ascii_letters + string.digits) for _ in range(length))
if query_type == 1:
# A record name length discovery
message = struct.pack("<i", binascii.crc32(message[7:]))
if query_type == 2:
# checking download encoding, echoing back request payload
message = message[7:]
module.send(common.CONTROL_CHANNEL_BYTE, self.CONTROL_AUTOTUNE_CLIENT+message, additional_data)
return True
# tune control message handler
# client sets the record type and encodings by calling this
# server side
示例2
def _GenerateCRCTable():
"""Generate a CRC-32 table.
ZIP encryption uses the CRC32 one-byte primitive for scrambling some
internal keys. We noticed that a direct implementation is faster than
relying on binascii.crc32().
"""
poly = 0xedb88320
table = [0] * 256
for i in range(256):
crc = i
for j in range(8):
if crc & 1:
crc = ((crc >> 1) & 0x7FFFFFFF) ^ poly
else:
crc = ((crc >> 1) & 0x7FFFFFFF)
table[i] = crc
return table
示例3
def get_checksum(cls, phrase):
"""Given a mnemonic word string, return a string of the computed checksum.
:rtype: str
"""
phrase_split = phrase.split(" ")
if len(phrase_split) < 12:
raise ValueError("Invalid mnemonic phrase")
if len(phrase_split) > 13:
# Standard format
phrase = phrase_split[:24]
else:
# MyMonero format
phrase = phrase_split[:12]
wstr = "".join(word[:cls.unique_prefix_length] for word in phrase)
wstr = bytearray(wstr.encode('utf-8'))
z = ((crc32(wstr) & 0xffffffff) ^ 0xffffffff ) >> 0
z2 = ((z ^ 0xffffffff) >> 0) % len(phrase)
return phrase_split[z2]
示例4
def send(self, rumble_high=0, rumble_low=0, red=0, green=0, blue=0, light_on=0, light_off=0):
"""Actuate the controller by setting its rumble or light color/blink"""
packet = bytearray(79)
packet[:5] = [0xA2, 0x11, 0x80, 0x00, 0xFF]
packet[7] = int(rumble_high * 255 + 0.5)
packet[8] = int(rumble_low * 255 + 0.5)
packet[9] = int(red * 255 + 0.5)
packet[10] = int(green * 255 + 0.5)
packet[11] = int(blue * 255 + 0.5)
packet[12] = int(light_on * 255 + 0.5)
packet[13] = int(light_off * 255 + 0.5)
crc = crc32(packet[:-4])
packet[-4] = crc & 0x000000FF
packet[-3] = (crc & 0x0000FF00) >> 8
packet[-2] = (crc & 0x00FF0000) >> 16
packet[-1] = (crc & 0xFF000000) >> 24
hid = bytearray((self.__report_id,))
if self.__fd is not None:
self.__fd.write(hid + packet[2:])
return True
return False
示例5
def _GenerateCRCTable():
"""Generate a CRC-32 table.
ZIP encryption uses the CRC32 one-byte primitive for scrambling some
internal keys. We noticed that a direct implementation is faster than
relying on binascii.crc32().
"""
poly = 0xedb88320
table = [0] * 256
for i in range(256):
crc = i
for j in range(8):
if crc & 1:
crc = ((crc >> 1) & 0x7FFFFFFF) ^ poly
else:
crc = ((crc >> 1) & 0x7FFFFFFF)
table[i] = crc
return table
示例6
def _GenerateCRCTable():
"""Generate a CRC-32 table.
ZIP encryption uses the CRC32 one-byte primitive for scrambling some
internal keys. We noticed that a direct implementation is faster than
relying on binascii.crc32().
"""
poly = 0xedb88320
table = [0] * 256
for i in range(256):
crc = i
for j in range(8):
if crc & 1:
crc = ((crc >> 1) & 0x7FFFFFFF) ^ poly
else:
crc = ((crc >> 1) & 0x7FFFFFFF)
table[i] = crc
return table
示例7
def delim_hash(value: str, delim_list: str = r'[\s\-\\/\.,"\'|&:;%$()]') -> int:
r"""
Return a hash (CRC32) of the delimiters from input column.
Parameters
----------
value : str
Data to process
delim_list : str, optional
delimiters to use. (the default is r'[\\s\\\\-\\\\\\\\/\.,"\\\\'|&:;%$()]')
Returns
-------
int
Hash of delimiter set in the string.
"""
return crc32(bytes("".join(re.findall(delim_list, value)), "utf-8"))
示例8
def crc32_hash(value: str) -> int:
"""
Return the CRC32 hash of the input column.
Parameters
----------
value : str
Data to process
Returns
-------
int
CRC32 hash
"""
return crc32(bytes(value.encode("utf-8")))
示例9
def crc32_hash_df(data: pd.DataFrame, column: str) -> pd.Series:
"""
Return the CRC32 hash of the input column.
Parameters
----------
data : pd.DataFrame
The DataFrame to process
column : str
Column name to process
Returns
-------
pd.Series
CRC32 hash of input column
"""
return data.apply(lambda x: crc32(bytes(x[column].encode("utf-8"))), axis=1)
# pylint: disable=too-many-arguments, too-many-statements
示例10
def shard_key(base, key, total_elements, shard_size): #A
if isinstance(key, (int, long)) or key.isdigit(): #B
shard_id = int(str(key), 10) // shard_size #C
else:
shards = 2 * total_elements // shard_size #D
shard_id = binascii.crc32(key) % shards #E
return "%s:%s"%(base, shard_id) #F
# <end id="calculate-shard-key"/>
#A We will call the shard_key() function with a base HASH name, along with the key to be stored in the sharded HASH, the total number of expected elements, and the desired shard size
#B If the value is an integer or a string that looks like an integer, we will use it directly to calculate the shard id
#C For integers, we assume they are sequentially assigned ids, so we can choose a shard id based on the upper 'bits' of the numeric id itself. We also use an explicit base here (necessitating the str() call) so that a key of '010' turns into 10, and not 8
#D For non-integer keys, we first calculate the total number of shards desired, based on an expected total number of elements and desired shard size
#E When we know the number of shards we want, we hash the key and find its value modulo the number of shards we want
#F Finally, we combine the base key with the shard id we calculated to determine the shard key
#END
# <start id="sharded-hset-hget"/>
示例11
def sampler(dataframe, modulo, column="client_id", sample_id=42):
""" Collect a sample of clients given an input column
Filter dataframe based on the modulus of the CRC32 of a given string
column matching a given sample_id. if dataframe has already been filtered
by sample_id, then modulo should be a multiple of 100, column should be
"client_id", and the given sample_id should match the value previously
used, optionally plus multiples of 100.
Args:
dataframe: A Dataframe to be sampled
modulo (int): selects a 1/modulo sampling of dataframe
column (str): name of a string column to sample on
sample_id (int): modulus result to select for sampling
Returns:
A DataFrame sampled on the given inputs.
"""
return dataframe \
.withColumn(
"sampler",
udf(lambda key: (crc32(key or "") & 0xffffffff) % modulo)(column),
).where("sampler = %s" % sample_id).drop("sampler")
示例12
def _GenerateCRCTable():
"""Generate a CRC-32 table.
ZIP encryption uses the CRC32 one-byte primitive for scrambling some
internal keys. We noticed that a direct implementation is faster than
relying on binascii.crc32().
"""
poly = 0xedb88320
table = [0] * 256
for i in range(256):
crc = i
for j in range(8):
if crc & 1:
crc = ((crc >> 1) & 0x7FFFFFFF) ^ poly
else:
crc = ((crc >> 1) & 0x7FFFFFFF)
table[i] = crc
return table
示例13
def _GenerateCRCTable():
"""Generate a CRC-32 table.
ZIP encryption uses the CRC32 one-byte primitive for scrambling some
internal keys. We noticed that a direct implementation is faster than
relying on binascii.crc32().
"""
poly = 0xedb88320
table = [0] * 256
for i in range(256):
crc = i
for j in range(8):
if crc & 1:
crc = ((crc >> 1) & 0x7FFFFFFF) ^ poly
else:
crc = ((crc >> 1) & 0x7FFFFFFF)
table[i] = crc
return table
示例14
def __init__(self, fileobj, mode, zipinfo, decrypter=None,
close_fileobj=False):
self._fileobj = fileobj
self._decrypter = decrypter
self._close_fileobj = close_fileobj
self._compress_type = zipinfo.compress_type
self._compress_left = zipinfo.compress_size
self._left = zipinfo.file_size
self._decompressor = _get_decompressor(self._compress_type)
self._eof = False
self._readbuffer = b''
self._offset = 0
self._universal = 'U' in mode
self.newlines = None
# Adjust read size for encrypted files since the first 12 bytes
# are for the encryption/password information.
if self._decrypter is not None:
self._compress_left -= 12
self.mode = mode
self.name = zipinfo.filename
if hasattr(zipinfo, 'CRC'):
self._expected_crc = zipinfo.CRC
self._running_crc = crc32(b'') & 0xffffffff
else:
self._expected_crc = None
示例15
def _update_crc(self, newdata):
# Update the CRC using the given data.
if self._expected_crc is None:
# No need to compute the CRC if we don't have a reference value
return
self._running_crc = crc32(newdata, self._running_crc) & 0xffffffff
# Check the CRC if we're at the end of the file
if self._eof and self._running_crc != self._expected_crc:
raise BadZipFile("Bad CRC-32 for file %r" % self.name)
示例16
def _readPage(self, page, tries=3):
"""Read a page from the flash and receive it's contents"""
self._debug('Command: FLASH_READ_PAGE %d' % page)
# Load page into the buffer
crc = self._loadPageMultiple(page, tries)
for _ in range(tries):
# Dump the buffer
self._sendCommand(COMMAND_BUFFER_LOAD)
# Wait for data start
if not self._waitForMessage(COMMAND_BUFFER_LOAD):
self._debug('Invalid / no response for BUFFER_LOAD command')
continue
# Load successful -> read sector with 2 nibbles per byte
page_data = self._readExactly(self.page_size * 2)
if page_data is None:
self._debug('Invalid / no response for page data')
continue
try:
data = binascii.a2b_hex(page_data.decode(ENCODING))
if crc == binascii.crc32(data):
self._debug('CRC did match with read data')
return data
else:
self._debug('CRC did not match with read data')
continue
except TypeError:
self._debug('CRC could not be parsed')
continue
self._debug('Page read tries exceeded')
return None
示例17
def get_file_crc32(file):
buffer = get_file_buffer(file=file)
crc_buffer = (binascii.crc32(buffer) & 0xFFFFFFFF)
return crc_buffer
示例18
def rar_crc32(data, prev=0):
"""CRC32 with unsigned values.
"""
if (prev > 0) and (prev & 0x80000000):
prev -= (1 << 32)
res = crc32(data, prev)
if res < 0:
res += (1 << 32)
return res
示例19
def make_chunk(chunk_type, chunk_data):
"""Create a raw chunk by composing chunk type and data. It
calculates chunk length and CRC for you.
:arg str chunk_type: PNG chunk type.
:arg bytes chunk_data: PNG chunk data, **excluding chunk length, type, and CRC**.
:rtype: bytes
"""
out = struct.pack("!I", len(chunk_data))
chunk_data = chunk_type.encode("latin-1") + chunk_data
out += chunk_data + struct.pack("!I", binascii.crc32(chunk_data) & 0xffffffff)
return out
示例20
def MAC(flags, handle, signingKey, seqNum, message):
# [MS-NLMP] Section 3.4.4
# Returns the right messageSignature depending on the flags
messageSignature = NTLMMessageSignature(flags)
if flags & NTLMSSP_NEGOTIATE_EXTENDED_SESSIONSECURITY:
if flags & NTLMSSP_NEGOTIATE_KEY_EXCH:
messageSignature['Version'] = 1
messageSignature['Checksum'] = \
struct.unpack('<q', handle(hmac_md5(signingKey, struct.pack('<i', seqNum) + message)[:8]))[0]
messageSignature['SeqNum'] = seqNum
seqNum += 1
else:
messageSignature['Version'] = 1
messageSignature['Checksum'] = struct.unpack('<q',hmac_md5(signingKey, struct.pack('<i',seqNum)+message)[:8])[0]
messageSignature['SeqNum'] = seqNum
seqNum += 1
else:
messageSignature['Version'] = 1
messageSignature['Checksum'] = struct.pack('<i',binascii.crc32(message))
messageSignature['RandomPad'] = 0
messageSignature['RandomPad'] = handle(struct.pack('<i',messageSignature['RandomPad']))
messageSignature['Checksum'] = struct.unpack('<i',handle(messageSignature['Checksum']))[0]
messageSignature['SeqNum'] = handle('\x00\x00\x00\x00')
messageSignature['SeqNum'] = struct.unpack('<i',messageSignature['SeqNum'])[0] ^ seqNum
messageSignature['RandomPad'] = 0
return messageSignature
示例21
def compute_checksum(self,bytes):
crcle=crc32(bytes)&0xffffffffL
# ggrr this crc32 is in little endian, convert it to big endian
crc=struct.pack('<L', crcle)
# Convert to long
(crc_long,) = struct.unpack('!L', crc)
return crc_long
示例22
def set_fcs(self, value = None):
"Set the 802.11 CTS control frame 'FCS' field. If value is None, is auto_checksum"
if not self.__FCS_at_end:
return
# calculate the FCS
if value is None:
payload = self.get_body_as_string()
crc32=self.compute_checksum(payload)
value=crc32
# set the bits
nb = value & 0xFFFFFFFF
self.tail.set_long(-4, nb)
示例23
def get_computed_icv(self):
crcle=crc32(self.body_string)&0xffffffffL
# This crc32 is in little endian, convert it to big endian
crc=struct.pack('<L', crcle)
# Convert to long
(crc_long,) = struct.unpack('!L', crc)
return crc_long
示例24
def save(self, filename):
with open(filename, 'wb') as f:
# Save first header for irom0 segment
f.write(struct.pack(b'<BBBBI', ESPBOOTLOADER.IMAGE_V2_MAGIC, ESPBOOTLOADER.IMAGE_V2_SEGMENT,
self.flash_mode, self.flash_size_freq, self.entrypoint))
irom_segment = self.get_irom_segment()
if irom_segment is not None:
# save irom0 segment, make sure it has load addr 0 in the file
irom_segment = irom_segment.copy_with_new_addr(0)
irom_segment.pad_to_alignment(16) # irom_segment must end on a 16 byte boundary
self.save_segment(f, irom_segment)
# second header, matches V1 header and contains loadable segments
normal_segments = self.get_non_irom_segments()
self.write_common_header(f, normal_segments)
checksum = ESPLoader.ESP_CHECKSUM_MAGIC
for segment in normal_segments:
checksum = self.save_segment(f, segment, checksum)
self.append_checksum(f, checksum)
# calculate a crc32 of entire file and append
# (algorithm used by recent 8266 SDK bootloaders)
with open(filename, 'rb') as f:
crc = esp8266_crc32(f.read())
with open(filename, 'ab') as f:
f.write(struct.pack(b'<I', crc))
# Backwards compatibility for previous API, remove in esptool.py V3
示例25
def esp8266_crc32(data):
"""
CRC32 algorithm used by 8266 SDK bootloader (and gen_appbin.py).
"""
crc = binascii.crc32(data, 0) & 0xFFFFFFFF
if crc & 0x80000000:
return crc ^ 0xFFFFFFFF
else:
return crc + 1
示例26
def _verify_checksum(cls, raw, crc):
expected = binascii.crc32(raw) & 0xffffffff # https://docs.python.org/3/library/binascii.html#binascii.crc32
if crc != expected:
raise MessageChecksumError("Message checksum mismatch: 0x%X != 0x%X" % (crc, expected))
示例27
def _compute_checksum(cls, raw):
crc = binascii.crc32(raw) & 0xffffffff # https://docs.python.org/3/library/binascii.html#binascii.crc32
return crc
示例28
def __init__(self, fileobj, mode, zipinfo, decrypter=None):
self._fileobj = fileobj
self._decrypter = decrypter
self._compress_type = zipinfo.compress_type
self._compress_size = zipinfo.compress_size
self._compress_left = zipinfo.compress_size
if self._compress_type == ZIP_DEFLATED:
self._decompressor = zlib.decompressobj(-15)
self._unconsumed = ''
self._readbuffer = ''
self._offset = 0
self._universal = 'U' in mode
self.newlines = None
# Adjust read size for encrypted files since the first 12 bytes
# are for the encryption/password information.
if self._decrypter is not None:
self._compress_left -= 12
self.mode = mode
self.name = zipinfo.filename
if hasattr(zipinfo, 'CRC'):
self._expected_crc = zipinfo.CRC
self._running_crc = crc32(b'') & 0xffffffff
else:
self._expected_crc = None
示例29
def _update_crc(self, newdata, eof):
# Update the CRC using the given data.
if self._expected_crc is None:
# No need to compute the CRC if we don't have a reference value
return
self._running_crc = crc32(newdata, self._running_crc) & 0xffffffff
# Check the CRC if we're at the end of the file
if eof and self._running_crc != self._expected_crc:
raise BadZipfile("Bad CRC-32 for file %r" % self.name)
示例30
def __init__(self, fileobj, mode, zipinfo, decrypter=None,
close_fileobj=False):
self._fileobj = fileobj
self._decrypter = decrypter
self._close_fileobj = close_fileobj
self._compress_type = zipinfo.compress_type
self._compress_size = zipinfo.compress_size
self._compress_left = zipinfo.compress_size
if self._compress_type == ZIP_DEFLATED:
self._decompressor = zlib.decompressobj(-15)
elif self._compress_type != ZIP_STORED:
descr = compressor_names.get(self._compress_type)
if descr:
raise NotImplementedError("compression type %d (%s)" % (self._compress_type, descr))
else:
raise NotImplementedError("compression type %d" % (self._compress_type,))
self._unconsumed = ''
self._readbuffer = ''
self._offset = 0
self._universal = 'U' in mode
self.newlines = None
# Adjust read size for encrypted files since the first 12 bytes
# are for the encryption/password information.
if self._decrypter is not None:
self._compress_left -= 12
self.mode = mode
self.name = zipinfo.filename
if hasattr(zipinfo, 'CRC'):
self._expected_crc = zipinfo.CRC
self._running_crc = crc32(b'') & 0xffffffff
else:
self._expected_crc = None
