eshangj/stackoverflow_q_and_a_sample huggingface dataset:
Question: I need my model to accept either a bytes type variable or a string type variable and to raise an exception if any other type was passed. from typing import Union from pydantic import BaseModel class MyModel(BaseModel): a: Union[bytes, str] m1 = MyModel(a='123') m2 = MyModel(a=b'123') print(type(m1.a)) print(type(m2.a))  In my case the model interprets both bytes and string as bytes. Output: <class 'bytes'> <class 'bytes'>  Desired output: <class 'str'> <class 'bytes'>  The desired output above can be achieved if I re-assign member a: m1 = MyModel(a='123') m1.a = '123'  Is it possible to get it in one go?

Answer: The problem you are facing is that the str type does some automatic conversions (here in the docs):  strings are accepted as-is, int float and Decimal are coerced using str(v), bytes and bytearray are converted using v.decode(), enums inheriting from str are converted using v.value, and all other types cause an error   bytes are accepted as-is, bytearray is converted using bytes(v), str are converted using v.encode(), and int, float, and Decimal are coerced using str(v).encode()  You can use StrictTypes to avoid automatic conversion between compatible types (e.g.: str and bytes): from typing import Union from pydantic import BaseModel, StrictStr, StrictBytes class MyModel(BaseModel): a: Union[StrictStr, StrictBytes] m1 = MyModel(a='123') m2 = MyModel(a=b'123') print(type(m1.a)) print(type(m2.a))  Output will be as expected: <class 'str'> <class 'bytes'>
Question: I have the following DataFrame, with over 3 million rows: VALID_FROM VALID_TO VALUE 0 2022-01-01 2022-01-02 5 1 2022-01-01 2022-01-03 2 2 2022-01-02 2022-01-04 7 3 2022-01-03 2022-01-06 3  I want to create one large date_range with a sum of the values for each timestamp. For the DataFrame above that would come out to:  dates val 0 2022-01-01 7 1 2022-01-02 14 2 2022-01-03 12 3 2022-01-04 10 4 2022-01-05 3 5 2022-01-06 3  However, as the DataFrame has a little over 3 Million rows I don't want to iterate over each row and I'm not sure how to do this without iterating. Any suggestions? Currently my code looks like this: new_df = pd.DataFrame() for idx, row in dummy_df.iterrows(): dr = pd.date_range(row["VALID_FROM"], end = row["VALID_TO"], freq = "D") tmp_df = pd.DataFrame({"dates": dr, "val": row["VALUE"]}) new_df = pd.concat(objs=[new_df, tmp_df], ignore_index=True) new_df.groupby("dates", as_index=False, group_keys=False).sum()  The result of the groupby would be my desired output.

Answer: If performance is important use Index.repeat with DataFrame.loc for new rows, create date colun with counter by GroupBy.cumcount and last aggregate sum: df['VALID_FROM'] = pd.to_datetime(df['VALID_FROM']) df['VALID_TO'] = pd.to_datetime(df['VALID_TO']) df1 = df.loc[df.index.repeat(df['VALID_TO'].sub(df['VALID_FROM']).dt.days + 1)] df1['dates'] = df1['VALID_FROM'] + pd.to_timedelta(df1.groupby(level=0).cumcount(),unit='d') df1 = df1.groupby('dates', as_index=False)['VALUE'].sum() print (df1) dates VALUE 0 2022-01-01 7 1 2022-01-02 14 2 2022-01-03 12 3 2022-01-04 10 4 2022-01-05 3 5 2022-01-06 3
Question: I am using Python 3.11. Type hinting for dict of dicts of strs will look like this: dict[dict[str, str]]  But what if I want to make hints for dict of unknown depth? For example, I want to write a function, which construct tree in dict form from list of tuples (parent, offspring): source = [('a', 'b'), ('b', 'c'), ('d', 'e')] target = {'a': {'b': {'c': {}}}, 'd': {'e': {}}} def tree_form(source: list[tuple[str, str]]) -> ???: """code""" pass  What should I write instead of '???'?

Answer: You can use a type alias with a forward reference to itself: from typing import TypeAlias NestedDict: TypeAlias = dict[str, str | 'NestedDict'] def tree_form(source: list[tuple[str, str]]) -> NestedDict: return {'a': {'b': {'c': {}}}, 'd': {'e': {}}} print(tree_form([('a', 'b'), ('b', 'c'), ('d', 'e')]))  Demo of this code passing mypy: https://mypy-play.net/?mypy=latest&python=3.10&gist=6d359c16ab3f5e82b5cd2cdf9e142a6d
Question: I want to apply a custom function which takes 2 columns and outputs a value based on those (row-based) In Pandas there is a syntax to apply a function based on values in multiple columns df['col_3'] = df.apply(lambda x: func(x.col_1, x.col_2), axis=1)  What is the syntax for this in Polars?

Answer: In polars, you don't add columns by assigning just the value of the new column. You always have to assign the whole df (in other words there's never ['col_3'] on the left side of the =) To that end if you want your original df with a new column then you use the with_columns method. you would do df = ( df .with_columns( pl.struct('col_1','col_2') .map_elements(lambda x: func(x['col_1'], x['col_2'])) .alias('col_3') ) )  A struct is a dataframe inside a column of a dataframe. This is helpful because map_elements (and indeed all expressions) can only be invoked from a single column. The map_elements turns the struct, in each row, into dict and that becomes the input to your function. map_elements is for functions which take a single input and output a single value. (If you're using a vectorized function that expects something like a list and returns another list then you should use map_batches). Finally, you do alias on that to give it the name you want it to have.
Question: I'm trying to upsert using SQLAlchemy. There's no upsert in SQL but SQLAlchemy provides this. The same thing I'm trying to perform with SQLAlchemy ORM session. My code: from sqlalchemy.orm import sessionmaker Session = sessionmaker(engine) with Session() as session: """Here upsert functionality""" session.insert_or_update(company) session.commit()  session.merge(company) works as I only need to check for primary key and not other unique values. The documentation says:  Session.merge() examines the primary key attributes of the source instance, and attempts to reconcile it with an instance of the same primary key in the session. If not found locally, it attempts to load the object from the database based on primary key, and if none can be located, creates a new instance. The state of each attribute on the source instance is then copied to the target instance. The resulting target instance is then returned by the method; the original source instance is left unmodified, and un-associated with the Session if not already.  How do I perform this for multiple objects?

Answer: As you have noted, Session.merge() will accomplish the task on an object-by-object basis. For example, if we have class Thing(Base): __tablename__ = "thing" id: Mapped[int] = mapped_column(primary_key=True, autoincrement=False) txt: Mapped[str] = mapped_column(String(50)) my_thing = Thing(id=1, txt="foo")  we can do with Session(engine) as sess: sess.merge(my_thing) sess.commit()  However, we don't want to do that for a large number of objects, e.g., rows_to_upsert = 4000 things = [Thing(id=i, txt=f"txt_{i}") for i in range(rows_to_upsert)] with Session(engine, autoflush=False) as sess: t0 = time.perf_counter() for thing in things: sess.merge(thing) sess.commit() print( f"merge: {rows_to_upsert:,} rows upserted in {(time.perf_counter() - t0):0.1f} seconds" )  because that will result in 4000 round-trips to the server to perform a SELECT for each object, and that will be slow. In my testing, upserting 4000 rows took approximately 40 seconds, or about 100 rows/sec. Instead, we should convert the list of objects to a list of dict list_of_dict = [dict(id=thing.id, txt=thing.txt) for thing in things]  and then use an INSERT … ON DUPLICATE KEY statement from sqlalchemy.dialects.mysql import insert insert_stmt = insert(Thing).values(list_of_dict) on_duplicate_stmt = insert_stmt.on_duplicate_key_update( dict(txt=insert_stmt.inserted.txt) ) with Session(engine) as sess: t0 = time.perf_counter() sess.execute(on_duplicate_stmt) sess.commit() print( f"insert: {rows_to_upsert:,} rows upserted in {(time.perf_counter() - t0):0.1f} seconds" )  which only took around 2 seconds, or about 2000 rows/sec.
Question: I have an SSH connection from a Windows machine to another, and then trying to do a poetry install. My problem is: I get this error when executing poetry install through ssh: [WinError 1312] A specified logon session does not exist. It may already have been terminated.  This command works perfectly when I execute it locally on the target machine, but fails when connecting through ssh. How can I get rid/fix the [WinError 1312]? I saw another user that posted the same question recently, but removed it. I've seen some clues regarding the MachineKeys, but have really no idea on how to proceed. Any suggestion will be highly appreciated.  Python: 3.10.8 Poetry: 1.2.1 Installing dependencies from lock file Package operations: 5 installs, 0 updates, 0 removals • Installing install-requires (0.3.0) OSError [WinError 1312] A specified logon session does not exist. It may already have been terminated. at ~\AppData\Roaming\pypoetry\venv\lib\site-packages\win32ctypes\core\ctypes\_util.py:53 in check_zero 49│ 50│ def check_zero_factory(function_name=None): 51│ def check_zero(result, function, arguments, *args): 52│ if result == 0: → 53│ raise make_error(function, function_name) 54│ return result 55│ return check_zero 56│ 57│ The following error occurred when trying to handle this error: error (1312, 'CredRead', 'A specified logon session does not exist. It may already have been terminated.') at ~\AppData\Roaming\pypoetry\venv\lib\site-packages\win32ctypes\pywin32\pywintypes.py:37 in pywin32error 33│ def pywin32error(): 34│ try: 35│ yield 36│ except WindowsError as exception: → 37│ raise error(exception.winerror, exception.function, exception.strerror) 38│ 

Answer: Based on similarities in the stack traces and your description, my guess is that you're facing the same bug from #1892 and #1917, where Poetry tries to use your keyring to access/publish modules, and hence fails when these credentials are invalid.  But it appears that poetry tries to access the keyring even for install operations.  One of the solutions proposed was to uninstall the keyring package remotely:  For me, I worked around the problem by pip uninstalling the 'keyring' package from that virt env.  Another solution is to export the environment variable PYTHON_KEYRING_BACKEND. Here's an example of how you can do that on Windows cmd: SET PYTHON_KEYRING_BACKEND=keyring.backends.null.Keyring  or Windows powershell: $env:PYTHON_KEYRING_BACKEND="keyring.backends.null.Keyring"  or Linux shell: export PYTHON_KEYRING_BACKEND=keyring.backends.null.Keyring  Unfortunately, it appears that issue #1917 is still open and unresolved, so this is the best workaround that you can find to fix the issue for now.
Question: I am currently using the diffusers StableDiffusionPipeline (from hugging face) to generate AI images with a discord bot which I use with my friends. I was wondering if it was possible to get a preview of the image being generated before it is finished? For example, if an image takes 20 seconds to generate, since it is using diffusion it starts off blury and gradually gets better and better. What I want is to save the image on each iteration (or every few seconds) and see how it progresses. How would I be able to do this? class ImageGenerator: def __init__(self, socket_listener, pretty_logger, prisma): self.model = StableDiffusionPipeline.from_pretrained("runwayml/stable-diffusion-v1-5", revision="fp16", torch_dtype=torch.float16, use_auth_token=os.environ.get("HF_AUTH_TOKEN")) self.model = self.model.to("cuda") async def generate_image(self, data): start_time = time.time() with autocast("cuda"): image = self.model(data.description, height=self.default_height, width=self.default_width, num_inference_steps=self.default_inference_steps, guidance_scale=self.default_guidance_scale) image.save(...)  The code I have currently is this, however it only returns the image when it is completely done. I have tried to look into how the image is generated inside of the StableDiffusionPipeline but I cannot find anywhere where the image is generated. If anybody could provide any pointers/tips on where I can begin that would be very helpful.

Answer: You can use the callback argument of the stable diffusion pipeline to get the latent space representation of the image: link to documentation The implementation shows how the latents are converted back to an image. We just have to copy that code and decode the latents. Here is a small example that saves the generated image every 5 steps: from diffusers import StableDiffusionPipeline import torch #load model model = StableDiffusionPipeline.from_pretrained("runwayml/stable-diffusion-v1-5", revision="fp16", torch_dtype=torch.float16, use_auth_token="YOUR TOKEN HERE") model = model.to("cuda") def callback(iter, t, latents): # convert latents to image with torch.no_grad(): latents = 1 / 0.18215 * latents image = model.vae.decode(latents).sample image = (image / 2 + 0.5).clamp(0, 1) # we always cast to float32 as this does not cause significant overhead and is compatible with bfloa16 image = image.cpu().permute(0, 2, 3, 1).float().numpy() # convert to PIL Images image = model.numpy_to_pil(image) # do something with the Images for i, img in enumerate(image): img.save(f"iter_{iter}_img{i}.png") # generate image (note the `callback` and `callback_steps` argument) image = model("tree", callback=callback, callback_steps=5)  To understand the stable diffusion model I highly recommend this blog post.
Question: I am trying to cache the Virtual environment folder (.venv) for my Python project with CodeBuild. Here's my buildspec.yml file: version: 0.2 env: shell: bash phases: install: commands: - python3 -m venv .venv && source .venv/bin/activate - pip3 install -r requirements.txt build: commands: - pytest -v tests/ cache: paths: - .venv/**/*  This is the error I get: [Container] 2022/11/10 11:33:20 MkdirAll: /codebuild/local-cache/custom/11615810f/.venv [Container] 2022/11/10 11:33:20 Symlinking: /codebuild/output/src682375820/src/hello.org/demo/.venv => /codebuild/local-cache/custom/11615810f/.venv [Container] 2022/11/10 11:34:01 Running command python3 -m venv .venv && source .venv/bin/activate Error: Unable to create directory '/codebuild/output/src682375820/src/hello.org/demo/.venv' [Container] 2022/11/10 11:34:01 Command did not exit successfully python3 -m venv .venv && source .venv/bin/activate exit status 1 [Container] 2022/11/10 11:34:01 Phase complete: INSTALL State: FAILED [Container] 2022/11/10 11:34:01 Phase context status code: COMMAND_EXECUTION_ERROR Message: Error while executing command: python3 -m venv .venv && source .venv/bin/activate. Reason: exit status 1  The problem is that you can't instantiate a Python virtual environment into a symlinked folder. Any ideas?

Answer: CodeBuild uses symlinks to link cached directories and python -m venv tries to create a new directory over a symlink, which isn't possible. Try to run python3 -m venv command inside the .venv directory: install: commands: - cd .venv && python3 -m venv . && cd - - source .venv/bin/activate - pip3 install -r requirements.txt  If you check the CodeBuild log, you'll see that symlink is created at the very beginning: # Example when /opt/poetry is cached [Container] 2023/12/11 08:10:45.046132 Expanded cache path /opt/poetry/ [Container] 2023/12/11 08:10:45.495963 MkdirAll: /codebuild/local-cache/custom/5694e39f8fd7f99453b01828b7946ba4ef802b635c8c9ec0be9b56824a16d47c/opt/poetry [Container] 2023/12/11 08:10:45.496095 Symlinking: /opt/poetry => /codebuild/local-cache/custom/5694e39f8fd7f99453b01828b7946ba4ef802b635c8c9ec0be9b56824a16d47c/opt/poetry
Question: I am using JupyterLab and trying to run tqdm. I've had an error persist for quite a while that seems to be a JS error. Extensions: Other labextensions (built into JupyterLab) app dir: /opt/homebrew/Cellar/python@3.9/3.9.10/Frameworks/Python.framework/Versions/3.9/share/jupyter/lab @jupyter-widgets/jupyterlab-manager v5.0.3 enabled OK jupyter-leaflet v0.17.2 enabled OK  Error Message: Failed to load model class 'VBoxModel' from module '@jupyter-widgets/controls' Error: Module @jupyter-widgets/controls, version ^1.5.0 is not registered, however, 2.0.0 is  I have tried, re-installing, re-building all of the extensions listed above to no avail.

Answer: I had the same issue when upgrading to jupyterlab==3.6.3. After upgrading ipywidgets to the at that time latest version 8.0.6, it still failed. I managed to fix the issue by downgrading then to ipywidgets==7.7.5.
Question: I am sending a request to some URL. I copied the curl command to python. So, all the headers are included, but my request is not working and I receive status code 403 and error code 1020 in the HTML output. The code is import requests headers = { 'User-Agent': 'Mozilla/5.0 (Windows NT 10.0; Win64; x64; rv:106.0) Gecko/20100101 Firefox/106.0', 'Accept': 'text/html,application/xhtml+xml,application/xml;q=0.9,image/avif,image/webp,*/*;q=0.8', 'Accept-Language': 'en-US,en;q=0.5', # 'Accept-Encoding': 'gzip, deflate, br', 'DNT': '1', 'Connection': 'keep-alive', 'Upgrade-Insecure-Requests': '1', 'Sec-Fetch-Dest': 'document', 'Sec-Fetch-Mode': 'navigate', 'Sec-Fetch-Site': 'none', 'Sec-Fetch-User': '?1', } response = requests.get('https://v2.gcchmc.org/book-appointment/', headers=headers) print(response.status_code) print(response.cookies.get_dict()) with open("test.html",'w') as f: f.write(response.text)  I also get cookies but not getting the desired response. I know I can do it with selenium but I want to know the reason behind this. Note: I have installed all the libraries and checked the versions, but it is still not working and throwing a 403 error.

Answer: The site is protected by cloudflare which aims to block, among other things, unauthorized data scraping. From What is data scraping?   The process of web scraping is fairly simple, though the implementation can be complex. Web scraping occurs in 3 steps:  First the piece of code used to pull the information, which we call a scraper bot, sends an HTTP GET request to a specific website. When the website responds, the scraper parses the HTML document for a specific pattern of data. Once the data is extracted, it is converted into whatever specific format the scraper bot’s author designed.   You can use urllib instead of requests, it seems to be able to deal with cloudflare req = urllib.request.Request('https://v2.gcchmc.org/book-appointment/') req.add_headers('User-Agent', 'Mozilla/5.0 (Windows NT 10.0; Win64; x64; rv:106.0) Gecko/20100101 Firefox/106.0') req.add_header('Accept', 'text/html,application/xhtml+xml,application/xml;q=0.9,image/avif,image/webp,*/*;q=0.8') req.add_header('Accept-Language', 'en-US,en;q=0.5') r = urllib.request.urlopen(req).read().decode('utf-8') with open("test.html", 'w', encoding="utf-8") as f: f.write(r)
Question: I installed a "dnspython" package with "pip install dnspython" under Ubuntu 22.10 and made a following short script: #!/usr/bin/env python3 import dns.zone import dns.query zone = dns.zone.Zone("example.net") dns.query.inbound_xfr("10.0.0.1", zone) for (name, ttl, rdata) in zone.iterate_rdatas("SOA"): serial_nr = rdata.serial  When I check this code snippet with mypy(version 0.990), then it reports an error: Module has no attribute "inbound_xfr" [attr-defined] for line number 7. According to mypy documentation, if a Python file and a stub file are both present in the same directory on the search path, then only the stub file is used. In case of "dnspython", the stub file query.pyi is present in the dns package and the stub file indeed has no attribute "inbound_xfr". When I rename or remove the stub file, then the query.py Python file is used instead of the stub file and mypy no longer complains about missing attribute. I guess this is a "dnspython" bug? Is there a way to tell to mypy that for query module, the stub file should be ignored?

Answer: I would recommend ignoring only the specific wrong line, not the whole module. dns.query.inbound_xfr("10.0.0.1", zone) # type: ignore[attr-defined]  This will suppress attr-defined error message that is generated on that line. If you're going to take this approach, I'd also recommend running mypy with the --warn-unused-ignores flag, which will report any redundant and unused # type: ignore statements (for example, after updating the library).
Question: I want to run a Snakemake workflow where the input is defined by a combination of different variables (e.g. pairs of samples, sample ID and Nanopore barcode,...): sample_1 = ["foo", "bar", "baz"] sample_2 = ["spam", "ham", "eggs"]  I've got a rule using these: rule frobnicate: input: assembly = "{first_sample}_{second_sample}.txt" output: frobnicated = "{first_sample}_{second_sample}.frob"  I now want to create a rule all that will do this for some combinations of the samples in sample_1 and sample_2, but not all of them. Using expand would give me all possible combinations of sample_1 and sample_2. How can I, for example, just combine the first variable in the first list with the first in the second and so on (foo_spam.frob, bar_ham.frob, and baz_eggs.frob)? And what if I want some more complex combination?

Answer: Using expand with other combinatoric functions By default, expand uses the itertools function product. However, it's possible to specify another function for expand to use.  To combine the first variable in the first with the first in the second and so on, one can tell expand to use zip: sample_1 = ["foo", "bar", "baz"] sample_2 = ["spam", "ham", "eggs"]  rule all: input: expand("{first_sample}_{second_sample}.frob", zip, first_sample=sample_1, second_sample=sample_2)  will yield foo_spam.frob, bar_ham.frob, and baz_eggs.frob as inputs to rule all. Using regular Python code to generate your list The input generated by expand is ultimately just a list of file names. If you can't get where you want to with expand and another combinatoric function, it could be easier to just use regular Python code to generate the list yourself (for an example of this in action, see this question). The brute-force solution: just write the list yourself If your combination of inputs can't be arrived at programmatically at all, one last resort would be to write out the combinations you want by hand. For example: sample_1 = ["foo", "bar", "baz"] sample_2 = ["spam", "ham", "eggs"] all_frobnicated = ["foo_eggs.frob", "bar_spam.frob", "baz_ham.frob"]  rule all: input: all_frobnicated  This will, of course, mean your inputs are completely hardcoded, so if you want to use this workflow with a new batch, you'll have to write the sample combinations you want there out by hand as well.
Question: If I run a basic TWS example, I receive the error message . If I comment out the error() call back it runs fine. I've tried this on several examples and get the same result.  Exception has occurred: TypeError error() takes 4 positional arguments but 5 were given File "/Users/jayurbain/Dropbox/twsapi/Algorithmic Trading using Interactive Broker's Python API /ib_basic_app.py", line 20, in <module> app.run()  Please advise. Here's the call back that is being overridden in wrapper.py:  def error(self, reqId:TickerId, errorCode:int, errorString:str, advancedOrderRejectJson = ""):  Here is the entire code: from ibapi.client import EClient from ibapi.wrapper import EWrapper class TradingApp(EWrapper, EClient): def __init__(self): EClient.__init__(self,self) def error(self, reqId, errorCode, errorString): print("Error {} {} {}".format(reqId,errorCode,errorString)) app = TradingApp() app.connect("127.0.0.1", 7497, clientId=1) app.run() 

Answer: The solution to the problem was to execute python setup install rather than using pip to install ibapi. Thanks for your answers.