Best Python code snippet using hypothesis
queues.py
Source:queues.py
1"""2-------------------------------------------------------------------------3Library of RTL queues4-------------------------------------------------------------------------5Author : Yanghui Ou6 Date : Mar 23, 20197"""8from pymtl3 import *9from pymtl3.stdlib.basic_rtl import Mux, RegisterFile10from .enq_deq_ifcs import DeqIfcRTL, EnqIfcRTL11#-------------------------------------------------------------------------12# Dpath and Ctrl for NormalQueueRTL13#-------------------------------------------------------------------------14class NormalQueueDpathRTL( Component ):15 def construct( s, EntryType, num_entries=2 ):16 # Interface17 s.enq_msg = InPort( EntryType )18 s.deq_ret = OutPort( EntryType )19 s.wen = InPort()20 s.waddr = InPort( clog2( num_entries ) )21 s.raddr = InPort( clog2( num_entries ) )22 # Component23 s.queue = m = RegisterFile( EntryType, num_entries )24 m.raddr[0] //= s.raddr25 m.rdata[0] //= s.deq_ret26 m.wen[0] //= s.wen27 m.waddr[0] //= s.waddr28 m.wdata[0] //= s.enq_msg29class NormalQueueCtrlRTL( Component ):30 def construct( s, num_entries=2 ):31 # Constants32 addr_nbits = clog2 ( num_entries )33 count_nbits = clog2 ( num_entries+1 )34 PtrType = mk_bits ( addr_nbits )35 CountType = mk_bits ( count_nbits )36 s.last_idx = PtrType ( num_entries-1 )37 s.num_entries = CountType( num_entries )38 # Interface39 s.enq_en = InPort ()40 s.enq_rdy = OutPort()41 s.deq_en = InPort ()42 s.deq_rdy = OutPort()43 s.count = OutPort( CountType )44 s.wen = OutPort()45 s.waddr = OutPort( PtrType )46 s.raddr = OutPort( PtrType )47 # Registers48 s.head = Wire( PtrType )49 s.tail = Wire( PtrType )50 # Wires51 s.enq_xfer = Wire( Bits1 )52 s.deq_xfer = Wire( Bits1 )53 # Connections54 connect( s.wen, s.enq_xfer )55 connect( s.waddr, s.tail )56 connect( s.raddr, s.head )57 s.enq_rdy //= lambda: ~s.reset & ( s.count < s.num_entries )58 s.deq_rdy //= lambda: ~s.reset & ( s.count > CountType(0) )59 s.enq_xfer //= lambda: s.enq_en & s.enq_rdy60 s.deq_xfer //= lambda: s.deq_en & s.deq_rdy61 @update_ff62 def up_reg():63 if s.reset:64 s.head <<= PtrType(0)65 s.tail <<= PtrType(0)66 s.count <<= CountType(0)67 else:68 if s.deq_xfer:69 s.head <<= s.head + PtrType(1) if s.head < s.last_idx else PtrType(0)70 if s.enq_xfer:71 s.tail <<= s.tail + PtrType(1) if s.tail < s.last_idx else PtrType(0)72 if s.enq_xfer & ~s.deq_xfer:73 s.count <<= s.count + CountType(1)74 if ~s.enq_xfer & s.deq_xfer:75 s.count <<= s.count - CountType(1)76#-------------------------------------------------------------------------77# NormalQueueRTL78#-------------------------------------------------------------------------79class NormalQueueRTL( Component ):80 def construct( s, EntryType, num_entries=2 ):81 # Interface82 s.enq = EnqIfcRTL( EntryType )83 s.deq = DeqIfcRTL( EntryType )84 s.count = OutPort( mk_bits( clog2( num_entries+1 ) ) )85 # Components86 assert num_entries > 087 if num_entries == 1:88 s.q = NormalQueue1EntryRTL( EntryType )89 connect( s.enq, s.q.enq )90 connect( s.deq, s.q.deq )91 connect( s.count, s.q.count )92 else:93 s.ctrl = NormalQueueCtrlRTL ( num_entries )94 s.dpath = NormalQueueDpathRTL( EntryType, num_entries )95 # Connect ctrl to data path96 connect( s.ctrl.wen, s.dpath.wen )97 connect( s.ctrl.waddr, s.dpath.waddr )98 connect( s.ctrl.raddr, s.dpath.raddr )99 # Connect to interface100 connect( s.enq.en, s.ctrl.enq_en )101 connect( s.enq.rdy, s.ctrl.enq_rdy )102 connect( s.deq.en, s.ctrl.deq_en )103 connect( s.deq.rdy, s.ctrl.deq_rdy )104 connect( s.count, s.ctrl.count )105 connect( s.enq.msg, s.dpath.enq_msg )106 connect( s.deq.ret, s.dpath.deq_ret )107 # Line trace108 def line_trace( s ):109 return f"{s.enq}({s.count}){s.deq}"110#-------------------------------------------------------------------------111# Ctrl for PipeQueue112#-------------------------------------------------------------------------113class PipeQueueCtrlRTL( Component ):114 def construct( s, num_entries=2 ):115 # Constants116 addr_nbits = clog2 ( num_entries )117 count_nbits = clog2 ( num_entries+1 )118 PtrType = mk_bits ( addr_nbits )119 CountType = mk_bits ( count_nbits )120 s.last_idx = PtrType ( num_entries-1 )121 s.num_entries = CountType( num_entries )122 # Interface123 s.enq_en = InPort ( Bits1 )124 s.enq_rdy = OutPort( Bits1 )125 s.deq_en = InPort ( Bits1 )126 s.deq_rdy = OutPort( Bits1 )127 s.count = OutPort( CountType )128 s.wen = OutPort( Bits1 )129 s.waddr = OutPort( PtrType )130 s.raddr = OutPort( PtrType )131 # Registers132 s.head = Wire( PtrType )133 s.tail = Wire( PtrType )134 # Wires135 s.enq_xfer = Wire( Bits1 )136 s.deq_xfer = Wire( Bits1 )137 # Connections138 connect( s.wen, s.enq_xfer )139 connect( s.waddr, s.tail )140 connect( s.raddr, s.head )141 s.deq_rdy //= lambda: ~s.reset & ( s.count > CountType(0) )142 s.enq_rdy //= lambda: ~s.reset & ( ( s.count < s.num_entries ) | s.deq_en )143 s.enq_xfer //= lambda: s.enq_en & s.enq_rdy144 s.deq_xfer //= lambda: s.deq_en & s.deq_rdy145 @update_ff146 def up_reg():147 if s.reset:148 s.head <<= PtrType(0)149 s.tail <<= PtrType(0)150 s.count <<= CountType(0)151 else:152 if s.deq_xfer:153 s.head <<= s.head + PtrType(1) if s.head < s.last_idx else PtrType(0)154 if s.enq_xfer:155 s.tail <<= s.tail + PtrType(1) if s.tail < s.last_idx else PtrType(0)156 if s.enq_xfer & ~s.deq_xfer:157 s.count <<= s.count + CountType(1)158 if ~s.enq_xfer & s.deq_xfer:159 s.count <<= s.count - CountType(1)160#-------------------------------------------------------------------------161# PipeQueueRTL162#-------------------------------------------------------------------------163class PipeQueueRTL( Component ):164 def construct( s, EntryType, num_entries=2 ):165 # Interface166 s.enq = EnqIfcRTL( EntryType )167 s.deq = DeqIfcRTL( EntryType )168 s.count = OutPort( mk_bits( clog2( num_entries+1 ) ) )169 # Components170 assert num_entries > 0171 if num_entries == 1:172 s.q = PipeQueue1EntryRTL( EntryType )173 connect( s.enq, s.q.enq )174 connect( s.deq, s.q.deq )175 connect( s.count, s.q.count )176 else:177 s.ctrl = PipeQueueCtrlRTL ( num_entries )178 s.dpath = NormalQueueDpathRTL( EntryType, num_entries )179 # Connect ctrl to data path180 connect( s.ctrl.wen, s.dpath.wen )181 connect( s.ctrl.waddr, s.dpath.waddr )182 connect( s.ctrl.raddr, s.dpath.raddr )183 # Connect to interface184 connect( s.enq.en, s.ctrl.enq_en )185 connect( s.enq.rdy, s.ctrl.enq_rdy )186 connect( s.deq.en, s.ctrl.deq_en )187 connect( s.deq.rdy, s.ctrl.deq_rdy )188 connect( s.count, s.ctrl.count )189 connect( s.enq.msg, s.dpath.enq_msg )190 connect( s.deq.ret, s.dpath.deq_ret )191 # Line trace192 def line_trace( s ):193 return "{}({}){}".format( s.enq, s.count, s.deq )194#-------------------------------------------------------------------------195# Ctrl and Dpath for BypassQueue196#-------------------------------------------------------------------------197class BypassQueueDpathRTL( Component ):198 def construct( s, EntryType, num_entries=2 ):199 # Interface200 s.enq_msg = InPort( EntryType )201 s.deq_ret = OutPort( EntryType )202 s.wen = InPort( Bits1 )203 s.waddr = InPort( mk_bits( clog2( num_entries ) ) )204 s.raddr = InPort( mk_bits( clog2( num_entries ) ) )205 s.mux_sel = InPort( Bits1 )206 # Component207 s.queue = m = RegisterFile( EntryType, num_entries )208 m.raddr[0] //= s.raddr209 m.wen[0] //= s.wen210 m.waddr[0] //= s.waddr211 m.wdata[0] //= s.enq_msg212 s.mux = m = Mux( EntryType, 2 )213 m.sel //= s.mux_sel214 m.in_[0] //= s.queue.rdata[0]215 m.in_[1] //= s.enq_msg216 m.out //= s.deq_ret217class BypassQueueCtrlRTL( Component ):218 def construct( s, num_entries=2 ):219 # Constants220 addr_nbits = clog2 ( num_entries )221 count_nbits = clog2 ( num_entries+1 )222 PtrType = mk_bits ( addr_nbits )223 CountType = mk_bits ( count_nbits )224 s.last_idx = PtrType ( num_entries-1 )225 s.num_entries = CountType( num_entries )226 # Interface227 s.enq_en = InPort ( Bits1 )228 s.enq_rdy = OutPort( Bits1 )229 s.deq_en = InPort ( Bits1 )230 s.deq_rdy = OutPort( Bits1 )231 s.count = OutPort( CountType )232 s.wen = OutPort( Bits1 )233 s.waddr = OutPort( PtrType )234 s.raddr = OutPort( PtrType )235 s.mux_sel = OutPort( Bits1 )236 # Registers237 s.head = Wire( PtrType )238 s.tail = Wire( PtrType )239 # Wires240 s.enq_xfer = Wire( Bits1 )241 s.deq_xfer = Wire( Bits1 )242 # Connections243 connect( s.wen, s.enq_xfer )244 connect( s.waddr, s.tail )245 connect( s.raddr, s.head )246 s.enq_rdy //= lambda: ~s.reset & ( s.count < s.num_entries )247 s.deq_rdy //= lambda: ~s.reset & ( (s.count > CountType(0) ) | s.enq_en )248 s.mux_sel //= lambda: s.count == CountType(0)249 s.enq_xfer //= lambda: s.enq_en & s.enq_rdy250 s.deq_xfer //= lambda: s.deq_en & s.deq_rdy251 @update_ff252 def up_reg():253 if s.reset:254 s.head <<= PtrType(0)255 s.tail <<= PtrType(0)256 s.count <<= CountType(0)257 else:258 if s.deq_xfer:259 s.head <<= s.head + PtrType(1) if s.head < s.last_idx else PtrType(0)260 if s.enq_xfer:261 s.tail <<= s.tail + PtrType(1) if s.tail < s.last_idx else PtrType(0)262 if s.enq_xfer & ~s.deq_xfer:263 s.count <<= s.count + CountType(1)264 if ~s.enq_xfer & s.deq_xfer:265 s.count <<= s.count - CountType(1)266#-------------------------------------------------------------------------267# BypassQueueRTL268#-------------------------------------------------------------------------269class BypassQueueRTL( Component ):270 def construct( s, EntryType, num_entries=2 ):271 # Interface272 s.enq = EnqIfcRTL( EntryType )273 s.deq = DeqIfcRTL( EntryType )274 s.count = OutPort( mk_bits( clog2( num_entries+1 ) ) )275 # Components276 assert num_entries > 0277 if num_entries == 1:278 s.q = BypassQueue1EntryRTL( EntryType )279 connect( s.enq, s.q.enq )280 connect( s.deq, s.q.deq )281 connect( s.count, s.q.count )282 else:283 s.ctrl = BypassQueueCtrlRTL ( num_entries )284 s.dpath = BypassQueueDpathRTL( EntryType, num_entries )285 # Connect ctrl to data path286 connect( s.ctrl.wen, s.dpath.wen )287 connect( s.ctrl.waddr, s.dpath.waddr )288 connect( s.ctrl.raddr, s.dpath.raddr )289 connect( s.ctrl.mux_sel, s.dpath.mux_sel )290 # Connect to interface291 connect( s.enq.en, s.ctrl.enq_en )292 connect( s.enq.rdy, s.ctrl.enq_rdy )293 connect( s.deq.en, s.ctrl.deq_en )294 connect( s.deq.rdy, s.ctrl.deq_rdy )295 connect( s.count, s.ctrl.count )296 connect( s.enq.msg, s.dpath.enq_msg )297 connect( s.deq.ret, s.dpath.deq_ret )298 # Line trace299 def line_trace( s ):300 return f"{s.enq}({s.count}){s.deq}"301#-------------------------------------------------------------------------302# NormalQueue1EntryRTL303#-------------------------------------------------------------------------304class NormalQueue1EntryRTL( Component ):305 def construct( s, EntryType ):306 # Interface307 s.enq = EnqIfcRTL( EntryType )308 s.deq = DeqIfcRTL( EntryType )309 s.count = OutPort ( Bits1 )310 # Components311 s.entry = Wire( EntryType )312 s.full = Wire( Bits1 )313 # Logic314 s.count //= s.full315 s.deq.ret //= s.entry316 s.enq.rdy //= lambda: ~s.reset & ~s.full317 s.deq.rdy //= lambda: ~s.reset & s.full318 @update_ff319 def ff_normal1():320 s.full <<= ~s.reset & ( ~s.deq.en & (s.enq.en | s.full) )321 if s.enq.en:322 s.entry <<= s.enq.msg323 def line_trace( s ):324 return f"{s.enq}({s.full}){s.deq}"325#-------------------------------------------------------------------------326# PipeQueue1EntryRTL327#-------------------------------------------------------------------------328class PipeQueue1EntryRTL( Component ):329 def construct( s, EntryType ):330 # Interface331 s.enq = EnqIfcRTL( EntryType )332 s.deq = DeqIfcRTL( EntryType )333 s.count = OutPort ( Bits1 )334 # Components335 s.entry = Wire( EntryType )336 s.full = Wire( Bits1 )337 # Logic338 s.count //= s.full339 s.deq.ret //= s.entry340 s.enq.rdy //= lambda: ~s.reset & ( ~s.full | s.deq.en )341 s.deq.rdy //= lambda: s.full & ~s.reset342 @update_ff343 def ff_pipe1():344 s.full <<= ~s.reset & ( s.enq.en | s.full & ~s.deq.en )345 if s.enq.en:346 s.entry <<= s.enq.msg347 def line_trace( s ):348 return f"{s.enq}({s.full}){s.deq}"349#-------------------------------------------------------------------------350# BypassQueue1EntryRTL351#-------------------------------------------------------------------------352class BypassQueue1EntryRTL( Component ):353 def construct( s, EntryType ):354 # Interface355 s.enq = EnqIfcRTL( EntryType )356 s.deq = DeqIfcRTL( EntryType )357 s.count = OutPort ( Bits1 )358 # Components359 s.entry = Wire( EntryType )360 s.full = Wire( Bits1 )361 s.bypass_mux = m = Mux( EntryType, 2 )362 m.in_[0] //= s.enq.msg363 m.in_[1] //= s.entry364 m.out //= s.deq.ret365 m.sel //= s.full366 # Logic367 s.count //= s.full368 s.enq.rdy //= lambda: ~s.reset & ~s.full369 s.deq.rdy //= lambda: ~s.reset & ( s.full | s.enq.en )370 @update_ff371 def ff_bypass1():372 s.full <<= ~s.reset & ( ~s.deq.en & (s.enq.en | s.full) )373 if s.enq.en & ~s.deq.en:374 s.entry <<= s.enq.msg375 def line_trace( s ):...
utils.py
Source:utils.py
1from datetime import datetime2import re3import os4import sqlite35DATABASE_PATH = os.path.join(os.path.dirname(os.path.abspath(__file__)), "courses_db.sqlite")6connection = sqlite3.connect(DATABASE_PATH)7courses_regex = {8 "python": re.compile(r"\bpython\b|\bprogramming\b|\bcoding\b"),9 "machine-learning": re.compile(r"\bmachine learn(ing)?\b|\bai\b|\bartificial intellegence\b|\bml\b"),10 "deep-learning": re.compile(r"deep learning|\bdl\b|neural networks?|\bann\b"),11 "data-science": re.compile(r"data[\s-]?[science|scientist|analysis|analyst|engineer|analytics]?"),12 "natural-language-processing": re.compile(r"\bnlp\b|natural? language processing|\brnn\b|\blstm\b|text [analysis|processing|mining]"),13 "computer-vision": re.compile(r"\bcv\b|\bcnn\b|computer vision|[image|face] recognition|image processing"),14}15# build a bank request to make things fun16response_bank = {17 "greetings": ["hello there!, how can i help?", "Hi i am cyborg how can i help?", "Hello friend! how can i help you?"],18 "thanks": ["Feel free to contact us anytime", "I am happy to help", "More than happy to help"],19 "availability_enq": ["we have {course} course starts at {start_date} for only {price}$ !!", "would like {course} course on {start_date}?"],20 "cost_enq": ["the course {course} will cost only {price}$", "the cost for {course} is only {price}$"],21 "duration_enq": ["The course {course} is {duration} hours from {start_date}", "{course} course duration is {duration} hours"],22 "reserve_enq": ["Sure, you would like to confirm your reservation in {course} course, on {start_date} session?", "Would you please confirm your {course} course reservation on {start_date}?"],23 "start_enq": ["The {course} course will start on {start_date}", "the next session for {course} course is {start_date}"],24 "unknown": ["i didn't get that sorry", "kindly would you try again, i didn't get this one", "sorry could you try this again"],25 "registered": ["you have registered to {course} course!, see you on {start_date} !!"]26}27examples = [28 # greetings29 "hello there",30 "good morning",31 "welcome",32 "hi!, how are you",33 "Hi there !!",34 35 # thanks36 "thanks",37 "that's it",38 "goodbye",39 "bye",40 "okay",41 42 # cost examples43 "How much for the python course?",44 "what is the cost for C++ course?",45 "can you tell me the price of java course",46 "what do i pay for data science track?",47 "how much will i pay for this?",48 49 # duration examples50 "how much time will the deep learning track be?",51 "how much will it take?",52 "can you tell me the duration of the whole specialization?",53 "what is the total duration of the python course?",54 "and the duration?",55 56 # availability examples57 "do you have courses on machine learning?",58 "i want to know if you have embedded systems course",59 "do you teach data science?",60 "i want to learn to code",61 "do you have android courses?",62 63 # start date examples64 "when will the course start?",65 "when can i start the next machine learning course",66 "can you tell me the starting date of the python course",67 "what is the start date for the upcoming session?",68 "what is the soonest session",69 70 # reservation71 "i want to reserve the python course",72 "i want to sign up in the data science track",73 "let me take this course",74 "sign me up in the machine learning track",75 "register me in the java course",76]77labels = [78 # greetings79 "greetings",80 "greetings",81 "greetings",82 "greetings",83 "greetings",84 85 # thanks86 "thanks",87 "thanks",88 "thanks",89 "thanks",90 "thanks",91 92 # cost labels93 "cost_enq",94 "cost_enq",95 "cost_enq",96 "cost_enq",97 "cost_enq",98 99 # duration labels100 "duration_enq",101 "duration_enq",102 "duration_enq",103 "duration_enq",104 "duration_enq",105 106 107 # availability labels108 "availability_enq",109 "availability_enq",110 "availability_enq",111 "availability_enq",112 "availability_enq",113 114 # start date 115 "start_enq",116 "start_enq",117 "start_enq",118 "start_enq",119 "start_enq",120 121 # reservation labels122 "reserve_enq",123 "reserve_enq",124 "reserve_enq",125 "reserve_enq",126 "reserve_enq",127]128 129#knowldge back ~ our database130KB = {131 "python":132 {133 "price": 100,134 "duration": 40,135 "start-date": datetime(2020, 5, 5),136 "registered": 0137 },138 "machine-learning":139 {140 "price": 200,141 "duration": 60,142 "start-date": datetime(2020, 5, 10),143 "registered": 0144 },145 "deep-learning":146 {147 "price": 150,148 "duration": 80,149 "start-date": datetime(2020, 5, 3),150 "registered": 0151 },152 "data-science":153 {154 "price": 200,155 "duration": 100,156 "start-date": datetime(2020, 5, 17),157 "registered": 0158 },159 "natural-language-processing":160 {161 "price": 250,162 "duration": 80,163 "start-date": datetime(2020, 6, 20),164 "registered": 0165 },166 "computer-vision":167 {168 "price": 200,169 "duration": 70,170 "start-date": datetime(2020, 7, 11),171 "registered": 0172 }173}174def load_data_to_db():175 connection.execute("delete from courses")176 for course_name in KB.keys():177 price = KB[course_name]['price']178 duration = KB[course_name]['duration']179 start_date = KB[course_name]['start-date'].strftime("%d-%m-%Y")180 register = KB[course_name]['registered']181 connection.execute(f"insert into courses (name, price, duration, start_date, registered)\182 values ('{course_name}', '{price}', '{duration}', '{start_date}', '{register}')")183 connection.commit()184 print("all done !")185def get_course_data(course_name):186 res = connection.execute(f"select * from courses where name = '{course_name}'").fetchall()[0]187 course_info = {188 "name": res[0],189 "price": res[2],190 "duration": res[3],191 "start_date": res[4],192 "registered": res[5]193 }194 return course_info195def register_user(course_name):196 connection.execute(f"update courses set registered=registered+1 where name = '{course_name}'")197 connection.commit()198 199if __name__ == "__main__":...
QueueArray_test.py
Source:QueueArray_test.py
...3 s = QueueArray()4 assert s.is_empty()5def test_enq_empty():6 s = QueueArray()7 s.enq(5)8 assert not s.is_empty()9def test_enq_tail():10 s = QueueArray()11 s.enq(5)12 assert s.get_tail() == 513def test_enq_front():14 s = QueueArray()15 s.enq(5)16 assert s.get_front() == 517def test_deq():18 s = QueueArray()19 s.enq(5)20 assert s.deq() == 521def test_deq_empty():22 s = QueueArray()23 s.enq(5)24 s.deq()25 assert s.is_empty()26def test_enq_many_empty():27 s = QueueArray()28 s.enq(15)29 s.enq(16)30 s.enq(17)31 assert not s.is_empty()32def test_enq_many_tail():33 s = QueueArray()34 s.enq(15)35 s.enq(16)36 s.enq(17)37 assert s.get_tail() == 1738def test_enq_many_front():39 s = QueueArray()40 s.enq(15)41 s.enq(16)42 s.enq(17)43 assert s.get_front() == 1544def test_enq_many_deq():45 s = QueueArray()46 s.enq(15)47 s.enq(16)48 s.enq(17)49 assert s.deq() == 1550def test_deq_empty():51 s = QueueArray()52 assert s.deq() is None53def test_clear_empty():54 s = QueueArray()55 s.enq(5)56 s.clear()57 assert s.is_empty()58def test_enq_deq_many():59 s = QueueArray()60 le = []61 la = []62 for i in range(20,201,10):63 s.enq(i)64 le.append(i)65 while not s.is_empty():66 la.append(s.deq())...
Learn to execute automation testing from scratch with LambdaTest Learning Hub. Right from setting up the prerequisites to run your first automation test, to following best practices and diving deeper into advanced test scenarios. LambdaTest Learning Hubs compile a list of step-by-step guides to help you be proficient with different test automation frameworks i.e. Selenium, Cypress, TestNG etc.
You could also refer to video tutorials over LambdaTest YouTube channel to get step by step demonstration from industry experts.
Get 100 minutes of automation test minutes FREE!!