eshangj/stackoverflow_q_and_a_sample huggingface dataset:
Question: I am trying to implement a predecessor constraint similar to the Job Scheduling example given in java. But I struggle with the ordering constraint definition to consider predecessors. I have defined my time slots simply as ordered integers : @dataclass class Timeslot: slot : int  And my operations as planning entities with few boolean operations and predecessors as a list of other operations id : @planning_entity @dataclass class Operation: id: Annotated[int, PlanningId] name: str predecessors: list[int] timeslot: Annotated[Timeslot, PlanningVariable] = field(default=None) def isPred(self,other): return other.id in self.predecessors def isAfter(self,other): return self.timeslot.slot < other.timeslot.slot  And then my constraint as : def precondition_conflict(constraint_factory: ConstraintFactory) -> Constraint: # Respect order constraints return ( constraint_factory.for_each_unique_pair(Operation) .filter(lambda op1, op2 : op1.isPred(op2)) .filter(lambda op1, op2 : op1.isAfter(op2)) .penalize(HardSoftScore.ONE_HARD) .as_constraint("Order conflict") )  Then I instanciate my problem :  operations.append(Operation(1,"OP N°1 : Bake the cake",[3])) operations.append(Operation(2,"OP N°2 : Enjoy your meal",[4])) operations.append(Operation(3,"OP N°3 : Mix flour, eggs and whatever they say in the recipe",[])) operations.append(Operation(4,"OP N°4 : take it out of the oven",[1]))  But the solver keeps giving me solutions where the operations order is incorrect, with 0 hard constraint violated. For instance : INFO:timefold.solver:Solving ended: time spent (30056), best score (0hard/0soft), move evaluation speed (219895/sec), phase total (2), environment mode (PHASE_ASSERT), move thread count (NONE). INFO:app:+------------------+------------------+ INFO:app:|0 |OP N°4 : take it out of the oven| INFO:app:+------------------+------------------+ INFO:app:|1 |OP N°2 : Enjoy your meal| INFO:app:+------------------+------------------+ INFO:app:|2 |OP N°3 : Mix flour, eggs and whatever they say in the recipe| INFO:app:+------------------+------------------+ INFO:app:|3 |OP N°1 : Bake the cake| INFO:app:+------------------+------------------+  Where obviously I expected 3 -> 1 -> 4 -> 2 Where did I do it wrong? SOLUTION (thanks to Lukáš Petrovický): def precondition_conflict(constraint_factory: ConstraintFactory) -> Constraint: # Respect order constraints return ( constraint_factory.for_each(Operation) .join(Operation) .filter(lambda op1, op2 : op1.isPred(op2)) .filter(lambda op1, op2 : op1.isAfter(op2)) .penalize(HardSoftScore.ONE_HARD) .as_constraint("Order conflict") ) 

Answer: Unique pairs are tricky. In this case, you probably want to avoid using them, and instead go for a standard join. Consider three operations: A, B, C. Unique pairs will give you A+B, A+C and B+C. It assumes that whatever is true for A+B, it is also true for B+A and therefore processing both would be redundant. But this is not true in your case. A standard join would fix it, enumerating all possible pairs. Or a better filter, which checks the condition from both sides.
Question: Shareplum is unable to retrieve entries from a Sharepoint list using DateTime variables (for example, to find entries that were modified after a certain date). Code: from requests_negotiate_sspi import HttpNegotiateAuth from shareplum import Site import datetime site = Site("sharepoint.com", version=Version.v2016, auth=HttpNegotiateAuth()) date = datetime.datetime(2025, 4, 8, 0, 0, 0) query = {'Where' : [('Gt', 'Modified', date)]} sp_list = site.List('Sharepoint List') data = sp_list.GetListItems('All Items', query=query)  If a datetime object is provided, the following error occurs: TypeError : Argument must be bytes or unicode, got 'datetime'  If a string/integer is provided, the following error occurs instead: Failed to download list items Shareplum HTTP Post Failed : 500 Server Error  As mentioned in these issues:  https://github.com/jasonrollins/shareplum/issues/156 https://github.com/jasonrollins/shareplum/issues/186 

Answer: You can manually craft the CAML query XML, rather than using SharePlum’s dictionary-style shorthand. Here’s how to do it: from requests_negotiate_sspi import HttpNegotiateAuth from shareplum import Site from shareplum import Office365 from shareplum.site import Version import datetime # SharePoint credentials site_url = "https://yourcompany.sharepoint.com/sites/yoursite" list_name = "Your List Name" auth = HttpNegotiateAuth() site = Site(site_url, auth=auth, version=Version.v2016) sp_list = site.List(list_name) # Construct CAML query dt = datetime.datetime(2025, 4, 8, 0, 0, 0) date_string = dt.strftime('%Y-%m-%dT%H:%M:%SZ') # Format must be UTC ISO8601 caml_query = f""" <View> <Query> <Where> <Gt> <FieldRef Name='Modified' /> <Value Type='DateTime'>{date_string}</Value> </Gt> </Where> </Query> </View> """ # Retrieve items items = sp_list.GetListItems(view_name=None, query=caml_query) print(items)
Question: I'm working on a todo app for a class project and the /add_todo route returns a 415 error code whenever I try to add a todo. @app.route("/add-todo", methods=["GET", "POST"]) def add_todo(): title = request.get_json().get("title") db.session.add(Todo(title=title)) db.session.commit() return '', 204  <form action="/add-todo" method="post" hx-boost="true"> <input type="text" id="todo_bar"> <button id="submit">Add</button> </form>  I've tried everything from changing what the route returns to adding the methods array when declaring the route, yet nothing seems to make a difference.

Answer: The data you're sending is form-encoded (by the browser). You can retrieve it by using form instead of get_json: @app.route("/add-todo", methods=["POST"]) def add_todo(): title = request.form.get("title") # Here ---------^ db.session.add(Todo(title=title)) db.session.commit() return '', 204
Question: I have two av streams, one video and one audio, i'm trying to pipe both as inputs to ffmpeg os.mkfifo(VIDEO_PIPE_NAME) os.mkfifo(AUDIO_PIPE_NAME) ffmpeg_process = subprocess.Popen([ "ffmpeg", "-i", VIDEO_PIPE_NAME, "-i", AUDIO_PIPE_NAME, "-listen", "1", "-c:v", "copy", "-c:a", "copy", "-f", "mp4", "-movflags", "frag_keyframe+empty_moov", "http://0.0.0.0:8081"], stdin=subprocess.PIPE) pipe_video = open(VIDEO_PIPE_NAME,'wb') pipe_audio = open(AUDIO_PIPE_NAME,'wb') #Code stuck here  The code stuck on pipe_audio = open(AUDIO_PIPE_NAME,'wb') line, I'm guessing it is happening because ffmpeg only reads the first/video input and ignore the second/audio input so the pipe is not being read. If I use only pipe_video and removing -i AUDIO_PIPE_NAME flag from ffmpeg everything works fine, but i only get the video without the audio.

Answer: I can confirm chrslg's comment under the OP. Placing each pipe's open() call in its own write thread should resolve your issue. I'm nearing to release a new version of ffmpegio to introduce this exact feature, and it works well.
Question: I am writing a Python tkinter rectangle function: def Rectangle(row,col,color="#FFFFFF",outline="gray"): """ Fills a block with a color. Args: row - The row col - The column *color - The rectangle color *outline - The color of the outline """ global blockwidth, blockheight, canvas x, y, i = blockCoords(row,col) canvas.create_rectangle(x//2,y//2,x*2,y*2,fill=color,outline=outline)  This function should make a rectangle at those (row, col) coordinates exactly one block (rectangle in the drawn grid, explained later) big. def blockCoords(row, col): """ Gets the row and col value of a digit. Args: rows - The block's row cols - The block's col Returns: The x and y coordinates and it's respective index """ global blockwidth, blockheight, canwidth, canheight row -= 1 col -= 1 x = col * blockwidth + blockwidth // 2 y = row * blockheight + blockheight // 2 i = row+col return x, y, i  Then the rectangle function is called in the main function: blockwidth = 20 blockheight = 40 win = tk.Tk() states = [] win.title("Terminos App") with open("set.json","r") as f: settings = json.load(f) # create a canvas canwidth = 1200 canheight = 800 canvas = tk.Canvas(win,width=canwidth,height=canheight,bg="#232627") canvas.pack(fill=tk.BOTH,expand=tk.YES) def main(): """ The main function that handles the window Args: event - A detected keypress """ global win, canvas, canwidth, canheight, blockwidth, blockheight cols = canwidth//blockwidth rows = canheight//blockheight for row in range(canwidth): for col in range(canheight): states.append("") x1 = col * blockwidth y1 = row * blockheight x2 = x1 + blockwidth y2 = y1 + blockheight canvas.create_rectangle(x1, y1, x2, y2, outline="gray") # Used for debugging purposes Rectangle(10,15)  The code should create the rectangle I explained, but it creates one with a different size depending on the row and col coordinates. What is the problem and what is the solution? Thanks!

Answer: You're currently treating the center point (x, y) as if it were the top-left and bottom-right corners. Instead of x//2 and x*2, calculate the corners based on width and height like this:  canvas.create_rectangle(x - blockwidth // 2, y - blockheight // 2, x + blockwidth // 2, y + blockheight // 2, fill=color, outline=outline)  That way, the rectangle will be exactly blockwidth × blockheight in size, centred at the calculated coordinates.
Question: I want to run the DeepSeek-V3 model inference using the Hugging-Face Transformer library (>= v4.51.0). I read that you can do the following to do that (download the model and run it) from transformers import pipeline messages = [ {"role": "user", "content": "Who are you?"}, ] pipe = pipeline("text-generation", model="deepseek-ai/DeepSeek-R1", trust_remote_code=True) pipe(messages)  My issue is that I already downloaded the DeepSeek-V3 hugging-face repository separately, and I just want to tell the Transformer where it is on my local machine, so that it can run the inference. The model repository is thus not (or not necessarily) in the Hugging-Face cache directory (it can be anywhere on the local machine). When loading the model, I want to provide the path which specifically points to the model's repository on the local machine. How can I achieve that?

Answer: Since you said you downloaded the model already from Huggingface, I assume you downloaded all of the related Huggingface files including the JSON files in the repo that describe the model for loading. In this case, the pipeline function can easily take a filesystem path in the model parameter instead of a model name. For example, if you downloaded the files to the folder /my-models/deepseek-r1, you just need to load it this way pipe = pipeline("text-generation", model="/my-models/deepseek-r1", trust_remote_code=True)  The pipeline function will try to load from the filesystem, and if not found, it will try to look for a model with that name on the Hub.
Question: I would like to create a NumPy structured array b with the same shape as a and (-1, 1) values, for example: import numpy as np Point = [('x', 'i4'), ('y', 'i4')] a = np.zeros((4, 4), dtype='u1') b = np.full_like(a, fill_value=(-1, 1), dtype=Point) # fails b = np.full_like(a, -1, dtype=Point) # works  Using full_like() works with the same value for all fields, but fails with different values, producing this error:  multiarray.copyto(res, fill_value, casting='unsafe') ValueError: could not broadcast input array from shape (2,) into shape (4,4)  . Is there a solution other than explicitly assigning (-1, 1) to each element in a loop?

Answer: Convert the fill value into an array with the Point dtype as well import numpy as np Point = [('x', 'i4'), ('y', 'i4')] a = np.zeros((4, 4), dtype='u1') b = np.full_like(a, fill_value=np.array((-1, 1), dtype=Point), dtype=Point) # works  Alternatively, if you don't need a, just create the array directly with your desired shape import numpy as np Point = [('x', 'i4'), ('y', 'i4')] b = np.full((4, 4), np.array((-1, 1), dtype=Point))
Question: I try to open whois EURID website with no luck. Selenium opens browser (tried with Chrome and FF), but when I try to open particular URL (http://whois.eurid.eu/): nothing opens, I got blank page only. I have driver set by this function:  def set_driver() -> WebDriver: service = Service() if cfg["browser"]["browser_name"].strip().lower() == "firefox": options = webdriver.FirefoxOptions() driver = webdriver.Firefox(service=service, options=options) elif cfg["browser"]["browser_name"].strip().lower() == "edge": options = webdriver.EdgeOptions() driver = webdriver.Edge(service=service, options=options) else: options = webdriver.ChromeOptions() driver = webdriver.Chrome(service=service, options=options) return driver  Using driver (set above) I try to open webiste and scrape some data (information about domain):  domain_name = "ddddddd.eu" # any name ending with '.eu' URL = f"https://whois.eurid.eu/en/search/?domain={domain_name}" drv = set_driver() drv.get(URL) time.sleep(5)  I think this website is currently somehow protected, as I have had no problems with opening it before.

Answer: Cloudflare Protection or User Agent Detection or JavaScript might cause the issue. Please find below an enhanced solution for your consideration: from selenium import webdriver from selenium.webdriver.chrome.service import Service from selenium.webdriver.chrome.options import Options from selenium.webdriver.common.by import By from selenium.webdriver.support.ui import WebDriverWait from selenium.webdriver.support import expected_conditions as EC import time def get_eurid_whois(domain): chrome_options = Options() chrome_options.add_argument("--start-maximized") chrome_options.add_argument("--disable-blink-features=AutomationControlled") chrome_options.add_argument("--disable-infobars") chrome_options.add_argument("--disable-extensions") chrome_options.add_argument("--disable-popup-blocking") chrome_options.add_argument("user-agent=Mozilla/5.0 (Windows NT 10.0; Win64; x64) " "AppleWebKit/537.36 (KHTML, like Gecko) " "Chrome/122.0.0.0 Safari/537.36") driver = webdriver.Chrome(service=Service(), options=chrome_options) try: url = f"https://whois.eurid.eu/en/search/?domain={domain}" driver.get(url) time.sleep(3) try: WebDriverWait(driver, 15).until( EC.presence_of_element_located((By.CSS_SELECTOR, ".btn")) ) except: print("Page didn't load properly. Check if blocked by CAPTCHA.") return None try: register_button = driver.find_element(By.XPATH, "//a[contains(text(), 'Register now')]") print("Domain is available. 'Register now' button found.") return "Available" except: print("Domain is not available. 'Register now' button not found.") whois_data = driver.find_element(By.CSS_SELECTOR, ".btn").text return whois_data finally: driver.quit() print(get_eurid_whois("ddddddd.eu"))
Question: The problem I am solving is optimizing a genetic algorithm for the Traveling Salesman Problem. Calculating the path takes the most time. Here is the current code I am working on: from itertools import pairwise import numpy as np from random import shuffle def get_path_len(adj_mat: np.ndarray, path: np.ndarray) -> float: return sum(adj_mat[i, j] for i, j in pairwise(path)) + adj_mat[path[-1], path[0]] mat = np.random.randint(1, 1000, (100, 100)) path = np.asarray(list(range(100))) shuffle(path) print(get_path_len(mat, path)) 

Answer: Here is an example of how to use Numba to do that efficiently: import numba as nb # Pre-compile the code for a int32/int64 adj_mat 2D array # and a int64 path 1D array @nb.njit(['(int32[:,:], int64[:])', '(int64[:,:], int64[:])']) def get_path_len(adj_mat, path): s = 0 for i in range(path.size-1): # Assume path contains integers less than min(adj_mat.shape) s += adj_mat[path[i], path[i+1]] return s + adj_mat[path[-1], path[0]]  Here is performance results on my i5-9600KF CPU and Numpy 2.1.3: Naive unvectorised code: 28.5 µs Numba code: 0.6 µs
Question: I have column column1 of type string with values like "some1,some2,some3". I need to create boolean columns based on column1 like: some1, some2, some3. For example, I have the dataframe with one column column1: column1 | -------------------+ "some1,some2,some3"| "some2,some3" | "some1" | "some1,some3" |  I need to get the dataframe with three boolean columns: some1 | some2 | some3 | ------+-------+-------+ True | True | True | False | True | True | True | False | False | True | False | True | 

Answer: You can use str.get_dummies + astype: df['column1'].str.get_dummies(sep=',').astype(bool)  Output:  some1 some2 some3 0 True True True 1 False True True 2 True False False 3 True False True  Data used import pandas as pd data = {'column1': ["some1,some2,some3", "some2,some3", "some1", "some1,some3"]} df = pd.DataFrame(data)
Question: Currently this code prints, MainThread done at sec 1 MainThread done at sec 1 MainThread done at sec 2 MainThread done at sec 1 MainThread done at sec 0 MainThread done at sec 3 MainThread done at sec 2 MainThread done at sec 5 MainThread done at sec 4  I need it to print MainThread 1 done at sec 1 MainThread 3 done at sec 1 MainThread 2 done at sec 2 MainThread 3 done at sec 1 MainThread 1 done at sec 0 MainThread 2 done at sec 3 MainThread 2 done at sec 2 MainThread 3 done at sec 5 MainThread 1 done at sec 4  How can I do this, by modifying this 1 line of code. print(threading.current_thread().name + ' done at sec ' + str(future.result()))  Here is the full code import concurrent.futures import threading import pdb import time import random def Threads2(time_sleep): time.sleep(time_sleep) return time_sleep all_sections = [random.randint(0, 5) for iii in range(10)] test1 = [] with concurrent.futures.ThreadPoolExecutor(max_workers=3, thread_name_prefix="Ok") as executor: working_threads = {executor.submit(Threads2, time_sleep): time_sleep for index, time_sleep in enumerate(all_sections)} for future in concurrent.futures.as_completed(working_threads): print(threading.current_thread().name + ' done at sec ' + str(future.result())) 

Answer: future doesn't know which thread ran the function, you need to store that yourself. you can wrap the functor in a wrapper that will return whatever you want from the executing thread. import concurrent.futures import threading import pdb import time import random from functools import wraps def wrap_function(func): @wraps(func) def wrapper_functor(*args, **kwargs): return (threading.current_thread().name, threading.current_thread().ident, func(*args, **kwargs)) # calls the wrapped function return wrapper_functor # return a functor that wraps the function def Threads2(time_sleep): time.sleep(time_sleep) return time_sleep all_sections = [random.randint(0, 5) for iii in range(10)] test1 = [] with concurrent.futures.ThreadPoolExecutor( max_workers=3, thread_name_prefix="Ok") as executor: working_threads = { executor.submit(wrap_function(Threads2), time_sleep): time_sleep for index, time_sleep in enumerate(all_sections)} for future in concurrent.futures.as_completed(working_threads): thread_name, thread_id, result = future.result() print(thread_name, thread_id, 'done at sec', str(result))  Ok_0 18532 done at sec 0 Ok_2 12424 done at sec 1 Ok_0 18532 done at sec 3 Ok_2 12424 done at sec 2 Ok_1 19356 done at sec 4 Ok_0 18532 done at sec 4 Ok_1 19356 done at sec 3 Ok_2 12424 done at sec 4 Ok_0 18532 done at sec 3 Ok_1 19356 done at sec 4  if you want just 0,1,2 then you can use thread_name.split('_')[-1]  a super compressed form to do this with a lambda (which shouldn't pass a code review) is working_threads = { executor.submit(lambda *args, **kwargs: ( threading.current_thread().name, threading.current_thread().ident, Threads2(*args, **kwargs)), time_sleep): time_sleep for index, time_sleep in enumerate(all_sections)}
Question: I've created a module persist containing a function obj_pickle which I'm using to pickle various objects in my project. It work fine but it's failing pytest, returning; > pickle.dump(obj, file_handle, protocol=protocol) E AttributeError: Can't pickle local object 'test_object.<locals>.TestObject'  Running:  python 3.12 pytest Version: 8.3.5  I see that the are similar looking issues tied to multiprocessing but I don't think pytest is exposing that issue. Code below, and many thanks. # lib.persist.py from pathlib import Path import pickle def obj_pickle(obj: object, dir:Path, protocol: int = pickle.HIGHEST_PROTOCOL) -> None: """ Pickle an object to a byte file. """ if not dir.exists(): dir.mkdir(parents=True, exist_ok=True) path = Path(dir, obj.instance_name + '.pkl') with open(path, "wb") as file_handle: pickle.dump(obj, file_handle, protocol=protocol) print(f"{obj.__class__.__name__} object {obj.instance_name} saved to {path}")  # tests.test_persist.py from pathlib import Path import pytest from lib.persist import obj_pickle TEST_DIR = Path("test_dir") @pytest.fixture def test_object(): class TestObject(): def __init__(self, instance_file_name): self.instance_file_name = instance_file_name self.data = "This is a test object." test_object = TestObject("test_object_file") return test_object def test_obj_pickle(test_object): obj_pickle(test_object, Path(TEST_DIR)) path = Path(TEST_DIR, "test_object_file" + ".pkl") assert path.exists() 

Answer: This doesn't really have anything to do with pytest. When you load an object from a pickle, the pickle machinery needs to know what the class of that object should be. But if you define a class inside a function, then every execution of that function generates a whole new class. pickle has no way to tell which version of the class to use, if any versions even exist in the Python session in which you're loading the pickle. Instead of trying to make it work, and creating a pile of weird inconsistencies and edge cases, pickle just doesn't support pickling instances of classes like that. You need to move your class definition out of the function you have it in: class TestObject: def __init__(self, instance_file_name): self.instance_file_name = instance_file_name self.data = "This is a test object." @pytest.fixture def test_object(): return TestObject("test_object_file")
Question: I have a pandas dataframe df. It has multi-index with Gx.Region and Scenario_Model. The Scenario_Model index is ordered in alphabetical order des, pes, tes. When I plot it, it comes in the same order. However, I want to reorder it as pes, tes and des, and plot it accordingly. Is it possible to achieve it in Python pandas dataframe?  dict = {('Value', 2023, 'BatteryStorage'): {('Central Africa', 'des'): 0.0, ('Central Africa', 'pes'): 0.0, ('Central Africa', 'tes'): 0.0, ('Eastern Africa', 'des'): 0.0, ('Eastern Africa', 'pes'): 0.0, ('Eastern Africa', 'tes'): 0.0, ('North Africa', 'des'): 0.0, ('North Africa', 'pes'): 0.0, ('North Africa', 'tes'): 0.0, ('Southern Africa', 'des'): 504.0, ('Southern Africa', 'pes'): 100.0, ('Southern Africa', 'tes'): 360.0, ('West Africa', 'des'): 0.0, ('West Africa', 'pes'): 0.0, ('West Africa', 'tes'): 0.0}, ('Value', 2023, 'Biomass PP'): {('Central Africa', 'des'): 0.0, ('Central Africa', 'pes'): 0.0, ('Central Africa', 'tes'): 0.0, ('Eastern Africa', 'des'): 40, ('Eastern Africa', 'pes'): 10, ('Eastern Africa', 'tes'): 50, ('North Africa', 'des'): 0.0, ('North Africa', 'pes'): 0.0, ('North Africa', 'tes'): 0.0, ('Southern Africa', 'des'): 90.0, ('Southern Africa', 'pes'): 43.0, ('Southern Africa', 'tes'): 50.0, ('West Africa', 'des'): 200.0, ('West Africa', 'pes'): 150.0, ('West Africa', 'tes'): 100}} df_sample = pd.DataFrame.from_dict(dict) df_sample.plot(kind = "bar", stacked = True)  

Answer: A quick an easy approach, if you know the categories, would be to reindex: (df_sample.reindex(['pes', 'tes', 'des'], level=1) .plot(kind='bar', stacked=True) )  A more canonical (but more complex) approach would be to make the second level an ordered Categorical: order = pd.CategoricalDtype(['pes', 'tes', 'des'], ordered=True) (df_sample .set_axis(pd.MultiIndex.from_frame(df_sample.index.to_frame() .astype({1: order}), names=[None, None])) .sort_index() .plot(kind='bar', stacked=True) )  Output: