External Files

    

• pxt.Image (Python value: PIL.Image.Image)
• pxt.Video (Python value: a local path string)
• pxt.Audio (Python value: a local path string)

%pip install -qU pixeltable

import pixeltable as pxt
import random
import shutil
import tempfile

# First drop the `external_data` directory if it exists, to ensure
# a clean environment for the demo
pxt.drop_dir('external_data', force=True)
pxt.create_dir('external_data')

Connected to Pixeltable database at: postgresql+psycopg://postgres:@/pixeltable?host=/Users/asiegel/.pixeltable/pgdata
Created directory `external_data`.
<pixeltable.catalog.dir.Dir at 0x176646bb0>

v = pxt.create_table('external_data/videos', {'video': pxt.Video})

prefix = 's3://multimedia-commons/'
paths = [
    'data/videos/mp4/ffe/ffb/ffeffbef41bbc269810b2a1a888de.mp4',
    'data/videos/mp4/ffe/feb/ffefebb41485539f964760e6115fbc44.mp4',
    'data/videos/mp4/ffe/f73/ffef7384d698b5f70d411c696247169.mp4',
]
v.insert({'video': prefix + p} for p in paths)

Created table `videos`.
Computing cells:   0%|                                                    | 0/6 [00:00<?, ? cells/s]
Inserting rows into `videos`: 3 rows [00:00, 1004.62 rows/s]
Computing cells: 100%|████████████████████████████████████████████| 6/6 [00:00<00:00, 79.14 cells/s]
Inserted 3 rows with 0 errors.
UpdateStatus(num_rows=3, num_computed_values=6, num_excs=0, updated_cols=[], cols_with_excs=[])
UpdateStatus(num_rows=3, num_computed_values=0, num_excs=0, updated_cols=[], cols_with_excs=[])

rows = list(v.select(v.video).collect())
rows[0]

{‘video’: ‘/Users/asiegel/.pixeltable/file_cache/682f022a704d4459adb2f29f7fe9577c_0_1fcfcb221263cff76a2853250fbbb2e90375dd495454c0007bc6ff4430c9a4a7.mp4’}

local_path = tempfile.mktemp(suffix='.mp4')
shutil.copyfile(rows[0]['video'], local_path)
local_path

‘/var/folders/hb/qd0dztsj43j_mdb6hbl1gzyc0000gn/T/tmp1jo4a7ca.mp4’

v.insert([{'video': local_path}])

Computing cells:   0%|                                                    | 0/2 [00:00<?, ? cells/s]
Inserting rows into `videos`: 1 rows [00:00, 725.78 rows/s]
Computing cells: 100%|████████████████████████████████████████████| 2/2 [00:00<00:00, 53.23 cells/s]
Inserted 1 row with 0 errors.
UpdateStatus(num_rows=1, num_computed_values=2, num_excs=0, updated_cols=[], cols_with_excs=[])

rows = list(v.select(v.video).collect())
rows

[{‘video’: ‘/Users/asiegel/.pixeltable/file_cache/682f022a704d4459adb2f29f7fe9577c_0_1fcfcb221263cff76a2853250fbbb2e90375dd495454c0007bc6ff4430c9a4a7.mp4’},
 {‘video’: ‘/Users/asiegel/.pixeltable/file_cache/682f022a704d4459adb2f29f7fe9577c_0_fc11428b32768ae782193a57ebcbad706f45bbd9fa13354471e0bcd798fee3ea.mp4’},
 {‘video’: ‘/Users/asiegel/.pixeltable/file_cache/682f022a704d4459adb2f29f7fe9577c_0_b9fb0d9411bc9cd183a36866911baa7a8834f22f665bce47608566b38485c16a.mp4’},
 {‘video’: ‘/var/folders/hb/qd0dztsj43j_mdb6hbl1gzyc0000gn/T/tmp1jo4a7ca.mp4’}]

@pxt.udf
def f(v: pxt.Video) -> int:
    print(f'{type(v)}: {v}')
    return 1

v.select(f(v.video)).show()

<class ‘str’>: /Users/asiegel/.pixeltable/file_cache/682f022a704d4459adb2f29f7fe9577c_0_1fcfcb221263cff76a2853250fbbb2e90375dd495454c0007bc6ff4430c9a4a7.mp4
<class ‘str’>: /Users/asiegel/.pixeltable/file_cache/682f022a704d4459adb2f29f7fe9577c_0_fc11428b32768ae782193a57ebcbad706f45bbd9fa13354471e0bcd798fee3ea.mp4
<class ‘str’>: /Users/asiegel/.pixeltable/file_cache/682f022a704d4459adb2f29f7fe9577c_0_b9fb0d9411bc9cd183a36866911baa7a8834f22f665bce47608566b38485c16a.mp4
<class ‘str’>: /var/folders/hb/qd0dztsj43j_mdb6hbl1gzyc0000gn/T/tmp1jo4a7ca.mp4

​

Dealing with errors

v.insert([{'video': prefix + 'bad_path.mp4'}])

Computing cells:   0%|                                                    | 0/2 [00:01<?, ? cells/s]
Error: Failed to download s3://multimedia-commons/bad_path.mp4: An error occurred (404) when calling the HeadObject operation: Not Found
---------------------------------------------------------------------------
Error                                     Traceback (most recent call last)
Cell In[9], line 1
----> 1 v.insert(video=prefix + ’bad_path.mp4’)File ~/Dropbox/workspace/pixeltable/pixeltable/pixeltable/catalog/insertable_table.py:125, in InsertableTable.insert(self, rows, print_stats, on_error, **kwargs)
    123         raise excs.Error(’rows must be a list of dictionaries’)
    124 self._validate_input_rows(rows)
—> 125 status = self._tbl_version.insert(rows, None, print_stats=print_stats, fail_on_exception=fail_on_exception)
    127 if status.num_excs == 0:
    128     cols_with_excs_str = ’’File ~/Dropbox/workspace/pixeltable/pixeltable/pixeltable/catalog/table_version.py:723, in TableVersion.insert(self, rows, df, conn, print_stats, fail_on_exception)
    721 if conn is None:
    722     with Env.get().engine.begin() as conn:
—> 723         return self._insert(
    724             plan, conn, time.time(), print_stats=print_stats, rowids=rowids(), abort_on_exc=fail_on_exception)
    725 else:
    726     return self._insert(
    727         plan, conn, time.time(), print_stats=print_stats, rowids=rowids(), abort_on_exc=fail_on_exception)File ~/Dropbox/workspace/pixeltable/pixeltable/pixeltable/catalog/table_version.py:737, in TableVersion._insert(self, exec_plan, conn, timestamp, rowids, print_stats, abort_on_exc)
    735 self.version += 1
    736 result = UpdateStatus()
—> 737 num_rows, num_excs, cols_with_excs = self.store_tbl.insert_rows(
    738     exec_plan, conn, v_min=self.version, rowids=rowids, abort_on_exc=abort_on_exc)
    739 result.num_rows = num_rows
    740 result.num_excs = num_excsFile ~/Dropbox/workspace/pixeltable/pixeltable/pixeltable/store.py:323, in StoreBase.insert_rows(self, exec_plan, conn, v_min, show_progress, rowids, abort_on_exc)
    321 try:
    322     exec_plan.open()
—> 323     for row_batch in exec_plan:
    324         num_rows += len(row_batch)
    325         for batch_start_idx in range(0, len(row_batch), self.__INSERT_BATCH_SIZE):
    326             # compute batch of rows and convert them into table rowsFile ~/Dropbox/workspace/pixeltable/pixeltable/pixeltable/exec/expr_eval_node.py:45, in ExprEvalNode.__next__(self)
     44 def __next__(self) -> DataRowBatch:
---> 45     input_batch = next(self.input)
     46     # compute target exprs
     47     for cohort in self.cohorts:File ~/Dropbox/workspace/pixeltable/pixeltable/pixeltable/exec/cache_prefetch_node.py:71, in CachePrefetchNode.__next__(self)
     68     futures[executor.submit(self._fetch_url, row, info.slot_idx)] = (row, info)
     69 for future in concurrent.futures.as_completed(futures):
     70     # TODO:  does this need to deal with recoverable errors (such as retry after throttling)?
---> 71     tmp_path = future.result()
     72     if tmp_path is None:
     73         continueFile /opt/miniconda3/envs/pxt/lib/python3.9/concurrent/futures/_base.py:439, in Future.result(self, timeout)
    437     raise CancelledError()
    438 elif self._state == FINISHED:
—> 439     return self.__get_result()
    441 self._condition.wait(timeout)
    443 if self._state in [CANCELLED, CANCELLED_AND_NOTIFIED]:File /opt/miniconda3/envs/pxt/lib/python3.9/concurrent/futures/_base.py:391, in Future.__get_result(self)
    389 if self._exception:
    390     try:
—> 391         raise self._exception
    392     finally:
    393         # Break a reference cycle with the exception in self._exception
    394         self = NoneFile /opt/miniconda3/envs/pxt/lib/python3.9/concurrent/futures/thread.py:58, in _WorkItem.run(self)
     55     return
     57 try:
---> 58     result = self.fn(self.args, *self.kwargs)
     59 except BaseException as exc:
     60     self.future.set_exception(exc)File ~/Dropbox/workspace/pixeltable/pixeltable/pixeltable/exec/cache_prefetch_node.py:115, in CachePrefetchNode._fetch_url(self, row, slot_idx)
    113     self.row_builder.set_exc(row, slot_idx, exc)
    114     if not self.ctx.ignore_errors:
—> 115         raise exc from None  # suppress original exception
    116 return NoneError: Failed to download s3://multimedia-commons/bad_path.mp4: An error occurred (404) when calling the HeadObject operation: Not Found

# create invalid .mp4
with tempfile.NamedTemporaryFile(
    mode='wb', suffix='.mp4', delete=False
) as temp_file:
    temp_file.write(random.randbytes(1024))
    corrupted_path = temp_file.name

v.insert([{'video': corrupted_path}])

Computing cells: 100%|██████████████████████████████████████████| 2/2 [00:00<00:00, 1084.64 cells/s]
Error: Not a valid video: /var/folders/hb/qd0dztsj43j_mdb6hbl1gzyc0000gn/T/tmp3djgfyjp.mp4
---------------------------------------------------------------------------
Error                                     Traceback (most recent call last)
Cell In[10], line 6
      3     temp_file.write(random.randbytes(1024))
      4     corrupted_path = temp_file.name
----> 6 v.insert(video=corrupted_path)File ~/Dropbox/workspace/pixeltable/pixeltable/pixeltable/catalog/insertable_table.py:125, in InsertableTable.insert(self, rows, print_stats, on_error, **kwargs)
    123         raise excs.Error(’rows must be a list of dictionaries’)
    124 self._validate_input_rows(rows)
—> 125 status = self._tbl_version.insert(rows, None, print_stats=print_stats, fail_on_exception=fail_on_exception)
    127 if status.num_excs == 0:
    128     cols_with_excs_str = ’’File ~/Dropbox/workspace/pixeltable/pixeltable/pixeltable/catalog/table_version.py:723, in TableVersion.insert(self, rows, df, conn, print_stats, fail_on_exception)
    721 if conn is None:
    722     with Env.get().engine.begin() as conn:
—> 723         return self._insert(
    724             plan, conn, time.time(), print_stats=print_stats, rowids=rowids(), abort_on_exc=fail_on_exception)
    725 else:
    726     return self._insert(
    727         plan, conn, time.time(), print_stats=print_stats, rowids=rowids(), abort_on_exc=fail_on_exception)File ~/Dropbox/workspace/pixeltable/pixeltable/pixeltable/catalog/table_version.py:737, in TableVersion._insert(self, exec_plan, conn, timestamp, rowids, print_stats, abort_on_exc)
    735 self.version += 1
    736 result = UpdateStatus()
—> 737 num_rows, num_excs, cols_with_excs = self.store_tbl.insert_rows(
    738     exec_plan, conn, v_min=self.version, rowids=rowids, abort_on_exc=abort_on_exc)
    739 result.num_rows = num_rows
    740 result.num_excs = num_excsFile ~/Dropbox/workspace/pixeltable/pixeltable/pixeltable/store.py:334, in StoreBase.insert_rows(self, exec_plan, conn, v_min, show_progress, rowids, abort_on_exc)
    332 if abort_on_exc and row.has_exc():
    333     exc = row.get_first_exc()
—> 334     raise exc
    336 rowid = (next(rowids),) if rowids is not None else row.pk[:-1]
    337 pk = rowid + (v_min,)File ~/Dropbox/workspace/pixeltable/pixeltable/pixeltable/exprs/column_ref.py:159, in ColumnRef.eval(self, data_row, row_builder)
    156     return
    158 try:
—> 159     self.col.col_type.validate_media(data_row.file_paths[unvalidated_slot_idx])
    160     # access the value only after successful validation
    161     val = data_row[unvalidated_slot_idx]File ~/Dropbox/workspace/pixeltable/pixeltable/pixeltable/type_system.py:906, in VideoType.validate_media(self, val)
    904             raise excs.Error(f’Not a valid video: {val}’)
    905 except av.AVError:
—> 906     raise excs.Error(f’Not a valid video: {val}’) from NoneError: Not a valid video: /var/folders/hb/qd0dztsj43j_mdb6hbl1gzyc0000gn/T/tmp3djgfyjp.mp4

v.insert(
    [{'video': prefix + 'bad_path.mp4'}, {'video': corrupted_path}],
    on_error='ignore',
)

Computing cells: 100%|████████████████████████████████████████████| 4/4 [00:00<00:00, 20.98 cells/s]
Inserting rows into `videos`: 2 rows [00:00, 671.63 rows/s]
Computing cells: 100%|████████████████████████████████████████████| 4/4 [00:00<00:00, 20.13 cells/s]
Inserted 2 rows with 4 errors across 2 columns (videos.video, videos.None).
UpdateStatus(num_rows=2, num_computed_values=4, num_excs=4, updated_cols=[], cols_with_excs=[‘videos.video’, ‘videos.None’])

v.select(v.video == None, v.video.errortype, v.video.errormsg).collect()

v.where(v.video.errortype != None).select(v.video.errormsg).collect()

​

Accessing the original file paths

v.select(v.video.fileurl, v.video.localpath).collect()

​

Ship it as an app

import pixeltable as pxt

TableModel = pxt.model_base()


class Videos(TableModel, name='videos'):
    video: pxt.Video

pxt schema update app.py external_data