How to use parse_num method in Pytest

Best Python code snippet using pytest

protocol.py

Source:protocol.py Github

copy

Full Screen

...231 names = list(mapping.keys())232 unpacker = bitstruct.compile(fmt, names=names)233 # decode with bit reversal234 return lambda seq: unpacker.unpack(bytes(reverse_bits8(seq)))235def parse_num(encoded, signed, *, inval=None, num_bytes=None):236 """Parse a sequence of bytes into a number237 Args:238 encoded: sequence of ints or bytes239 signed: whether this is a signed value or not240 inval: optionally a number that represents invalid data241 num_bytes: number of bytes to extract from encoded (if not provided,242 all is used)243 Returns:244 the value or NaN if matching the invalid code245 """246 # optionally deduce num_bytes247 if num_bytes is None:248 num_bytes = len(encoded)249 if num_bytes > 4:250 raise ValueError("num_bytes not specified")251 # extract from bytes252 num = 0253 for val in reversed(encoded[:num_bytes]):254 num = num*256 + int(val)255 # replace by nan if matching invalid256 if inval is not None and num == inval:257 num = math.nan258 # convert to two's complement259 if signed and encoded[num_bytes-1] > 127:260 num -= 2**(8*num_bytes)261 return num262def parse_timestamp(encoded):263 """Parse a bytes-encoded UNIX timestamp."""264 year = parse_num(encoded[0:2], False)265 month, day = encoded[2:4]266 msec = parse_num(encoded[4:8], False)267 stamp = date2stamp_cached(year, month, day) + msec * 0.001268 return stamp269class Message:270 """A message that can be exchanged with the device (sent or received)."""271 def __init__(self, msgid, payload=(), fin=MC.ETX):272 self.msgid = msgid273 if not isinstance(payload, (list, tuple, bytes)):274 raise TypeError("payload must be a list, tuple, or bytes.")275 self.payload = payload276 self.fin = fin277 @classmethod278 def assert_length(cls, payload, expected_len, at_least=False):279 """Check if the length matches (or exceeds) the given expected length."""280 if (at_least and len(payload) < expected_len) or (not at_least and len(payload) != expected_len):281 raise AssertionError(f"{cls.__name__} requires at least {expected_len} "282 f"bytes of payload, but got {len(payload)}.")283 def payload_str(self, encoding='utf-8'):284 """Get the payload as a string."""285 return bytes(self.payload).decode(encoding)286 def ensure_fin_ok(self):287 """Checks if fin was OK, and otherwise raises an error"""288 if self.fin not in (MC.ACK, MC.ETX):289 raise RuntimeError(f"Error invoking {MI(self.msgid).name}: {self}")290 def as_dict(self):291 """Get the content as a dictionary."""292 return {k: v for k, v in self.__dict__.items()293 if not k.startswith('_') and k not in ('msgid', 'payload', 'fin')}294 def __str__(self):295 """Render a human-readable string."""296 content = ', '.join([f'{k}={v}' for k, v in self.__dict__.items() if not k.startswith('_')])297 return f'{self.__class__.__name__}({content})'298class StreamingMessage(Message):299 """Base class for messages that have a sequence num and time stamp."""300 def __init__(self, msgid, payload, fin=MC.ETX):301 self.assert_length(payload, 9, at_least=True)302 super().__init__(msgid, payload, fin)303 self.seq_no = payload[0]304 self.stamp = parse_timestamp(payload[1:])305class GeneralDataMessage(StreamingMessage):306 """A general data packet with various slow-changing state."""307 srate = 1.0308 def __init__(self, msgid, payload, fin=MC.ETX):309 self.assert_length(payload, 53)310 super().__init__(msgid, payload, fin)311 self.heart_rate = parse_num(payload[9:11], False, inval=0xFFFF)312 self.respiration_rate = parse_num(payload[11:13], False, inval=0xFFFF) * 0.1313 self.skin_temperature = parse_num(payload[13:15], True, inval=0x8000) * 0.1314 self.posture = parse_num(payload[15:17], True, inval=0x8000)315 self.vmu_activity = parse_num(payload[17:19], False, inval=0xFFFF) * 0.01316 self.peak_acceleration = parse_num(payload[19:21], False, inval=0xFFFF) * 0.01317 self.battery_voltage = parse_num(payload[21:23], False, inval=0xFFFF) * 0.001318 self.breathing_wave_amplitude = parse_num(payload[23:25], False, inval=0xFFFF)319 self.ecg_amplitude = parse_num(payload[25:27], False, inval=0xFFFF) * 0.000001320 self.ecg_noise = parse_num(payload[27:29], False, inval=0xFFFF) * 0.000001321 self.vertical_accel_min = parse_num(payload[29:31], True, inval=0x8000) * 0.01322 self.vertical_accel_peak = parse_num(payload[31:33], True, inval=0x8000) * 0.01323 self.lateral_accel_min = parse_num(payload[33:35], True, inval=0x8000) * 0.01324 self.lateral_accel_peak = parse_num(payload[35:37], True, inval=0x8000) * 0.01325 self.sagittal_accel_min = parse_num(payload[37:39], True, inval=0x8000) * 0.01326 self.sagittal_accel_peak = parse_num(payload[39:41], True, inval=0x8000) * 0.01327 self.system_channel = parse_num(payload[41:43], False)328 self.gsr = parse_num(payload[43:45], False, inval=0xFFFF)329 self.unused1 = parse_num(payload[45:47], False, inval=0xFFFF)330 self.unused2 = parse_num(payload[47:49], False, inval=0xFFFF)331 self.rog = parse_num(payload[49:51], False, inval=0xFFFF)332 self.alarm = parse_num(payload[49:51], False, inval=0xFFFF)333 status = parse_num(payload[51:53], False)334 self.physio_monitor_worn = status & (2**15) > 0335 self.ui_button_pressed = status & (2 ** 14) > 0336 self.heart_rate_is_low_quality = status & (2 ** 13) > 0337 self.external_sensors_connected = status & (2 ** 12) > 0338 self.battery_percent = status & 127339class SummaryDataMessage(StreamingMessage):340 """Common base class for the two versions of the summary data packet."""341 def _decode_status_info(self, status_info):342 """Parse status info word and write into state."""343 self.device_worn_confidence = 1 - (status_info & 3)/3344 self.button_pressed = (status_info & 2**2) > 0345 self.not_fitted_to_garment = (status_info & 2**3) > 0346 self.heart_rate_unreliable = (status_info & 2**4) > 0347 self.respiration_rate_unreliable = (status_info & 2**5) > 0348 self.skin_temperature_unreliable = (status_info & 2**6) > 0349 self.posture_unreliable = (status_info & 2**7) > 0350 self.activity_unreliable = (status_info & 2**8) > 0351 self.hrv_unreliable = (status_info & 2**9) > 0352 self.estimated_core_temp_unreliable = (status_info & 2**10) > 0353 self.usb_power_connected = (status_info & 2**11) > 0354 self.resting_state_detected = (status_info & 2**14) > 0355 self.external_sensors_connected = (status_info & 2**15) > 0356class SummaryDataMessageV2(SummaryDataMessage):357 """A summary data packet with various slow-changing state."""358 srate = 1.0359 def __init__(self, msgid, payload, fin=MC.ETX):360 self.assert_length(payload, 71)361 super().__init__(msgid, payload, fin)362 ver = payload[9]363 assert ver == 2, """Version must be 2."""364 self.heart_rate = parse_num(payload[10:12], False, inval=0xFFFF)365 self.respiration_rate = parse_num(payload[12:14], False, inval=0xFFFF) * 0.1366 self.skin_temperature = parse_num(payload[14:16], True, inval=0x8000) * 0.1367 self.posture = parse_num(payload[16:18], True, inval=0x8000)368 self.activity = parse_num(payload[18:20], False, inval=0xFFFF) * 0.01369 self.peak_acceleration = parse_num(payload[20:22], False, inval=0xFFFF) * 0.01370 self.battery_voltage = parse_num(payload[22:24], False, inval=0xFFFF) * 0.001371 self.battery_percent = parse_num(payload[24:25], False, inval=0xFF)372 self.breathing_wave_amplitude = parse_num(payload[25:27], False, inval=0xFFFF)373 self.breathing_wave_noise = parse_num(payload[27:29], False, inval=0xFFFF)374 self.breathing_rate_confidence = parse_num(payload[29:30], False, inval=0xFF)375 self.ecg_amplitude = parse_num(payload[30:32], False, inval=0xFFFF) * 0.000001376 self.ecg_noise = parse_num(payload[32:34], False, inval=0xFFFF) * 0.000001377 self.heart_rate_confidence = parse_num(payload[34:35], False, inval=0xFF)378 self.heart_rate_variability = parse_num(payload[35:37], False, inval=0xFFFF)379 self.system_confidence = parse_num(payload[37:38], False, inval=0xFF)380 self.gsr = parse_num(payload[38:40], False, inval=0xFFFF)381 self.rog = parse_num(payload[40:42], False, inval=0)382 self.vertical_accel_min = parse_num(payload[42:44], True, inval=0x8000) * 0.01383 self.vertical_accel_peak = parse_num(payload[44:46], True, inval=0x8000) * 0.01384 self.lateral_accel_min = parse_num(payload[46:48], True, inval=0x8000) * 0.01385 self.lateral_accel_peak = parse_num(payload[48:50], True, inval=0x8000) * 0.01386 self.sagittal_accel_min = parse_num(payload[50:52], True, inval=0x8000) * 0.01387 self.sagittal_accel_peak = parse_num(payload[52:54], True, inval=0x8000) * 0.01388 self.device_internal_temp = parse_num(payload[54:56], True, inval=0x8000) * 0.1389 status_info = parse_num(payload[56:58], False, inval=0)390 self._decode_status_info(status_info)391 self.link_quality = parse_num(payload[58:59], False, inval=0xFF)*100/254392 self.rssi = parse_num(payload[59:60], False, inval=0x80)393 self.tx_power = parse_num(payload[60:61], False, inval=0x80)394 self.estimated_core_temperature = parse_num(payload[61:63], False, inval=0xFFFF) * 0.1395 self.aux_adc_chan1 = parse_num(payload[63:65], False, inval=0xFFFF)396 self.aux_adc_chan2 = parse_num(payload[65:67], False, inval=0xFFFF)397 self.aux_adc_chan3 = parse_num(payload[67:69], False, inval=0xFFFF)398 ext_status_info = parse_num(payload[69:71], False, inval=0xFFFF)399 flags_valid = 0 if (ext_status_info & 2**15) > 0 else math.nan400 self.resp_rate_low = (ext_status_info & 2 ** 0) > 0 + flags_valid401 self.resp_rate_high = (ext_status_info & 2 ** 1) > 0 + flags_valid402 self.br_amplitude_low = (ext_status_info & 2 ** 2) > 0 + flags_valid403 self.br_amplitude_high = (ext_status_info & 2 ** 3) > 0 + flags_valid404 self.br_amplitude_variance_high = (ext_status_info & 2 ** 4) > 0 + flags_valid405 self.br_signal_eval_state = (ext_status_info >> 5) & 3 + flags_valid406class SummaryDataMessageV3(SummaryDataMessage):407 """A general data packet with various slow-changing state."""408 srate = 1.0409 # unpacker function for GPS position data410 gps_pos_unpacker = make_gps_pos_unpacker()411 # unpacker for accelerometry data412 accelerometry_unpacker = make_accelerometry_unpacker()413 # noinspection PyUnresolvedReferences414 def __init__(self, msgid, payload, fin=MC.ETX):415 self.assert_length(payload, 71)416 super().__init__(msgid, payload, fin)417 ver = payload[9]418 assert ver == 3, """Version must be 3."""419 self.heart_rate = parse_num(payload[10:12], False, inval=0xFFFF)420 self.respiration_rate = parse_num(payload[12:14], False, inval=0xFFFF) * 0.1421 self.posture = parse_num(payload[14:16], True, inval=0x8000)422 self.activity = parse_num(payload[16:18], False, inval=0xFFFF) * 0.01423 self.peak_acceleration = parse_num(payload[18:20], False, inval=0xFFFF) * 0.01424 self.battery_percent = parse_num(payload[20:21], False)425 self.breathing_wave_amplitude = parse_num(payload[21:23], False, inval=0xFFFF)426 self.ecg_amplitude = parse_num(payload[23:25], False, inval=0xFFFF) * 0.000001427 self.ecg_noise = parse_num(payload[25:27], False, inval=0xFFFF) * 0.000001428 self.heart_rate_confidence = parse_num(payload[27:28], False)429 self.heart_rate_variability = parse_num(payload[28:30], False, inval=0xFFFF)430 self.rog = parse_num(payload[30:32], False, inval=0)431 status_info = parse_num(payload[32:34], False, inval=0)432 self._decode_status_info(status_info)433 self.link_quality = parse_num(payload[34:35], False, inval=0xFF)*100/254434 self.rssi = parse_num(payload[35:36], False, inval=0x80)435 self.tx_power = parse_num(payload[36:37], False, inval=0x80)436 self.estimated_core_temperature = parse_num([payload[37], 256], False, inval=0xFFFF) * 0.1437 self.__dict__.update(SummaryDataMessageV3.gps_pos_unpacker(payload[38:48]))438 self.gps_speed = parse_num(payload[48:50], False) & 0x3FFF439 self.__dict__.update(SummaryDataMessageV3.accelerometry_unpacker(payload[51:71]))440 self.avg_rate_of_force_development *= 0.01441 self.avg_step_impulse *= 0.01442 self.avg_step_period *= 0.001443 self.last_jump_flight_time *= 0.01444class WaveformMessage(StreamingMessage):445 """A message that holds a waveform."""446 def __init__(self, msgid, payload, fin, bytes_per_chunk, vals_per_packet, signed):447 """448 Create a new WaveformMessage.449 Args:450 msgid: the message id451 payload: payload bytes452 fin: the finalizer of the message453 bytes_per_chunk: number of bytes per repeating "chunk" in the payload454 (each chunk has the same bit-packing pattern)455 vals_per_packet: total values encoded in packet (needed to identify456 truncated chunks at end of payload)457 signed: whether the values are 2's complement signed (True) or458 unsigned (False), or unsigned but range-shifted ('shift')459 """460 super().__init__(msgid, payload, fin)461 # extract waveform, skipping the seq no & timestamp462 waveform = []463 vals_per_chunk = bytes_per_chunk*4//5 # here: 10-bit values464 unpacker = make_sequence_unpacker(vals_per_chunk, is_signed=signed)465 for ofs in range(9, len(payload), bytes_per_chunk):466 packed = payload[ofs:ofs+bytes_per_chunk]467 # at the end we may have a truncated packet, need to decode fewer468 # values469 if vals_per_packet < vals_per_chunk:470 unpacker = make_sequence_unpacker(vals_per_packet, is_signed=signed)471 vals = unpacker(packed)472 vals_per_packet -= vals_per_chunk473 waveform.extend(vals)474 self.waveform = waveform475class ECGWaveformMessage(WaveformMessage):476 """ECG waveform message."""477 srate = 250478 def __init__(self, msgid, payload, fin):479 self.assert_length(payload, 88)480 super().__init__(msgid, payload, fin, bytes_per_chunk=5,481 vals_per_packet=63, signed='shift')482 self.waveform = [w*0.025 for w in self.waveform] # to mV483class BreathingWaveformMessage(WaveformMessage):484 """Breathing (respiration) waveform message."""485 srate = 1000.0/56486 def __init__(self, msgid, payload, fin):487 self.assert_length(payload, 32)488 super().__init__(msgid, payload, fin, bytes_per_chunk=5,489 vals_per_packet=18, signed='shift')490class AccelerometerWaveformMessage(WaveformMessage):491 """Accelerometer waveform message."""492 srate = 50493 def __init__(self, msgid, payload, fin):494 self.assert_length(payload, 84)495 super().__init__(msgid, payload, fin, bytes_per_chunk=15,496 vals_per_packet=3*20, signed='shift')497 self.accel_x = self.waveform[::3]498 self.accel_y = self.waveform[1::3]499 self.accel_z = self.waveform[2::3]500 del self.waveform501class Accelerometer100MgWaveformMessage(WaveformMessage):502 """Accelerometer waveform message in units of 0.1g."""503 srate = 50504 def __init__(self, msgid, payload, fin):505 self.assert_length(payload, 84)506 super().__init__(msgid, payload, fin, bytes_per_chunk=15,507 vals_per_packet=3*20, signed=True)508 waveform = [w*0.1 for w in self.waveform] # to g509 self.accel_x = waveform[::3]510 self.accel_y = waveform[1::3]511 self.accel_z = waveform[2::3]512 del self.waveform513class RtoRMessage(StreamingMessage):514 srate = 1000.0 / 56515 def __init__(self, msgid, payload, fin):516 self.assert_length(payload, 45)517 super().__init__(msgid, payload, fin)518 # 16-bit values of alternating sign519 self.waveform = [parse_num(payload[ofs:ofs+2], True)520 for ofs in range(9, len(payload), 2)]521class EventMessage(StreamingMessage):522 # map of known event codes523 event_map = {524 0x0040: 'button press',525 0x0041: 'emergency button press',526 0x0080: 'battery level low',527 0x00C0: 'self test result',528 0x1000: 'ROG change',529 0x1040: 'worn status change',530 0x1080: 'HR reliability change',531 0x10C0: 'fall detected',532 0x1100: 'jump detected',533 0x1140: 'dash detected'534 }535 """A message that holds event codes."""536 def __init__(self, msgid, payload, fin):537 super().__init__(msgid, payload, fin)538 self.event_code = parse_num(payload[9:11], False)539 self.event_string = EventMessage.event_map.get(self.event_code, f'unknown:{self.event_code}')540 # event-specific data (we just store the bytes; see vendor SDK manual541 # for the interpretation)542 self.event_data = payload[11:]543def decode_message(msgid, payload=(), fin=MC.ETX):544 """Decode raw message data into a message object of appropriate type."""545 msgid = MI(msgid)546 if msgid == MI.ECGWaveformPacket:547 return ECGWaveformMessage(msgid, payload, fin)548 elif msgid == MI.Accelerometer100MgPacket:549 return Accelerometer100MgWaveformMessage(msgid, payload, fin)550 elif msgid == MI.AccelerometerPacket:551 return AccelerometerWaveformMessage(msgid, payload, fin)552 elif msgid == MI.BreathingWaveformPacket:...

Full Screen

Full Screen

code5.py

Source:code5.py Github

copy

Full Screen

...37 if key == 'Datatype':38 this.params['Datatype'] = value.strip(' \t\n')39 elif key == 'Wavelength':40 values = value.split()41 num = parse_num(values[0].strip(' '))42 unit = values[1].strip(' \t\n')43 if unit == 'nm':44 num *= 1e-745 this.params['Wavelength']=num46 else:47 raise C5_Error('Default unit for wavelength is nanometer. Please change your code5 file.')48 elif key == 'Grid spacing':49 values = value.split()50 dx = parse_num(values[0])51 dy = parse_num(values[1])52 unit = values[2].strip(' \t\n')53 if unit == 'mm':54 dx *= 0.155 dy *= 0.156 spacing = (dx,dy)57 this.params['Grid_spacing']=spacing58 else:59 raise C5_Error('Default unit for grid spacing is milimeter. Please change your code5 file.')60 elif key == 'Coordinates':61 values = value.split()62 x0 = parse_num(values[0])63 y0 = parse_num(values[1])64 z0 = parse_num(values[2])65 unit = values[3].strip(' \t\n')66 if unit == 'mm':67 x0 *= 0.168 y0 *= 0.169 z0 *= 0.170 coords = (x0,y0,z0)71 this.params['Coordinates'] = coords72 else:73 raise C5_Error('Default unit for coordinates is milimeter. Please change your code5 file.')74 elif key == 'Direction':75 values = value.split()76 x_dir = parse_num(values[0])77 y_dir = parse_num(values[1])78 z_dir = parse_num(values[2])79 dirs = (x_dir,y_dir,z_dir)80 this.params['Direction'] = dirs81 elif key == 'Array size':82 values = value.split()83 nx = parse_num(values[0])84 ny = parse_num(values[1])85 this.params['Array_size']=(nx,ny)86 else:87 raise C5_Error('Unexpected keyword occured:{0}! Please double check the compatibility of the Code5 file version and this program.'.format(key) )88 else:89 #print 'header loading finished.'90 break91 f.close()92 def read_Efield(this):93 """read the complex Efield from Code 5 file.94 Note that the result array is in the shape (ny,nx), where ny and nx are contained in this.params['Array_size'], which is read in read_header() method. And each element in the result array is a complex number.95 """96 (nx,ny)= this.params['Array_size']97 data = np.loadtxt(this.filename,comments = '!', skiprows = this.comment_line + 6)98 if(this.params['Datatype'] == 'Complex'):...

Full Screen

Full Screen

minneapolis_neighborhood_census_population_2000.py

Source:minneapolis_neighborhood_census_population_2000.py Github

copy

Full Screen

...19 'stevens_square_loring_heights': 'stevens_square',20 'stevens_square_loring_hgts': 'stevens_square',21 'u_of_m': 'university_of_minnesota',22}23def parse_num(s):24 try:25 return int(s)26 except ValueError:27 try:28 return int(float(s))29 except ValueError:30 return 031def string_to_key(str):32 str = re.sub('[^0-9a-zA-Z]+', '_', str)33 str = str.replace('__', '_')34 str = str.replace('__', '_')35 return str.lower()36def get_neighborhood_key(str):37 key = string_to_key(str)38 if key in n_translate_keys:39 key = n_translate_keys[key]40 41 return key42# Start scraping43html = scraperwiki.scrape(url)44root = lxml.html.fromstring(html)45rowCount = 046for tr in root.cssselect('#maincontent table tr'):47 if rowCount > 1:48 tds = tr.cssselect('td')49 data = {50 'neighborhood_key' : get_neighborhood_key(tds[0].text_content()),51 'neighborhood' : tds[0].text_content(),52 'total' : parse_num(tds[7].text_content()),53 'white' : parse_num(tds[1].text_content()),54 'black' : parse_num(tds[2].text_content()),55 'native' : parse_num(tds[3].text_content()),56 'asian' : parse_num(tds[4].text_content()),57 'other' : parse_num(tds[5].text_content()),58 'two' : parse_num(tds[6].text_content()),59 'dispanic' : parse_num(tds[8].text_content()),60 }61 #pprint.pprint(data)62 scraperwiki.sqlite.save(unique_keys=['neighborhood_key'], data=data)63 rowCount = rowCount + 1;import scraperwiki64import lxml.html65import pprint66import re67url = 'http://www.ci.minneapolis.mn.us/census/2000/census_2000-race-and-ethnicity-by-neighborhood'68n_translate_keys = {69 'camden_ind_area': 'camden_industrial',70 'columbia': 'columbia_park',71 'east_phillips': 'phillips_east',72 'fowell': 'folwell',73 'humboldt_ind_area': 'humboldt_industrial_area',74 'midtown_phillips': 'phillips_midtown',75 'nicollet_island_east_bank': 'nicollet_island',76 'north_river_ind_area': 'north_river_industrial_area',77 'northrop': 'northrup',78 'prospect_park_east_river_rd': 'prospect_park',79 'prospect_park_east_river_road': 'prospect_park',80 'steven_s_square_loring_heights': 'stevens_square',81 'stevens_square_loring_heights': 'stevens_square',82 'stevens_square_loring_hgts': 'stevens_square',83 'u_of_m': 'university_of_minnesota',84}85def parse_num(s):86 try:87 return int(s)88 except ValueError:89 try:90 return int(float(s))91 except ValueError:92 return 093def string_to_key(str):94 str = re.sub('[^0-9a-zA-Z]+', '_', str)95 str = str.replace('__', '_')96 str = str.replace('__', '_')97 return str.lower()98def get_neighborhood_key(str):99 key = string_to_key(str)100 if key in n_translate_keys:101 key = n_translate_keys[key]102 103 return key104# Start scraping105html = scraperwiki.scrape(url)106root = lxml.html.fromstring(html)107rowCount = 0108for tr in root.cssselect('#maincontent table tr'):109 if rowCount > 1:110 tds = tr.cssselect('td')111 data = {112 'neighborhood_key' : get_neighborhood_key(tds[0].text_content()),113 'neighborhood' : tds[0].text_content(),114 'total' : parse_num(tds[7].text_content()),115 'white' : parse_num(tds[1].text_content()),116 'black' : parse_num(tds[2].text_content()),117 'native' : parse_num(tds[3].text_content()),118 'asian' : parse_num(tds[4].text_content()),119 'other' : parse_num(tds[5].text_content()),120 'two' : parse_num(tds[6].text_content()),121 'dispanic' : parse_num(tds[8].text_content()),122 }123 #pprint.pprint(data)124 scraperwiki.sqlite.save(unique_keys=['neighborhood_key'], data=data)...

Full Screen

Full Screen

Parser.py

Source:Parser.py Github

copy

Full Screen

...25 jdata = json.loads(mydivs[0]["data-initial-data"])26 objects.append(jdata)27 self.df = pd.DataFrame.from_dict([t['listing'] for t in objects])28 @staticmethod29 def parse_num(x, n_remove_suff=0):30 if x == None:31 return None32 t = "".join([v for v in x if v.isnumeric() or v == ","])33 t = t.replace(",", ".")34 if n_remove_suff:35 t = t[:-n_remove_suff]36 return float(t)37 @staticmethod38 def contains_num(s):39 return any(filter(lambda x: x.isnumeric(), s))40 @staticmethod41 def floor(x):42 try:43 x = x.lower()44 if "vån " in x:45 last = x.split("vån")[-1]46 return Parser.parse_num(last)47 if x[-2:] == "tr":48 return Parser.parse_num(x[-4:-2])49 except:50 return 051 return 052 def clean_data(self):53 df = self.df54 df_parsed = pd.DataFrame()55 df_parsed["price_per_area"] = df.price_per_area.apply(lambda x: Parser.parse_num(x, 1))56 df_parsed["rooms"] = df.rooms.apply(Parser.parse_num)57 df_parsed["fee"] = df.fee.apply(Parser.parse_num)58 df_parsed["living_space"] = df.living_space.apply(lambda x: Parser.parse_num(x, 1))59 df_parsed["supplemental_area"] = df.supplemental_area.apply(lambda x: Parser.parse_num(x, 1))60 df_parsed["price"] = df.price.apply(Parser.parse_num)61 df_parsed["asked_price"] = df.asked_price.apply(Parser.parse_num)62 df_parsed["land_area"] = df.land_area.apply(lambda x: Parser.parse_num(x, 1))63 df_parsed["longitude"] = df.coordinate.apply(lambda x: x[0])64 df_parsed["latitude"] = df.coordinate.apply(lambda x: x[1])65 df_parsed["typeSummary"] = df.typeSummary66 df_parsed["year"] = df.sale_date.apply(lambda x: x[5:].split("-")[0])67 df_parsed["month"] = df.sale_date.apply(lambda x: x[5:].split("-")[1])68 df_parsed["day"] = df.sale_date.apply(lambda x: x[5:].split("-")[2])69 df_parsed["floor"] = df.address.apply(Parser.floor)...

Full Screen

Full Screen

Pytest Tutorial

Looking for an in-depth tutorial around pytest? LambdaTest covers the detailed pytest tutorial that has everything related to the pytest, from setting up the pytest framework to automation testing. Delve deeper into pytest testing by exploring advanced use cases like parallel testing, pytest fixtures, parameterization, executing multiple test cases from a single file, and more.

Chapters

  1. What is pytest
  2. Pytest installation: Want to start pytest from scratch? See how to install and configure pytest for Python automation testing.
  3. Run first test with pytest framework: Follow this step-by-step tutorial to write and run your first pytest script.
  4. Parallel testing with pytest: A hands-on guide to parallel testing with pytest to improve the scalability of your test automation.
  5. Generate pytest reports: Reports make it easier to understand the results of pytest-based test runs. Learn how to generate pytest reports.
  6. Pytest Parameterized tests: Create and run your pytest scripts while avoiding code duplication and increasing test coverage with parameterization.
  7. Pytest Fixtures: Check out how to implement pytest fixtures for your end-to-end testing needs.
  8. Execute Multiple Test Cases: Explore different scenarios for running multiple test cases in pytest from a single file.
  9. Stop Test Suite after N Test Failures: See how to stop your test suite after n test failures in pytest using the @pytest.mark.incremental decorator and maxfail command-line option.

YouTube

Skim our below pytest tutorial playlist to get started with automation testing using the pytest framework.

https://www.youtube.com/playlist?list=PLZMWkkQEwOPlcGgDmHl8KkXKeLF83XlrP

Run Pytest automation tests on LambdaTest cloud grid

Perform automation testing on 3000+ real desktop and mobile devices online.

... View sample repo... View DocumentationSign up Free
›
_

Try LambdaTest Now !!

Get 100 minutes of automation test minutes FREE!!

...
Sign up with GoogleSign up with Email
Next-Gen App & Browser Testing Cloud

Was this article helpful?

Helpful

NotHelpful