Main
View or edit on GitHub
This page is synchronized from trase/models/brazil/soy_2023_2024_v27/main.ipynb. Last modified on 2026-09-20 13:50 CEST by Nicolas Martin.
Please view or edit the original file there; changes should be reflected here after a midnight build (CET time),
or manually triggering it with a GitHub action (link).
from trase.tools import sps
import warnings
import pandas as pd
from quality_assurance import quality_assurance
from post_processing import (
export_flows_for_supply_shed,
export_flows_for_visualization,
export_supplyshed,
reconcile_supplyshed_and_supplychain,
)
from constants import *
warnings.filterwarnings("ignore")
pd.options.display.float_format = "{:,.2f}".format
for year in [2022]:
try:
print(f"Running {year}")
supplychain = sps.SupplyChain("brazil/soy_2023_2024_v27", year=year)
supplychain.max_crushing_distance = MAX_CRUSHING_DISTANCE
supplychain.preparation()
supplychain.load()
supplychain.run()
supplychain.flow_report_by_attribute("vol", ["branch"], 8)
quality_assurance(supplychain)
# Export flows for
export_flows_for_supply_shed(supplychain, year, upload_to_s3=True)
export_flows_for_visualization(supplychain, year, upload_to_s3=False)
supplychain.flow_report_by_attribute("vol", ["branch"], 8)
# Running Supply Shed Model
print(f"Running supply shed model...")
supplyshed = sps.SupplyChain("brazil/soy_supply_sheds", year=year)
supplyshed.max_crushing_distance = MAX_CRUSHING_DISTANCE
supplyshed.preparation()
supplyshed.load()
supplyshed.run()
# Exporting Data TODO - removing stiching and post_processing from main.py
print(f"Exporting supply shed results")
export_supplyshed(year, upload_to_s3=False)
# Reconcile supplyshed and supplychain
print("Reconciling Supply Shed")
reconcile_supplyshed_and_supplychain(supplychain, year)
except Exception as e:
print(f"Failed {year}: {e}")
Running 2022
Extracting data from source ...[0m [97m took 0.2 seconds[0m
[97mSkipping re-process of Municipality[0m
Extracting data from source ...[0mDownloading s3://trase-storage/brazil/logistics/abiove/out/CRUSHING_FACILITIES_2003_2025.csv[0m
---------------------------------------------------------------------------
KeyboardInterrupt Traceback (most recent call last)
Cell In[1], line 60
56 print('Reconciling Supply Shed')
57 reconcile_supplyshed_and_supplychain(supplychain, year)
58
59
---> 60 except Exception as e:
61 print(f"Failed {year}: {e}")
File ~/repos/TRASE/trase/tools/sei_pcs/utilities.py:104, in trace_memory_allocation.<locals>.wrapper(*args, **kwargs)
102 return return_value
103 else:
--> 104 return func(*args, **kwargs)
File ~/repos/TRASE/trase/tools/sei_pcs/supply_chain.py:199, in SupplyChain.preparation(self)
197 OUTPUT_FILE_OVERWRITE_TRACKER.enable()
198 with warnings_if_pd_merge_called_without_validate():
--> 199 self._call_scripts_with_function(
200 self.preparation_scripts, "preparation", self.context
201 )
202 finally:
203 OUTPUT_FILE_OVERWRITE_TRACKER.disable()
File ~/repos/TRASE/trase/tools/sei_pcs/supply_chain.py:266, in SupplyChain._call_scripts_with_function(self, scripts, function_name, *function_arguments)
262 continue
264 LOGGER.info(f"Running {function_name} from {script_file}")
--> 266 return_value = function(*function_arguments)
267 results.append(return_value)
269 return results
File ~/repos/TRASE/trase/models/brazil/soy_2023_2024_v27/preparation.py:55, in preparation(context)
53 def preparation(context):
54 df_municipality = Municipality(context).run()
---> 55 IndustrialCapacity(context).run()
56 IndustrialCapacityFacilities(context).run()
58 df_special_cases = ExporterSpecialCases(context).run()
File ~/repos/TRASE/trase/tools/etl_internal/processors.py:106, in Preprocessor.run(self, returns, *args)
104 print_message("Extracting data from source ...", end="")
105 with timing(threshold=0, indent=1):
--> 106 self._extract_from_source()
108 if self.should_rerun(args):
109 return super(DataframePreprocessor, self).run(
110 *args, returns=returns, force=True
111 )
File ~/repos/TRASE/trase/tools/etl_internal/processors.py:57, in S3Mixin._extract_from_source(self)
55 self.debug(f"Fetching source data into {path}")
56 os.makedirs(os.path.dirname(path), exist_ok=True)
---> 57 self.extract(path)
File ~/repos/TRASE/trase/tools/etl_internal/processors.py:32, in S3Mixin.extract(self, path)
30 os.makedirs(os.path.dirname(path), exist_ok=True)
31 try:
---> 32 self.client.download_file(
33 Filename=str(path),
34 Bucket=self._bucket,
35 Key=key,
36 **(
37 {}
38 if self.version_id is None
39 else {"ExtraArgs": {"VersionId": self.version_id}}
40 ),
41 )
42 except Exception as e:
43 if e.response.get("Error", {}).get("Code") == "404":
File ~/repos/TRASE/.venv/lib/python3.11/site-packages/botocore/context.py:123, in with_current_context.<locals>.decorator.<locals>.wrapper(*args, **kwargs)
121 if hook:
122 hook()
--> 123 return func(*args, **kwargs)
File ~/repos/TRASE/.venv/lib/python3.11/site-packages/boto3/s3/inject.py:223, in download_file(self, Bucket, Key, Filename, ExtraArgs, Callback, Config)
188 """Download an S3 object to a file.
189
190 Usage::
(...) 220 transfer.
221 """
222 with S3Transfer(self, Config) as transfer:
--> 223 return transfer.download_file(
224 bucket=Bucket,
225 key=Key,
226 filename=Filename,
227 extra_args=ExtraArgs,
228 callback=Callback,
229 )
File ~/repos/TRASE/.venv/lib/python3.11/site-packages/boto3/s3/transfer.py:407, in S3Transfer.download_file(self, bucket, key, filename, extra_args, callback)
403 future = self._manager.download(
404 bucket, key, filename, extra_args, subscribers
405 )
406 try:
--> 407 future.result()
408 # This is for backwards compatibility where when retries are
409 # exceeded we need to throw the same error from boto3 instead of
410 # s3transfer's built in RetriesExceededError as current users are
411 # catching the boto3 one instead of the s3transfer exception to do
412 # their own retries.
413 except S3TransferRetriesExceededError as e:
File ~/repos/TRASE/.venv/lib/python3.11/site-packages/s3transfer/futures.py:114, in TransferFuture.result(self)
112 except KeyboardInterrupt as e:
113 self.cancel()
--> 114 raise e
File ~/repos/TRASE/.venv/lib/python3.11/site-packages/s3transfer/futures.py:111, in TransferFuture.result(self)
106 def result(self):
107 try:
108 # Usually the result() method blocks until the transfer is done,
109 # however if a KeyboardInterrupt is raised we want want to exit
110 # out of this and propagate the exception.
--> 111 return self._coordinator.result()
112 except KeyboardInterrupt as e:
113 self.cancel()
File ~/repos/TRASE/.venv/lib/python3.11/site-packages/s3transfer/futures.py:282, in TransferCoordinator.result(self)
272 """Waits until TransferFuture is done and returns the result
273
274 If the TransferFuture succeeded, it will return the result. If the
275 TransferFuture failed, it will raise the exception associated to the
276 failure.
277 """
278 # Doing a wait() with no timeout cannot be interrupted in python2 but
279 # can be interrupted in python3 so we just wait with the largest
280 # possible value integer value, which is on the scale of billions of
281 # years...
--> 282 self._done_event.wait(MAXINT)
284 # Once done waiting, raise an exception if present or return the
285 # final result.
286 if self._exception:
File /opt/anaconda3/lib/python3.11/threading.py:629, in Event.wait(self, timeout)
627 signaled = self._flag
628 if not signaled:
--> 629 signaled = self._cond.wait(timeout)
630 return signaled
File /opt/anaconda3/lib/python3.11/threading.py:327, in Condition.wait(self, timeout)
325 try: # restore state no matter what (e.g., KeyboardInterrupt)
326 if timeout is None:
--> 327 waiter.acquire()
328 gotit = True
329 else:
KeyboardInterrupt:
import pandas as pd
unknown_beans_alloc = pd.read_csv("allocation_unknowns_beans.csv")
unknown_cake_oil_alloc = pd.read_csv("allocation_unknowns_cake_oil.csv")
unknown_cake_oil_alloc.shape
unknown_beans_alloc.shape