# core


<!-- WARNING: THIS FILE WAS AUTOGENERATED! DO NOT EDIT! -->

> This library follows the [fastai style
> guide](https://docs.fast.ai/dev/style.html), and is crafted with
> [nbdev](https://nbdev.fast.ai/).

This is a general purpose library that allows you to use a VLM (Vision
Language Model) to thoroughly describe the contents of a video, with
subtitles included for additional context.

In essence, this library provides a second set of eyes.

The output is a database object containing the video description.

If you want to directly run this notebook, you want to have an
`OPENROUTER_API_KEY` set.

### Preface

This is a library that allows you to thoroughly describe what occurs in
a video.

**Concisely:**

- A VLM describes the video frame by frame. Each time, it is provided
  with an empty history together with the frame, the subtitle, and
  instructions about what exactly to do to describe the frame. **If one
  is processing a 5 minute video with a sample rate of 1 frame per
  second, the output is 300 individual, isolated frame descriptions.**
- An LLM then takes all the isolated frame descriptions and pieces them
  together to form a summary/description of the video. **The user could,
  for instance, piece together a video description whos summary windows
  consist of 60 frames (in which case, when the LLM produces the summary
  of the next window, it will keep the previous windows in its chat
  history). Or, the user could piece toether an overall video
  description consisting of a single window comprising of 300 frames.**

The biggest beneficiary of this approach is LLM context. Traditional VLM
description systems keep the image in the chat history. Images are token
heavy. Storing the desription of the image, rather than the image
itself, saves the necessary information whilst allowing higher
definition: you can describe videos at 1 frame per second, or even lower
if you desire so.

**More concretely, this library works as follows:**

1.  Set up a database to store the data.
2.  Load the videos and their frames into the database.
3.  Allow a VLM to describe the frames.
4.  Allow a LLM to piece together a description.

**And at an even lower level, as follows:**

1.  Set up a database consisting of 4 tables.
    - A `video` table to store metadata about your videos
    - A `frame` table to store metadata about the frames in each of your
      videos
    - A `run` table to store metadata about each description process
    - A `runframe` table to store metadata about each described frame
2.  Populate the `video` and `frame` tables
3.  Define the VLM settings
4.  Process the frames through the VLM, storing the frame descriptions
    in `runframe`
5.  Process the resulting descriptions through the LLM, storing the
    resulting summary in `video`

What follows is an exposition of the source code. You’ll typically see
source code written first, and then some exposition afterward.

## Database

<details open class="code-fold">
<summary>Exported source</summary>

``` python
from fastcore.all import *
```

</details>

A database is needed to store information about the:

- videos
- frames each video
- each deployed run
- and the descriptions made for each frame in a run.

These classes hold the definition of the tables.

<details open class="code-fold">
<summary>Exported source</summary>

``` python
from fastlite import *
```

</details>

I first need a database object. For that, I’ll be using
[fastlite](https://fastlite.answer.ai/) which is a wrapper over
[sqlite-utils](https://sqlite-utils.datasette.io/en/stable/).

``` python
!rm db.db
db = database('db.db'); db
```

    <Database <apsw.Connection "/app/data/vlm-monitor/nbs/db.db">>

``` python
videos = db.create(Video, transform=True); print(videos.schema)
```

    CREATE TABLE [video] (
       [id] INTEGER PRIMARY KEY,
       [title] TEXT,
       [overview] TEXT,
       [description] TEXT,
       [transcript] TEXT,
       [length] INTEGER,
       [sample_rate] INTEGER,
       [path] TEXT
    )

I’m now creating the tables from the classes I’ve defined.

While I can view the schema, I’ll add some markdown highlighting to make
things easier to visually distinguish. sqlite-utils tables inherit from
`Queryable`. So I’ll patch its `.schema` method with fastcore’s `hl_md`
function.

``` python
??Queryable.schema
```

``` python
@property
def schema(self) -> str:
    "SQL schema for this table or view."
    return self.db.execute(
        "select sql from sqlite_master where name = ?", (self.name,)
    ).fetchone()[0]
```

**File:** `/usr/local/lib/python3.12/site-packages/apswutils/db.py`;
line: 1359

``` python
?hl_md
```

``` python
def hl_md(
    s, lang:str='html', show:bool=True
):
    "Syntax highlight `s` using `lang`."
```

**File:** `~/.local/lib/python3.12/site-packages/fastcore/xtras.py`;
line: 1098

**Type:** function

------------------------------------------------------------------------

### Queryable.schema

``` python
def schema()->str:
```

*SQL schema for this table or view.*

<details open class="code-fold">
<summary>Exported source</summary>

``` python
@patch(as_prop=True)
def schema(self:Queryable) -> str:
    "SQL schema for this table or view."
    return hl_md(self.db.execute(
        "select sql from sqlite_master where name = ?", (self.name,)
    ).fetchone()[0], lang='sql')
```

</details>

Things should now look much visually nicer.

``` python
videos.schema
```

``` sql
CREATE TABLE [video] (
   [id] INTEGER PRIMARY KEY,
   [title] TEXT,
   [overview] TEXT,
   [description] TEXT,
   [transcript] TEXT,
   [length] INTEGER,
   [sample_rate] INTEGER,
   [path] TEXT
)
```

I’ll create the remaining tables.

``` python
frames = db.create(Frame, transform=True, foreign_keys=[('video_id', 'video', 'id')]); frames.schema
```

``` sql
CREATE TABLE [frame] (
   [id] INTEGER PRIMARY KEY,
   [video_id] INTEGER REFERENCES [video]([id]) ON UPDATE CASCADE ON DELETE CASCADE,
   [frame_number] INTEGER,
   [subtitle] TEXT
)
```

``` python
runs = db.create(Run, transform=True); runs.schema
```

``` sql
CREATE TABLE [run] (
   [id] INTEGER PRIMARY KEY,
   [deploy_time] TEXT,
   [finish_time] TEXT,
   [start_time] TEXT,
   [total_duration] TEXT,
   [video_id] INTEGER,
   [model] TEXT,
   [usage] TEXT,
   [num_frames] INTEGER,
   [start_sec] INTEGER,
   [end_sec] INTEGER,
   [step] INTEGER,
   [description] TEXT
)
```

``` python
runframes = db.create(RunFrame, pk=['run_id', 'frame_id', 'type'], foreign_keys=[('run_id', 'run', 'id'), ('frame_id', 'frame', 'id')], transform=True); runframes.schema
```

``` sql
CREATE TABLE [run_frame] (
   [run_id] INTEGER REFERENCES [run]([id]) ON UPDATE CASCADE ON DELETE CASCADE,
   [frame_id] INTEGER REFERENCES [frame]([id]) ON UPDATE CASCADE ON DELETE CASCADE,
   [type] TEXT,
   [system_prompt] TEXT,
   [prompt] TEXT,
   [description] TEXT,
   [usage] TEXT,
   [skipped] INTEGER,
   PRIMARY KEY ([run_id], [frame_id], [type])
)
```

------------------------------------------------------------------------

### init_db

``` python
def init_db(
    path:str | pathlib.Path='db.db', # Path to database
)->Database:
```

*Initialize a database and return it.*

<details open class="code-fold">
<summary>Exported source</summary>

``` python
from apswutils.db import Database
```

</details>

<details open class="code-fold">
<summary>Exported source</summary>

``` python
def init_db(
    path:str|Path='db.db' # Path to database
) -> Database:
    "Initialize a database and return it."
    db = database(path)
    db.create(Video, transform=True)
    db.create(Frame, transform=True, foreign_keys=[('video_id', 'video', 'id')])
    db.create(Run, transform=True)
    db.create(RunFrame, pk=['run_id', 'frame_id', 'type'], foreign_keys=[('run_id', 'run', 'id'), ('frame_id', 'frame', 'id')], transform=True)
    for t in db.t: t.dataclass()
    return db
```

</details>

I’ve now wrapped everything together that initializes the database in a
single function.

``` python
!rm db.db
db = init_db(); db
```

    <Database <apsw.Connection "/app/data/vlm-monitor/nbs/db.db">>

``` python
db.t.video.schema
```

``` sql
CREATE TABLE [video] (
   [id] INTEGER PRIMARY KEY,
   [title] TEXT,
   [overview] TEXT,
   [description] TEXT,
   [transcript] TEXT,
   [length] INTEGER,
   [sample_rate] INTEGER,
   [path] TEXT
)
```

------------------------------------------------------------------------

### view_col

``` python
def view_col(
    table:Table, # Database table
    col:str, # Column to view
    where:str | None=None, # SQL lookup statement
    where_args:str | None=None, # SQL lookup statement arguments
)->list: # List of rows
```

*Return all rows of a given column in table, optionally setting
`where`.*

<details open class="code-fold">
<summary>Exported source</summary>

``` python
from operator import attrgetter, itemgetter
```

</details>

<details open class="code-fold">
<summary>Exported source</summary>

``` python
def view_col(
    table:Table, # Database table
    col:str, # Column to view
    where:str|None=None, # SQL lookup statement
    where_args:str|None=None # SQL lookup statement arguments
)->list: # List of rows
    "Return all rows of a given column in table, optionally setting `where`."
    if where is None: return L(table()).map(attrgetter(col))
    else: return L(table(where=where, where_args=where_args)).map(attrgetter(col))
```

</details>

I’ve also defined a little helper function here that makes it easier to
view all rows of a given column in a table.

## Populate

In this section, I write the logic that will populate the `video` and
`frame` tables in one function.

For this to work, the following assumptions are made: - the provided
directory is flat, - all folders contain the video frames, and - the
same folders contain the respective video’s transcript.

``` python
!tree ../../data/timss -L 1
```

    ../../data/timss
    ├── M-AU1
    ├── M-AU2
    ├── M-AU3
    ├── M-AU4
    ├── M-CZ1
    ├── M-CZ2
    ├── M-CZ3
    ├── M-CZ4
    ├── M-HK1
    ├── M-HK2
    ├── M-HK3
    ├── M-HK4
    ├── M-JP1
    ├── M-JP2
    ├── M-JP3
    ├── M-JP4
    ├── M-NL1
    ├── M-NL2
    ├── M-NL3
    ├── M-NL4
    ├── M-SW1
    ├── M-SW2
    ├── M-SW3
    ├── M-SW4
    ├── M-US1
    ├── M-US2
    ├── M-US3
    ├── M-US4
    ├── S-AU1
    ├── S-AU2
    ├── S-AU3
    ├── S-AU4
    ├── S-AU5
    ├── S-CZ1
    ├── S-CZ2
    ├── S-CZ3
    ├── S-CZ4
    ├── S-CZ5
    ├── S-JP1
    ├── S-JP2
    ├── S-JP3
    ├── S-JP4
    ├── S-JP5
    ├── S-NL1
    ├── S-NL2
    ├── S-NL3
    ├── S-NL4
    ├── S-NL5
    ├── S-US1
    ├── S-US3
    ├── S-US4
    └── S-US5

    53 directories, 0 files

``` python
dpath = Path('../../data/timss/'); dpath.ls()
```

    [Path('../../data/timss/M-CZ3'), Path('../../data/timss/M-AU2'), Path('../../data/timss/M-CZ4'), Path('../../data/timss/M-JP2'), Path('../../data/timss/S-AU4'), Path('../../data/timss/S-US3'), Path('../../data/timss/S-CZ1'), Path('../../data/timss/M-AU1'), Path('../../data/timss/S-AU5'), Path('../../data/timss/S-US4'), Path('../../data/timss/S-NL2'), Path('../../data/timss/S-CZ5'), Path('../../data/timss/S-AU2'), Path('../../data/timss/S-JP5'), Path('../../data/timss/M-NL2'), Path('../../data/timss/M-AU3'), Path('../../data/timss/S-US5'), Path('../../data/timss/M-SW2'), Path('../../data/timss/M-HK4'), Path('../../data/timss/S-AU3'), Path('../../data/timss/M-HK2'), Path('../../data/timss/S-NL5'), Path('../../data/timss/S-CZ4'), Path('../../data/timss/.DS_Store'), Path('../../data/timss/M-NL4'), Path('../../data/timss/M-CZ1'), Path('../../data/timss/S-JP2'), Path('../../data/timss/M-NL1'), Path('../../data/timss/M-US3'), Path('../../data/timss/M-SW3'), Path('../../data/timss/M-US1'), Path('../../data/timss/M-JP4'), Path('../../data/timss/M-US2'), Path('../../data/timss/M-JP3'), Path('../../data/timss/M-CZ2'), Path('../../data/timss/M-HK3'), Path('../../data/timss/S-JP1'), Path('../../data/timss/S-CZ3'), Path('../../data/timss/S-US1'), Path('../../data/timss/M-SW4'), Path('../../data/timss/M-HK1'), Path('../../data/timss/M-JP1'), Path('../../data/timss/S-NL3'), Path('../../data/timss/M-SW1'), Path('../../data/timss/M-AU4'), Path('../../data/timss/S-NL4'), Path('../../data/timss/S-NL1'), Path('../../data/timss/S-AU1'), Path('../../data/timss/S-JP4'), Path('../../data/timss/S-JP3'), Path('../../data/timss/M-NL3'), Path('../../data/timss/M-US4'), Path('../../data/timss/S-CZ2')]

I can see that my data has some unneccessary files, such as `.DS_Store`.

------------------------------------------------------------------------

### filter_paths

``` python
def filter_paths(
    paths:list, # List of paths to filter
    chs:str='.', # Characters to check for in the component
    comp:str='stem', # Path attribute to inspect (e.g. 'stem', 'name')
    negate:bool=True, # If True, exclude paths whose `comp` contains `chs`; if False, keep only those
)->list: # Filtered list of paths
```

*Filter paths by whether a path component contains specified
characters.*

<details open class="code-fold">
<summary>Exported source</summary>

``` python
def filter_paths(
    paths:list[Path], # List of paths to filter
    chs:str='.', # Characters to check for in the component
    comp:str='stem', # Path attribute to inspect (e.g. 'stem', 'name')
    negate:bool=True, # If True, exclude paths whose `comp` contains `chs`; if False, keep only those
)->list[Path]: # Filtered list of paths
    "Filter paths by whether a path component contains specified characters."
    return paths.filter(~getattr(Self, comp).count(chs), negate=negate)
```

</details>

I’ve written a function that both filters out those unnecessary files,
at the same time also filters in the necessary ones!

``` python
dpaths = filter_paths(dpath.ls()).sorted(lambda o: (o.stem[:-1], o.stem[-1])); dpaths
```

    [Path('../../data/timss/M-AU1'), Path('../../data/timss/M-AU2'), Path('../../data/timss/M-AU3'), Path('../../data/timss/M-AU4'), Path('../../data/timss/M-CZ1'), Path('../../data/timss/M-CZ2'), Path('../../data/timss/M-CZ3'), Path('../../data/timss/M-CZ4'), Path('../../data/timss/M-HK1'), Path('../../data/timss/M-HK2'), Path('../../data/timss/M-HK3'), Path('../../data/timss/M-HK4'), Path('../../data/timss/M-JP1'), Path('../../data/timss/M-JP2'), Path('../../data/timss/M-JP3'), Path('../../data/timss/M-JP4'), Path('../../data/timss/M-NL1'), Path('../../data/timss/M-NL2'), Path('../../data/timss/M-NL3'), Path('../../data/timss/M-NL4'), Path('../../data/timss/M-SW1'), Path('../../data/timss/M-SW2'), Path('../../data/timss/M-SW3'), Path('../../data/timss/M-SW4'), Path('../../data/timss/M-US1'), Path('../../data/timss/M-US2'), Path('../../data/timss/M-US3'), Path('../../data/timss/M-US4'), Path('../../data/timss/S-AU1'), Path('../../data/timss/S-AU2'), Path('../../data/timss/S-AU3'), Path('../../data/timss/S-AU4'), Path('../../data/timss/S-AU5'), Path('../../data/timss/S-CZ1'), Path('../../data/timss/S-CZ2'), Path('../../data/timss/S-CZ3'), Path('../../data/timss/S-CZ4'), Path('../../data/timss/S-CZ5'), Path('../../data/timss/S-JP1'), Path('../../data/timss/S-JP2'), Path('../../data/timss/S-JP3'), Path('../../data/timss/S-JP4'), Path('../../data/timss/S-JP5'), Path('../../data/timss/S-NL1'), Path('../../data/timss/S-NL2'), Path('../../data/timss/S-NL3'), Path('../../data/timss/S-NL4'), Path('../../data/timss/S-NL5'), Path('../../data/timss/S-US1'), Path('../../data/timss/S-US3'), Path('../../data/timss/S-US4'), Path('../../data/timss/S-US5')]

The way I’ve defined `filter_paths` means I can filter for any file.
I’ll filter for the transcripts and take a look inside one of them.

``` python
dp = dpaths[0]; dp
```

    Path('../../data/timss/M-AU1')

``` python
tr = filter_paths(dp.ls(), chs='txt', comp='suffix', negate=False)[0]
print(tr.read_text()[:500])
```

    1
    00:00:20,000 --> 00:00:42,980
    [T] I'm wired.

    2
    00:00:43,000 --> 00:00:53,980
    [T] It's running.

    3
    00:00:54,000 --> 00:00:56,980
    [SN] (inaudible) please turn off the air.

    4
    00:00:57,000 --> 00:01:06,980
    [T] I'll make it a bit warmer.

    5
    00:01:07,000 --> 00:01:16,980
    [T] Well not really, but if you really have to I suppose. Okay.

    6
    00:01:17,000 --> 00:01:20,980
    [SN] (inaudible) go out to my locker and get my maths book?

    7
    00:01:21,000 --> 00:01:22,980
    [T] Oh, you won't need it today.

    8
    00:0

The transcripts are in SRT format. I’ll be converting them to TSV,
making them token efficient. To do that, I’ll need a way to convert all
timestamps to seconds.

``` python
f'45 seconds is {int(45)*60**1} seconds'
```

    '45 seconds is 2700 seconds'

``` python
f'45 minutes is {int(45)*60**2} seconds'
```

    '45 minutes is 162000 seconds'

``` python
f'45 hours is {int(45)*60**3} seconds'
```

    '45 hours is 9720000 seconds'

``` python
'01:50'.split(':')
```

    ['01', '50']

``` python
L(enumerate(reversed('01:50'.split(':'))))
```

    [(0, '50'), (1, '01')]

``` python
sum(int(x)*60**i for i,x in enumerate(reversed('01:50'.split(':'))))
```

    110

------------------------------------------------------------------------

### time2sec

``` python
def time2sec(
    time:str, # Time string in HH:MM:SS or MM:SS format
)->int: # Total seconds
```

*Convert a time string to total seconds.*

<details open class="code-fold">
<summary>Exported source</summary>

``` python
import re
```

</details>

<details open class="code-fold">
<summary>Exported source</summary>

``` python
def time2sec(
    time:str, # Time string in HH:MM:SS or MM:SS format
)->int: # Total seconds
    "Convert a time string to total seconds."
    return sum(int(x)*60**i for i,x in enumerate(reversed(time.split(':'))))
```

</details>

``` python
time2sec('1:50'), time2sec('00:01:30')
```

    (110, 90)

With that sorted, I can now write the function that will convert from
SRT to TSV.

------------------------------------------------------------------------

### srt2tsv

``` python
def srt2tsv(
    srt:str, # SRT subtitle text to parse
)->str: # TSV string with columns: timerange, speaker, text
```

*Convert SRT format to TSV string.*

<details open class="code-fold">
<summary>Exported source</summary>

``` python
def srt2tsv(
    srt:str, # SRT subtitle text to parse
)->str: # TSV string with columns: timerange, speaker, text
    "Convert SRT format to TSV string."
    blocks = re.split(r'\n\n+', srt.strip())
    res = ''
    for b in blocks:
        lines = b.strip().splitlines()
        if len(lines)<3: continue
        parts = lines[1].split(' --> ')
        start = time2sec(parts[0].replace(',', '')[:-3])
        end = time2sec(parts[1].replace(',', '')[:-3])
        m = re.match(r'\[(.*?)\]\s*(.*)', lines[2]); spk,text = (m[1], m[2]) if m else ('', lines[2])
        res += f"{start}→{end}\t{spk}\t{text}\n"
    return res
```

</details>

``` python
print(srt2tsv(tr.read_text())[:500])
```

    20→42   T   I'm wired.
    43→53   T   It's running.
    54→56   SN  (inaudible) please turn off the air.
    57→66   T   I'll make it a bit warmer.
    67→76   T   Well not really, but if you really have to I suppose. Okay.
    77→80   SN  (inaudible) go out to my locker and get my maths book?
    81→82   T   Oh, you won't need it today.
    83→97   SN  Oh, okay.
    98→98   T   Are you going to be sitting down?
    99→100  SN  I haven't (inaudible)
    101→102 T   Oh, then you've got to go outside then.
    103→105 S   (inaudible) books (inaudible)
    106→135 T   There'll be some

I can now try create an entry in the `video` table.

``` python
dt = db.t; dt
```

    frame, run, run_frame, video

``` python
for t in dt: t.dataclass()
```

``` python
v = dt.video.insert(title=dp.stem, transcript=srt2tsv(tr.read_text()),
                   length=len(dp.ls()), path=str(dp), sample_rate=1)
type(v), v.transcript[:50]
```

    (fastlite.core.Video,
     "20→42\tT\tI'm wired.\n43→53\tT\tIt's running.\n54→56\tSN\t")

Now I want to attempt creating some entries in the `frame` table. For
that, I’ll have to fetch the corresponding subtitle for each frame.

``` python
db.t.frame.schema
```

``` sql
CREATE TABLE [frame] (
   [id] INTEGER PRIMARY KEY,
   [video_id] INTEGER REFERENCES [video]([id]) ON UPDATE CASCADE ON DELETE CASCADE,
   [frame_number] INTEGER,
   [subtitle] TEXT
)
```

``` python
dp
```

    Path('../../data/timss/M-AU1')

``` python
dp.ls()[:5]
```

    [Path('../../data/timss/M-AU1/frame_001886.jpg'), Path('../../data/timss/M-AU1/frame_001303.jpg'), Path('../../data/timss/M-AU1/frame_001378.jpg'), Path('../../data/timss/M-AU1/frame_002173.jpg'), Path('../../data/timss/M-AU1/frame_002259.jpg')]

``` python
filter_paths(dp.ls(), chs='.', comp='stem')[:3]
```

    [Path('../../data/timss/M-AU1/frame_001886.jpg'), Path('../../data/timss/M-AU1/frame_001303.jpg'), Path('../../data/timss/M-AU1/frame_001378.jpg')]

``` python
filter_paths(dp.ls(), chs='txt', comp='suffix')[:3]
```

    [Path('../../data/timss/M-AU1/frame_001886.jpg'), Path('../../data/timss/M-AU1/frame_001303.jpg'), Path('../../data/timss/M-AU1/frame_001378.jpg')]

To connect each frame to their corresponding transcript, I’ll be
converting the TSV transcripts to Python dictionaries. If the video had
a subtitle between 3 seconds and 7 seconds. Then all frames sampled in
that range should have the corresponding same subtitle.

------------------------------------------------------------------------

### tsv2dict

``` python
def tsv2dict(
    tsv:str, # TSV string with columns: timerange, speaker, text
    max_len:int, # Max frame number to cover
    default:str='', # Value for timestamps without subtitles
)->dict: # {second: "speaker: text"} lookup dict
```

*Convert TSV string to a {second: subtitle} lookup dict.*

<details open class="code-fold">
<summary>Exported source</summary>

``` python
def tsv2dict(
    tsv:str, # TSV string with columns: timerange, speaker, text
    max_len:int, # Max frame number to cover
    default:str='', # Value for timestamps without subtitles
)->dict: # {second: "speaker: text"} lookup dict
    "Convert TSV string to a {second: subtitle} lookup dict."
    lookup = {}
    for line in tsv.splitlines():
        if not line.strip(): continue
        tr, spk, text = line.split('\t')
        start, end = tr.split('→')
        start, end = int(start), int(end)
        entry = f"{spk}: {text}" if spk else text
        for s in range(start, end+1): lookup[s] = entry
    end = max(max(lookup) if lookup else 0, max_len)
    for s in range(end+1): lookup.setdefault(s, default)
    return lookup
```

</details>

``` python
r = srt2tsv(tr.read_text())
```

Double checking whether I got what I wanted…

``` python
from itertools import islice
```

``` python
list(islice(tsv2dict(r, 2000).items(), 10))
```

    [(20, "T: I'm wired."),
     (21, "T: I'm wired."),
     (22, "T: I'm wired."),
     (23, "T: I'm wired."),
     (24, "T: I'm wired."),
     (25, "T: I'm wired."),
     (26, "T: I'm wired."),
     (27, "T: I'm wired."),
     (28, "T: I'm wired."),
     (29, "T: I'm wired.")]

``` python
lookup = tsv2dict(srt2tsv(tr.read_text()), len(dp.ls()))
lookup[20], lookup[100]
```

    ("T: I'm wired.", "SN: I haven't (inaudible)")

I can go ahead an produce an entry.

``` python
fpaths = filter_paths(Path(v.path).ls(), chs='txt', comp='suffix')
fpaths = filter_paths(fpaths, chs='.', comp='stem').sorted(key=~Self.stem.split('_'))
fpaths[:5]
```

    [Path('../../data/timss/M-AU1/frame_000001.jpg'), Path('../../data/timss/M-AU1/frame_000002.jpg'), Path('../../data/timss/M-AU1/frame_000003.jpg'), Path('../../data/timss/M-AU1/frame_000004.jpg'), Path('../../data/timss/M-AU1/frame_000005.jpg')]

``` python
fp = fpaths[0]; fp
```

    Path('../../data/timss/M-AU1/frame_000001.jpg')

``` python
fnum = int(fp.stem.split('_')[1]); fnum
```

    1

``` python
db.t.frame.insert(video_id=v.id, frame_number=fnum, subtitle=tsv2dict(v.transcript, v.length)[fnum])
```

    Frame(id=1, video_id=1, frame_number=1, subtitle='')

I can wrap this all up into a function that will perform this on all
videos and all frames.

------------------------------------------------------------------------

### populate_db

``` python
def populate_db(
    db:Database, # Database to populate
    paths:list, # List of video directories
    sample_rate:int=1, # Sampling rate for frames
    trans_suffix:str='txt', # Transcript file suffix
)->None:
```

*Populate video and frame tables from a list of video directories.*

<details open class="code-fold">
<summary>Exported source</summary>

``` python
from fastprogress.fastprogress import master_bar, progress_bar
```

</details>

<details open class="code-fold">
<summary>Exported source</summary>

``` python
def populate_db(
    db:Database, # Database to populate
    paths:list[Path], # List of video directories
    sample_rate:int=1, # Sampling rate for frames
    trans_suffix:str='txt', # Transcript file suffix
)->None:
    "Populate video and frame tables from a list of video directories."
    t = db.t
    filter_trans = partial(filter_paths, chs=trans_suffix, comp='suffix', negate=False)
    filter_dots = partial(filter_paths, chs='.', comp='stem')
    for p in (mb:=master_bar(paths)):
        mb.main_bar.comment = f'video {p.stem}'
        tr = filter_trans(p.ls())[0]
        v = t.video.insert(title=p.stem, transcript=srt2tsv(tr.read_text()),
                           length=len(p.ls()), path=str(p), sample_rate=sample_rate)
        lookup = tsv2dict(v.transcript, v.length)
        fpaths = filter_dots(filter_paths(Path(v.path).ls(), chs=trans_suffix, comp='suffix')).sorted(key=~Self.stem.split('_'))
        for fp in (pb:=progress_bar(fpaths, parent=mb)):
            fnum = int(fp.stem.split('_')[1])
            t.frame.insert(video_id=v.id, frame_number=fnum, subtitle=lookup[fnum])
```

</details>

``` python
!rm db.db
db = init_db()
populate_db(db, dpaths)
```

``` python
len(db.t.video()), len(db.t.frame())
```

    (52, 149880)

## VLM

In this section I create helper functions for working with the LLM,
using the [fastllm](https://github.com/AnswerDotAI/fastllm) library.
fastllm is still in active development, and the helpers here work for
version 0.0.36.

<details open class="code-fold">
<summary>Exported source</summary>

``` python
from aidialog.msg_parts import Msg, Part, PartType
```

</details>

``` python
?Msg
```

``` python
def Msg(
    role:str, content:List
)->None:
    "A normalized message."
```

**File:**
`/usr/local/lib/python3.12/site-packages/aidialog/msg_parts.py`; line:
55

**Type:** type

``` python
?Part
```

``` python
def Part(
    type:str, text:str=None, data:dict=None
)->None:
    "A normalized content part."
```

**File:**
`/usr/local/lib/python3.12/site-packages/aidialog/msg_parts.py`; line:
21

**Type:** type

``` python
?PartType
```

``` python
def PartType(
    *args, **kwds
):
```

**File:**
`/usr/local/lib/python3.12/site-packages/aidialog/msg_parts.py`

**Type:** EnumType

------------------------------------------------------------------------

### user

``` python
def user(
    txt:str, # User message text
    img:str | None=None, # Base64 data URL of image to include
)->Msg: # User message with optional image
```

*Build a user message with optional image.*

<details open class="code-fold">
<summary>Exported source</summary>

``` python
def user(
    txt:str, # User message text
    img:str|None=None, # Base64 data URL of image to include
)->Msg: # User message with optional image
    "Build a user message with optional image."
    if img is None: return Msg(role='user', content=[Part(PartType.text, text=txt)])
    else: return Msg(role='user', content=[Part(PartType.input_image, text=img), Part(PartType.text, text=txt)])
```

</details>

``` python
user('你好')
```

**Msg**

- role: `user`

<contents>

**Part** (`text`)

你好

<details>

- data: `None`

</details>

</contents>

------------------------------------------------------------------------

### assistant

``` python
def assistant(
    txt:str, # Assistant message text
    citations:list=None, # Optional citation list
)->Msg: # Assistant message
```

*Build an assistant message.*

<details open class="code-fold">
<summary>Exported source</summary>

``` python
def assistant(
    txt:str, # Assistant message text
    citations:list=None, # Optional citation list
)->Msg: # Assistant message
    "Build an assistant message."
    return Msg(role='assistant', content=[Part(PartType.text, text=txt, data={'citations': citations} if citations else None)])
```

</details>

``` python
assistant('嗨')
```

**Msg**

- role: `assistant`

<contents>

**Part** (`text`)

嗨

<details>

- data: `None`

</details>

</contents>

<details open class="code-fold">
<summary>Exported source</summary>

``` python
from fastllm.acomplete import acomplete
```

</details>

``` python
?acomplete
```

``` python
async def acomplete(
    msgs, model, api_name:NoneType=None, vendor_name:NoneType=None, api_key:NoneType=None, base_url:NoneType=None,
    xtra_body:NoneType=None, xtra_hdrs:NoneType=None, stream:bool=False, stop_callables:NoneType=None, retries:int=2,
    retry_delay:float=0.5, system:NoneType=None, max_tokens:NoneType=None, temperature:NoneType=None,
    tools:NoneType=None, tool_choice:NoneType=None, reasoning_effort:NoneType=None, web_search_options:NoneType=None
):
    "Unified completion across different APIs."
```

**File:**
`/usr/local/lib/python3.12/site-packages/fastllm/acomplete.py`; line:
170

**Type:** function

<details open class="code-fold">
<summary>Exported source</summary>

``` python
from cachy import enable_cachy, disable_cachy
```

</details>

[Cachy](https://github.com/AnswerDotAI/cachy) caches http requests and
stores the responses locally. This reduces spend and avoids the need to
also repeatedly wait for the remote server to process my request.

I’ll now send a test query.

``` python
enable_cachy()
await acomplete([user('hi')], 'deepseek-v4-flash', vendor_name='deepseek')
```

<details>

<summary>

Thinking
</summary>

好的，用户只发了一个“hi”，这是非常简单的打招呼。用户可能刚进入对话，想测试我是否在线或者开始一个友好的交流。深层需求应该是希望得到热情、友好的回应，开启一次对话。我不需要复杂分析，直接礼貌问候并表达乐于助人的态度，用开放式的邀请让用户提出具体问题。想到了用“你好！”开头，加上表情符号显得亲切，然后自我介绍并说明能力范围，最后用提问引导对话继续。

</details>

你好！很高兴见到你！😊

有什么我可以帮你的吗？无论是回答问题、帮你整理信息、提供创作灵感，还是聊聊天，我都很乐意陪你一起。你只需告诉我需求，剩下的交给我！

<details>

- model: `deepseek-v4-flash`
- finish_reason: `stop`
- usage:
  `Usage(prompt_tokens=5, completion_tokens=143, total_tokens=148, cached_tokens=0, cache_creation_tokens=0, reasoning_tokens=97, raw={'prompt_tokens': 5, 'completion_tokens': 143, 'total_tokens': 148, 'prompt_tokens_details': {'cached_tokens': 0}, 'completion_tokens_details': {'reasoning_tokens': 97}, 'prompt_cache_hit_tokens': 0, 'prompt_cache_miss_tokens': 5})`

</details>

<details open class="code-fold">
<summary>Exported source</summary>

``` python
from fastllm.types import Completion
```

</details>

I’m now creating a function that’ll stream in the response with the
`Completion` object. We’ll see the response here as it generates.

------------------------------------------------------------------------

### stream

``` python
async def stream(
    msgs:list | None=None, # Messages to send
    model:str='', # Model name (e.g. 'deepseek-v4-flash')
    max_think:float=inf, # Max thinking tokens to display
    usage:bool=True, # Show usage info in output
    display:bool=True, # Print text/thinking as it arrives
    **kwargs
)->Completion: # Return the final completion
```

*Stream a response, printing text/thinking as it arrives. Returns the
final completion.*

<details open class="code-fold">
<summary>Exported source</summary>

``` python
async def stream(
    msgs:list|None=None,  # Messages to send
    model:str='',  # Model name (e.g. 'deepseek-v4-flash')
    max_think:float=float('inf'),  # Max thinking tokens to display
    usage:bool=True,  # Show usage info in output
    display:bool=True,  # Print text/thinking as it arrives
    **kwargs,  # Passed to acomplete
) -> Completion:  # Return the final completion
    "Stream a response, printing text/thinking as it arrives. Returns the final completion."
    assert msgs is not None, 'no messages provided'
    assert model!='', 'no model name provided'
    think_cnt, seen_txt = 0, False
    async for o in await acomplete(msgs, model, stream=True, **kwargs):
        if not isinstance(o, Completion) and display:
            if isinstance(o, Part) and o.type==PartType.thinking and think_cnt<max_think: print('🤔', end='', flush=True)
            if isinstance(o, Part) and o.type==PartType.text and (txt:=o.text): print(f"{'\n\n' if not seen_txt else ''}{txt}", end='', flush=True) or not seen_txt and (seen_txt:=True)
            think_cnt+=1
    if display: print()
    return o
```

</details>

``` python
r = await stream([user('hi')], 'deepseek-v4-flash', vendor_name='deepseek')
```

------------------------------------------------------------------------

### img2b64

``` python
def img2b64(
    path:Path, # Path to image file
)->str: # Base64 data URL
```

*Encode an image file as a base64 data URL.*

<details open class="code-fold">
<summary>Exported source</summary>

``` python
from base64 import b64encode
```

</details>

<details open class="code-fold">
<summary>Exported source</summary>

``` python
def img2b64(
    path:Path, # Path to image file
)->str: # Base64 data URL
    "Encode an image file as a base64 data URL."
    return 'data:image/png;base64,'+b64encode(Path.read_bytes(path)).decode()
```

</details>

And a helper function to more easily pass images to the VLM.

``` python
r = await stream([user('what do ye elf eyes see', img2b64(Path('./test.jpg')))], 'bytedance-seed/seed-2.0-lite', vendor_name='openrouter', reasoning_effort='high')
```

``` python
r.usage.raw
```

    {'prompt_tokens': 1343,
     'completion_tokens': 702,
     'total_tokens': 2045,
     'cost': 0.00173975,
     'is_byok': False,
     'prompt_tokens_details': {'cached_tokens': 0,
      'cache_write_tokens': 0,
      'audio_tokens': 0,
      'video_tokens': 0},
     'cost_details': {'upstream_inference_cost': 0.00173975,
      'upstream_inference_prompt_cost': 0.00033575,
      'upstream_inference_completions_cost': 0.001404},
     'completion_tokens_details': {'reasoning_tokens': 559,
      'image_tokens': 0,
      'audio_tokens': 0}}

------------------------------------------------------------------------

### session

``` python
def session(
    msgs:list | None=None, model:str='', max_think:float=inf, usage:bool=True, display:bool=True, **kwargs
):
```

*Create a stream partial with preset model/kwargs.*

<details open class="code-fold">
<summary>Exported source</summary>

``` python
@delegates(stream, keep=True)
def session(
    **kwargs,
):
    "Create a stream partial with preset model/kwargs."
    return partial(stream, **kwargs)
```

</details>

And another helper function to create sessions more easily without
having to write the same params each time.

## Run

Now it’s time to run the VLM across all frames of a video. I’ll begin
small by working on a single frame: fetch its subtitle from the
database, attach the image, and see what comes back.

``` python
frame = db.t.frame.selectone('video_id=? AND frame_number=?', (v.id, int(fp.stem.split('_')[1]))); frame
```

    Frame(id=1, video_id=1, frame_number=1, subtitle='')

``` python
sub = frame.subtitle or '[No speech]'
```

``` python
prompt = f"Describe what you see\n\nSubtitle: {sub}"
r = await stream([user(prompt, img=img2b64(fp))], 'bytedance-seed/seed-2.0-lite', vendor_name='openrouter', reasoning_effort='high'); r.message.text[:200]
```

    'This is a low-resolution (blurry, likely older recorded) scene of a school classroom, with no spoken audio:\nThe space is a standard classroom, with rows of student desks holding seated school-aged chi'

Running a single frame works nicely, but when scaling to hundreds of
frames across a video, problems emerge:

- **Rate limits (429):** the API throttles concurrent requests, so some
  frames will fail
- **Connection timeouts:** the provider may be temporarily unreachable
- **Insufficient credits (402):** if the budget runs out mid-run, we
  need to stop gracefully without losing completed work

To handle these problems, I’ll wrap the single-frame call that catches
`APIError`s, stores successful results to the `run_frame` table via
upsert, and flags any failed frame as skipped for later retry.

<details open class="code-fold">
<summary>Exported source</summary>

``` python
from fastspec.errors import APIError
```

</details>

<details open class="code-fold">
<summary>Exported source</summary>

``` python
async def _process_frame(
    p:Path, # Path to frame image
    db, # Database object
    session, # Stream session partial
    run_id:int, # Run ID to store results under
    video_id:int, # Video ID the frame belongs to
    prompt:str, # Prompt template for the VLM
    prompt_type:str, # Type label for this prompt (e.g. 'description')
    include_subs:bool=True, # Whether to append subtitle to prompt
):
    "Run VLM on a single frame and store the result."
    fnum = int(p.stem.split('_')[1])
    frame = db.t.frame.selectone('video_id=? AND frame_number=?', (video_id, fnum))
    if include_subs:
        sub = frame.subtitle or '[No speech]'
        prompt = f"{prompt}\n\nSubtitle: {sub}"
    try:
        r = await session([user(prompt, img=img2b64(p))])
        u = r.usage.raw if r.usage else {}
        db.t.run_frame.upsert(run_id=run_id, frame_id=frame.id, type=prompt_type, prompt=prompt, description=r.message.text, usage={**u, 'cost': u.get('cost', 0.0)}, skipped=False)
    except Exception as e:
        if not isinstance(e, APIError): raise
        sc = getattr(e,'status_code',None)
        print(f'!! APIError sc={sc} frame {fnum}: {e}')
        if sc==402:
            print(f'!! Insufficient credits — stopped at frame {fnum}')
            raise
        db.t.run_frame.upsert(run_id=run_id, frame_id=frame.id, type=prompt_type, prompt=prompt, description='', skipped=True)
        print(f'!! Skipped: frame {fnum}')
```

</details>

Throughput is a bottleneck. Running frames one at a time across hundreds
of frames would take too long. I’ll query multiple requests in parallel
with fastcore’s `parallel_async_gen`.

``` python
?parallel_async_gen
```

``` python
def parallel_async_gen(
    f, items, *args, n_workers:int=16, pause:int=0, timeout:NoneType=None, return_exceptions:bool=False,
    cancel_on_exit:bool=True, **kwargs
):
    "Yield `(index,result)` pairs as `f` applied to each of `items` completes, in completion order"
```

**File:** `~/.local/lib/python3.12/site-packages/fastcore/parallel.py`;
line: 154

**Type:** function

``` python
s = session(model='bytedance-seed/seed-2.0-lite', vendor_name='openrouter', reasoning_effort='high'); s
```

    functools.partial(<function stream>, model='bytedance-seed/seed-2.0-lite', vendor_name='openrouter', reasoning_effort='high')

``` python
run = db.t.run.insert(deploy_time='', video_id=v.id, model='bytedance-seed/seed-2.0-lite'); run
```

    Run(id=1, deploy_time='', finish_time=None, start_time=None, total_duration=None, video_id=1, model='bytedance-seed/seed-2.0-lite', usage=None, num_frames=None, start_sec=None, end_sec=None, step=None, description=None)

``` python
run_id = run.id
video_id, prompt_type = v.id, 'description'
```

``` python
async for i,r in parallel_async_gen(_process_frame, fpaths[:4], db, s, run_id, video_id, prompt, prompt_type, True, n_workers=4, pause=0.1): print(f'frame {i} done')
view_col(db.t.run_frame, 'description', where='run_id=?', where_args=(run_id,))
```

    frame 0 done

    frame 1 done


    frame 2 done
    frame 3 done

    ['This is a low-resolution, blurry clip filmed from the back of a school classroom, looking toward the front of the room.\nNearly all students, viewed from behind as they face the front of class, are seated at student desks, oriented toward the front of the room. At the front, a male teacher wearing a collared shirt and tie leans over a student at a front desk, interacting with or assisting that child.\nStandard classroom features fill the space: large windows on the left wall let in natural light, a retractable overhead projector screen is mounted to the center wall above a whiteboard, with colorful educational banners hung above the whiteboard. Fluorescent lights are set into the ceiling, and a brick wall section is visible on the right side of the room. There is no speech or audio in this clip.', "This is a quiet scene inside a school classroom, captured from a static in-room camera, with no speech/audio.\nThe space is a standard classroom: student desks and chairs including small green student seating are arranged in rows facing the front of the room. Natural light filters through a large window on the left wall. The front wall has a whiteboard, with small colorful educational posters mounted above it, plus a large rolled-down projection screen to the whiteboard's right; a brick accent wall on the right has a security camera mounted to it, which matches the footage's camera perspective.\n\nAt the front of the room, an adult man who appears to be the teacher, with light hair, a grey long-sleeve shirt, and a dark strap across his chest leans over a front desk, interacting with a seated male student in a white shirt at that desk. A small number of other students are seated at their desks across the room, facing the front of the class: including a blonde girl in a dark top, and another girl with dark hair in a ponytail wearing a white shirt. The space feels calm, like an ongoing quiet classroom session.", 'This is a classroom scene, set in a school:\n1.  At the front of the room, an adult man (likely a teacher, with an ID lanyard, wearing a light grey long-sleeve shirt) stands before a wall of whiteboards plus a pulled-down blank projection screen, positioned to address the class.\n2.  Seated at student desks, all facing front in standard green plastic school chairs, are a small number of students: visible are a blonde girl with a ponytail in a dark top, and several boys in white school uniform shirts, all oriented toward the man at the front.\n3.  Additional classroom details: above the front whiteboards, three colorful decorative educational banners are mounted. A large window on the left side of the frame lets in natural light, next to a white interior door. The right side of the back wall has an exposed brick column, with a security camera mounted to it, and a framed poster hung above the projection screen. This is a small class, with only a handful of students visible in this view.', "This is an elementary school classroom, hosting a small tech demonstration for students:\n1.  At the front of the room, a tall light-haired adult man (wearing a light grey long-sleeve shirt and dark trousers) stands at the head of the class, behind a small robotic arm set up on a student desk. A young boy (seen from behind, in a white school shirt) is at this desk interacting with the robotic arm.\n2.  Other students are seated at grouped classroom desks (with standard green plastic school chairs) all facing the front to watch the demo: a girl with a blonde ponytail in a dark top sits to the boy's left, and another student's head is visible in the bottom right corner, also observing.\n3.  The background has typical classroom features: a back wall with a whiteboard, a pulled-down projector screen to the whiteboard's right, colorful educational banners mounted above the whiteboard, a closed white door on the far left, and a ledge under the whiteboards holding a small supply box."]

Time to wrap that up.

<details open class="code-fold">
<summary>Exported source</summary>

``` python
async def _run_batch(
    frames:L, # List of frame paths to process
    db, # Database object
    session, # Stream session partial
    run_id:int, # Run ID to store results under
    video_id:int, # Video ID the frames belong to
    prompt:str, # Prompt template for the VLM
    prompt_type:str, # Type label for this prompt
    include_subs:bool, # Whether to append subtitle to prompt
    n_workers:int, # Number of parallel workers
    pause:float, # Seconds to pause between dispatches
):
    "Run _process_frame across a batch of frame paths in parallel."
    done = 0
    async for i,r in parallel_async_gen(_process_frame, frames, db, session, run_id, video_id,
                                         prompt, prompt_type, include_subs,
                                         n_workers=n_workers, pause=pause):
        done += 1; print(f'\r{done}/{len(frames)}', end='', flush=True)
    print()
```

</details>

Skipped frames are flagged with `skipped=1`. I’ll query those rows once
again, repeating until they succeed.

I’ll manually mark a couple of frames as skipped.

``` python
for rf in L(db.t.run_frame('run_id=?', (run_id,)))[:2]:
    db.t.run_frame.update(run_id=rf.run_id, frame_id=rf.frame_id, type=rf.type, skipped=True)
L(db.t.run_frame('run_id=? AND skipped=1', (run_id,))).map(attrgetter('frame_id'))
```

    [1, 2]

``` python
rl_rows = L(db.t.run_frame('run_id=? AND skipped=1', (run_id,)))
rl_fids = rl_rows.map(attrgetter('frame_id')); rl_fids
```

    [1, 2]

``` python
ph = ','.join('?'*len(rl_fids)); ph
```

    '?,?'

``` python
rl_frames = L(db.t.frame(f'id IN ({ph})', tuple(rl_fids))); rl_frames
```

    [Frame(id=1, video_id=1, frame_number=1, subtitle=''), Frame(id=2, video_id=1, frame_number=2, subtitle='')]

``` python
rl_paths = rl_frames.map(lambda f: Path(v.path)/f'frame_{f.frame_number:06d}.jpg'); rl_paths
```

    [Path('../../data/timss/M-AU1/frame_000001.jpg'), Path('../../data/timss/M-AU1/frame_000002.jpg')]

<details open class="code-fold">
<summary>Exported source</summary>

``` python
async def _retry_skipped(
    db, # Database object
    video_path:Path, # Path to the video's frame directory
    run_id:int, # Run ID to retry frames for
    video_id:int, # Video ID the frames belong to
    session, # Stream session partial
    prompt:str, # Prompt template for the VLM
    prompt_type:str, # Type label for this prompt
    include_subs:bool, # Whether to append subtitle to prompt
    n_workers:int, # Number of parallel workers
    pause:float, # Seconds to pause between dispatches
    max_retries:int=2, # Max retry attempts for skipped frames
):
    "Retry skipped frames up to max_retries times."
    for attempt in range(max_retries):
        rl_rows = L(db.t.run_frame('run_id=? AND skipped=1', (run_id,)))
        if not rl_rows: break
        rl_fids = rl_rows.map(attrgetter('frame_id'))
        ph = ','.join('?'*len(rl_fids))
        rl_frames = L(db.t.frame(f'id IN ({ph})', tuple(rl_fids)))
        rl_paths = rl_frames.map(lambda f: video_path/f'frame_{f.frame_number:06d}.jpg')
        print(f'!! Retrying {len(rl_paths)} skipped frames (attempt {attempt+1}/{max_retries})')
        try: await _run_batch(rl_paths, db, session, run_id, video_id, prompt, prompt_type, include_subs, n_workers, pause)
        except Exception as e:
            if not isinstance(e, APIError): raise
            print(f'!! APIError in retry: {e}')
            break
```

</details>

With the run complete, I want to see how much it cost. Each `run_frame`
row stores a `usage` JSON string from the API response. I’ll look at
one.

``` python
rows = db.t.run_frame('run_id=? AND skipped=0', (run_id,))
if rows: print(loads(rows[0].usage))
else: print('No successful frames yet')
```

    {'prompt_tokens': 1369, 'completion_tokens': 954, 'total_tokens': 2323, 'cost': 0.00225025, 'is_byok': False, 'prompt_tokens_details': {'cached_tokens': 0, 'cache_write_tokens': 0, 'audio_tokens': 0, 'video_tokens': 0}, 'cost_details': {'upstream_inference_cost': 0.00225025, 'upstream_inference_prompt_cost': 0.00034225, 'upstream_inference_completions_cost': 0.001908}, 'completion_tokens_details': {'reasoning_tokens': 735, 'image_tokens': 0, 'audio_tokens': 0}}

``` python
usgs = L(db.t.run_frame('run_id=? AND skipped=0', (run_id,))).map(lambda r: loads(r.usage))
usgs.map(lambda u: tuple(u.keys())).unique()
```

    [('prompt_tokens', 'completion_tokens', 'total_tokens', 'cost', 'is_byok', 'prompt_tokens_details', 'cost_details', 'completion_tokens_details')]

``` python
sum(u['cost'] for u in usgs)
```

    0.0044785

``` python
sum(u['prompt_tokens'] for u in usgs)
```

    2738

``` python
u = usgs[0]; u['prompt_tokens_details']
```

    {'cached_tokens': 0,
     'cache_write_tokens': 0,
     'audio_tokens': 0,
     'video_tokens': 0}

``` python
sum(u['prompt_tokens_details']['cached_tokens'] for u in usgs)
```

    0

Scalar fields sum directly; nested dict fields need to be unpacked
first. Now I’ll aggregate usage across all frames in a run, and stamp
the run with its finish time and total duration.

<details open class="code-fold">
<summary>Exported source</summary>

``` python
from datetime import datetime
from zoneinfo import ZoneInfo
```

</details>

<details open class="code-fold">
<summary>Exported source</summary>

``` python
tz = ZoneInfo('Asia/Hong_Kong')
```

</details>

I’ll also add a header box at the start of each run, so the notebook
output frames the run visually.

------------------------------------------------------------------------

### compute_usage

``` python
def compute_usage(
    db, # Database object
    run_id:int, # Run ID to aggregate usage for
)->str: # JSON string of aggregated usage stats
```

*Aggregate usage stats across all frames in a run. Returns JSON string.*

<details open class="code-fold">
<summary>Exported source</summary>

``` python
def _run_header(
    run, # Run record
    model:str, # Model name
    start:int, # Start second
    stop:int, # Stop second
    step:int, # Frame step interval
    cache:bool, # Whether cache is enabled
):
    "Print run header box."
    print(f'╭─ Run #{run.id} ═══════════════════════════════╮\n│ Model    {model}\n│ Start    {start}\n│ Stop     {stop}\n│ Step     {step}\n│ Frames   {(stop-start)//step}\n│ Cache    {cache}\n│ Time     {datetime.now(tz).strftime("%H:%M:%S")}\n╰──────────────────────────────────────────────╯')
```

</details>

<details open class="code-fold">
<summary>Exported source</summary>

``` python
def compute_usage(
    db, # Database object
    run_id:int, # Run ID to aggregate usage for
)->str: # JSON string of aggregated usage stats
    "Aggregate usage stats across all frames in a run. Returns JSON string."
    usgs = L(db.t.run_frame('run_id=? AND skipped=0', (run_id,))).map(lambda r: loads(r.usage))
    if not usgs: return '{}'
    tot = {k: ({k2:0 for k2 in v} if isinstance(v,dict) else 0) for k,v in usgs[0].items()}
    for u in usgs:
        for k,v in u.items():
            if isinstance(v,dict):
                for k2,v2 in v.items(): tot[k][k2] += v2
            else: tot[k] += v
    return dumps(tot)

def _finish_run(
    db, # Database object
    run_id:int, # Run ID to finalize
)->Run: # Updated run record
    "Update run with finish time and usage, print summary box."
    finish = datetime.now(tz)
    run = db.t.run[run_id]
    start = datetime.fromisoformat(run.deploy_time)
    elapsed = finish - start
    db.t.run.update(id=run_id, finish_time=finish, start_time=start.isoformat(), total_duration=str(elapsed).split('.')[0], usage=compute_usage(db, run_id))
    run = db.t.run[run_id]
    tot = dict2obj(loads(run.usage))
    cost = getattr(tot, 'cost', 0)
    print(f'╭─ Run #{run.id} Complete ═════════════════════╮\n│ Finish   {datetime.fromisoformat(run.finish_time).strftime("%H:%M:%S")}\n│ Elapsed  {str(elapsed).split(".")[0]}\n│ Cost     ${cost:.4f} (HKD {cost*7.84:.2f})\n╰──────────────────────────────────────────────╯')
    return run
```

</details>

<details open class="code-fold">
<summary>Exported source</summary>

``` python
from typing import Callable
from fastprogress.fastprogress import NBMasterBar as master_bar
```

</details>

Finally, I’ll tie everything together, batch processing, retry, usage
tracking, and visual framing, into a single `deploy_run` function. It
takes a video ID, sets up the run record, dispatches frames in parallel,
retries rate-limited ones, and stamps the run with its finish time and
total cost.

------------------------------------------------------------------------

### deploy_run

``` python
async def deploy_run(
    video_id:int, # Video ID to process
    db:Database, # Database object
    session:Callable, # Stream session partial
    prompt:str, # Prompt template for the VLM
    prompt_type:str, # Type label for this prompt
    start:int=0, # Start frame index
    stop:int | None=None, # Stop frame index (defaults to last)
    step:int=1, # Frame step interval
    cache:bool=False, # Whether to use HTTP cache
    include_subs:bool=True, # Whether to append subtitle to prompt
    n_workers:int=8, # Number of parallel workers
    pause:float=3, # Seconds to pause between dispatches
    max_retries:int=2, # Max retry attempts for skipped frames
)->Run: # Completed run record with usage stats
```

*Run a single prompt across a range of frames, storing results in the
database.*

<details open class="code-fold">
<summary>Exported source</summary>

``` python
from fastcore.parallel import parallel_async_gen
async def deploy_run(
    video_id:int, # Video ID to process
    db:Database, # Database object
    session:Callable, # Stream session partial
    prompt:str, # Prompt template for the VLM
    prompt_type:str, # Type label for this prompt
    start:int=0, # Start frame index
    stop:int|None=None, # Stop frame index (defaults to last)
    step:int=1, # Frame step interval
    cache:bool=False, # Whether to use HTTP cache
    include_subs:bool=True, # Whether to append subtitle to prompt
    n_workers:int=8, # Number of parallel workers
    pause:float=3, # Seconds to pause between dispatches
    max_retries:int=2, # Max retry attempts for skipped frames
)->Run: # Completed run record with usage stats
    "Run a single prompt across a range of frames, storing results in the database."
    video = db.t.video[video_id]
    fpath = L(Path(video.path).glob('frame_*.jpg')).sorted(key=~Self.stem.split('_'))
    if stop is None: stop = len(fpath)
    if not cache: disable_cachy(); print('!! Cache disabled')
    else: print('!! Using cache')

    run = db.t.run.insert(deploy_time=datetime.now(tz), video_id=video_id, model=session.keywords['model'], num_frames=(stop-start)//step, start_sec=start, end_sec=stop, step=step)
    _run_header(run, session.keywords['model'], start, stop, step, cache)

    frames = fpath[start:stop:step]
    try: await _run_batch(frames, db, session, run.id, video_id, prompt, prompt_type, include_subs, n_workers, pause)
    except Exception as e:
        if not isinstance(e, APIError): raise
        print(f'!! APIError in deploy: {e}')
    await _retry_skipped(db, Path(video.path), run.id, video_id, session, prompt, prompt_type, include_subs, n_workers, pause, max_retries)

    if not cache: enable_cachy(); print('!! Cache enabled')
    return _finish_run(db, run.id)
```

</details>

## Summary

Each frame now has a VLM description stored in the database. But a video
with hundreds of frames means hundreds of separate descriptions. I want
to now take a window of frames and compress it into a single coherent
passage. Then for each of those passages, I stich them together into a
full narrative.

To do that, I first need to pull the frame descriptions back out of the
database, joined with their frame numbers so I can order them
chronologically. A run might cover a subset of frames, and I might want
to merge multiple runs for the same video, so the query should accept
either a single run ID or a list.

I’ll also need to count tokens, since I’ll be feeding windows of frame
descriptions into the LLM and want to track how much gets compressed.
For that, `tiktoken` gives me an encoder that approximates what the
model sees.

<details open class="code-fold">
<summary>Exported source</summary>

``` python
from tiktoken import encoding_for_model
```

</details>

Let me start by querying the run frames for the run I just did, and
joining with the frame table to get frame numbers. First I’ll see what
the raw rows look like:

``` python
rows = L(db.t.run_frame('run_id=? AND skipped=0', (run_id,)))
rows[:3]
```

    [Run_Frame(run_id=1, frame_id=3, type='description', system_prompt=None, prompt='Describe what you see\n\nSubtitle: [No speech]\n\nSubtitle: [No speech]', description='This is a classroom scene, set in a school:\n1.  At the front of the room, an adult man (likely a teacher, with an ID lanyard, wearing a light grey long-sleeve shirt) stands before a wall of whiteboards plus a pulled-down blank projection screen, positioned to address the class.\n2.  Seated at student desks, all facing front in standard green plastic school chairs, are a small number of students: visible are a blonde girl with a ponytail in a dark top, and several boys in white school uniform shirts, all oriented toward the man at the front.\n3.  Additional classroom details: above the front whiteboards, three colorful decorative educational banners are mounted. A large window on the left side of the frame lets in natural light, next to a white interior door. The right side of the back wall has an exposed brick column, with a security camera mounted to it, and a framed poster hung above the projection screen. This is a small class, with only a handful of students visible in this view.', usage='{"prompt_tokens": 1369, "completion_tokens": 954, "total_tokens": 2323, "cost": 0.00225025, "is_byok": false, "prompt_tokens_details": {"cached_tokens": 0, "cache_write_tokens": 0, "audio_tokens": 0, "video_tokens": 0}, "cost_details": {"upstream_inference_cost": 0.00225025, "upstream_inference_prompt_cost": 0.00034225, "upstream_inference_completions_cost": 0.001908}, "completion_tokens_details": {"reasoning_tokens": 735, "image_tokens": 0, "audio_tokens": 0}}', skipped=0), Run_Frame(run_id=1, frame_id=4, type='description', system_prompt=None, prompt='Describe what you see\n\nSubtitle: [No speech]\n\nSubtitle: [No speech]', description="This is an elementary school classroom, hosting a small tech demonstration for students:\n1.  At the front of the room, a tall light-haired adult man (wearing a light grey long-sleeve shirt and dark trousers) stands at the head of the class, behind a small robotic arm set up on a student desk. A young boy (seen from behind, in a white school shirt) is at this desk interacting with the robotic arm.\n2.  Other students are seated at grouped classroom desks (with standard green plastic school chairs) all facing the front to watch the demo: a girl with a blonde ponytail in a dark top sits to the boy's left, and another student's head is visible in the bottom right corner, also observing.\n3.  The background has typical classroom features: a back wall with a whiteboard, a pulled-down projector screen to the whiteboard's right, colorful educational banners mounted above the whiteboard, a closed white door on the far left, and a ledge under the whiteboards holding a small supply box.", usage='{"prompt_tokens": 1369, "completion_tokens": 943, "total_tokens": 2312, "cost": 0.00222825, "is_byok": false, "prompt_tokens_details": {"cached_tokens": 0, "cache_write_tokens": 0, "audio_tokens": 0, "video_tokens": 0}, "cost_details": {"upstream_inference_cost": 0.00222825, "upstream_inference_prompt_cost": 0.00034225, "upstream_inference_completions_cost": 0.001886}, "completion_tokens_details": {"reasoning_tokens": 726, "image_tokens": 0, "audio_tokens": 0}}', skipped=0)]

Now let me join with the frame table to get frame numbers, and group by
frame number.

``` python
fids = sorted(set(r.frame_id for r in rows))
fn_map = {f.id: f.frame_number for f in db.t.frame(f'id IN ({",".join("?"*len(fids))})', tuple(fids))}
fn_map
```

    {3: 3, 4: 4}

``` python
grouped = rows.groupby(lambda r: fn_map[r.frame_id])
sorted(grouped.keys())
```

    [3, 4]

<details open class="code-fold">
<summary>Exported source</summary>

``` python
def _get_runframes(
    db:Database, # Database object
    run_ids:int|list[int], # Single run ID or list of run IDs
    start:int|None=None, # Filter frames at or after this second
    stop:int|None=None, # Filter frames at or before this second
)->L: # Grouped (frame_number, L[RunFrame]) pairs sorted by frame_number
    "Query runframes for given run(s), join with frame table for frame_number, group by frame_number."
    if isinstance(run_ids, int): run_ids = [run_ids]
    ph = ','.join('?' * len(run_ids))
    rows = L(db.t.run_frame(f'run_id IN ({ph}) AND skipped=0', tuple(run_ids)))
    fids = sorted(set(r.frame_id for r in rows))
    fn_map = {f.id: f.frame_number for f in db.t.frame(f'id IN ({",".join("?"*len(fids))})', tuple(fids))}
    grouped = rows.groupby(lambda r: fn_map[r.frame_id])
    fnums = sorted(grouped.keys())
    if start is not None: fnums = [f for f in fnums if f >= start]
    if stop is not None: fnums = [f for f in fnums if f <= stop]
    return L((fn, L(grouped[fn])) for fn in fnums)
```

</details>

``` python
?encoding_for_model
```

``` python
def encoding_for_model(
    model_name:str
)->Encoding:
    "Returns the encoding used by a model.

    Raises a KeyError if the model name is not recognised.
    "
```

**File:** `/usr/local/lib/python3.12/site-packages/tiktoken/model.py`;
line: 113

**Type:** function

Now I need to format those grouped runframes into a single text block
the LLM can read. Each frame’s description should be prefixed with its
timestamp, and multiple descriptions for the same frame (e.g. from
different prompt types) separated by `--`. This is a “window”. A unit
I’ll feed into the LLM for summarization.

Let me take the first grouped runframe and see what I’m working with.

``` python
rfs = _get_runframes(db, run_id, 0, 10)
i, rf = rfs[0]; i, rf
```

    (3,
     [Run_Frame(run_id=1, frame_id=3, type='description', system_prompt=None, prompt='Describe what you see\n\nSubtitle: [No speech]\n\nSubtitle: [No speech]', description='This is a classroom scene, set in a school:\n1.  At the front of the room, an adult man (likely a teacher, with an ID lanyard, wearing a light grey long-sleeve shirt) stands before a wall of whiteboards plus a pulled-down blank projection screen, positioned to address the class.\n2.  Seated at student desks, all facing front in standard green plastic school chairs, are a small number of students: visible are a blonde girl with a ponytail in a dark top, and several boys in white school uniform shirts, all oriented toward the man at the front.\n3.  Additional classroom details: above the front whiteboards, three colorful decorative educational banners are mounted. A large window on the left side of the frame lets in natural light, next to a white interior door. The right side of the back wall has an exposed brick column, with a security camera mounted to it, and a framed poster hung above the projection screen. This is a small class, with only a handful of students visible in this view.', usage='{"prompt_tokens": 1369, "completion_tokens": 954, "total_tokens": 2323, "cost": 0.00225025, "is_byok": false, "prompt_tokens_details": {"cached_tokens": 0, "cache_write_tokens": 0, "audio_tokens": 0, "video_tokens": 0}, "cost_details": {"upstream_inference_cost": 0.00225025, "upstream_inference_prompt_cost": 0.00034225, "upstream_inference_completions_cost": 0.001908}, "completion_tokens_details": {"reasoning_tokens": 735, "image_tokens": 0, "audio_tokens": 0}}', skipped=0)])

Each frame gets a timestamp header. Let me build that prefix.

``` python
prefix = f'TIMESTAMP {i}s\n'; prefix + len(prefix.strip())*'=' + '\n'
```

    'TIMESTAMP 3s\n============\n'

Then each description entry under that header, separated by `--`.

``` python
r = rf[0]; '\n--\n' + f'{r.type.upper()}\n{r.description}' + '\n\n'
```

    '\n--\nDESCRIPTION\nThis is a classroom scene, set in a school:\n1.  At the front of the room, an adult man (likely a teacher, with an ID lanyard, wearing a light grey long-sleeve shirt) stands before a wall of whiteboards plus a pulled-down blank projection screen, positioned to address the class.\n2.  Seated at student desks, all facing front in standard green plastic school chairs, are a small number of students: visible are a blonde girl with a ponytail in a dark top, and several boys in white school uniform shirts, all oriented toward the man at the front.\n3.  Additional classroom details: above the front whiteboards, three colorful decorative educational banners are mounted. A large window on the left side of the frame lets in natural light, next to a white interior door. The right side of the back wall has an exposed brick column, with a security camera mounted to it, and a framed poster hung above the projection screen. This is a small class, with only a handful of students visible in this view.\n\n'

Now I can assemble a few frames into a window and see what the full text
looks like.

``` python
window = ''
for i,rf in rfs[:3]:
    prefix = f'TIMESTAMP {i}s\n'
    window += prefix + len(prefix.strip())*'=' + '\n'
    for r in rf:
        window += '\n--\n' + f'{r.type.upper()}\n{r.description}' + '\n\n'
print(window[:500])
```

    TIMESTAMP 3s
    ============

    --
    DESCRIPTION
    This is a classroom scene, set in a school:
    1.  At the front of the room, an adult man (likely a teacher, with an ID lanyard, wearing a light grey long-sleeve shirt) stands before a wall of whiteboards plus a pulled-down blank projection screen, positioned to address the class.
    2.  Seated at student desks, all facing front in standard green plastic school chairs, are a small number of students: visible are a blonde girl with a ponytail in a dark top, and

That looks right. Let me extract this into a function, and add a `step`
parameter so I can subsample frames when windows get large.

<details open class="code-fold">
<summary>Exported source</summary>

``` python
def _build_window(
    runframes, # Grouped runframes from `_get_runframes`
    step:int=1, # Subsample every Nth frame
)->str: # Formatted window text for LLM input
    "Build window text from grouped runframes."
    window = ''
    for i,rf in runframes[::step]:
        prefix = f'TIMESTAMP {i}s\n'
        window += prefix+len(prefix.strip())*'='+'\n'
        for r in rf:
            window += '\n--\n'+f'{r.type.upper()}\n{r.description}'+'\n\n'
    return window
```

</details>

With the window built, I want to see how many tokens it is before
feeding it to the LLM. That tells me how much I’m asking the model to
compress. Let me try the encoder on a sample window.

``` python
enc = encoding_for_model('gpt-4o')
rfs = _get_runframes(db, run_id, 0, 10)
win = _build_window(rfs)
len(enc.encode(win))
```

    451

I’ll add visual framing like the run header — a box at the start showing
the window size and token count, and a box at the end showing the
summary size, compression ratio, and cost.

<details open class="code-fold">
<summary>Exported source</summary>

``` python
def _summary_header(
    run_id:int, # Run ID (or label for multi-run)
    start:int, # Start frame second
    stop:int, # Stop frame second
    step:int, # Frame step interval
    model:str, # Model name
    window:str, # Full window text
    win_tokens:int, # Token count of window
    cache:bool, # Whether cache is enabled
    t0:datetime, # Start timestamp
):
    "Print summary run header box."
    print(f'╭─ Summary Run #{run_id} ═══════════════════════╮\n│ Frames   {start}–{stop} (step {step})\n│ Model    {model}\n│ Window   {len(window)} chars / {win_tokens} tokens\n│ Cache    {cache}\n│ Start    {t0.strftime("%H:%M:%S")}\n╰──────────────────────────────────────────────╯')
```

</details>

<details open class="code-fold">
<summary>Exported source</summary>

``` python
def _summary_footer(
    summary:str, # Summary text from LLM
    win_tokens:int, # Token count of original window
    t0:datetime, # Start timestamp
    t1:datetime, # Finish timestamp
    cost:float, # Dollar cost of this summary call
):
    "Print summary completion box."
    enc = encoding_for_model('gpt-4o')
    sum_tokens = len(enc.encode(summary))
    reduction = (1 - sum_tokens/win_tokens)*100 if win_tokens else 0
    elapsed = t1 - t0
    print(f'╭─ Summary Complete ═══════════════════════════╮\n│ Finish   {t1.strftime("%H:%M:%S")}\n│ Elapsed  {str(elapsed).split(".")[0]}\n│ Summary  {len(summary)} chars / {sum_tokens} tokens\n│ Reduced  {reduction:.1f}%\n│ Cost     ${cost:.4f} (HKD {cost*7.84:.2f})\n╰──────────────────────────────────────────────╯')
```

</details>

Let me try running a single window through the LLM to see what kind of
summary comes back.

``` python
sys_prompt = "You are a careful observer. Summarize the following frame descriptions into a coherent narrative passage."
```

``` python
r = await s([user(win)], system=sys_prompt)
print(r.message.text[:500])
```

    Across two consecutive moments in a small elementary school classroom, the scene shifts from the teacher preparing to address his group to the start of a hands-on tech demonstration. At the 3-second mark, a male teacher in a light grey long-sleeve shirt and ID lanyard stands at the front of the sunlit room, positioned before a wall of whiteboards and a blank pulled-down projection screen to speak to his handful of students. All the children sit in standard green plastic school chairs at forward-

With those in place, I can wrap the whole call into a single
`summarize_window` function that builds the window, sends it to the LLM
with a system prompt, and prints the boxes.

------------------------------------------------------------------------

### summarize_window

``` python
async def summarize_window(
    db:Database, # Database object
    run_ids:int | list[int], # Single run ID or list of run IDs
    start:int, # Start frame second
    stop:int, # Stop frame second
    session:Callable, # Stream session partial
    sys_prompt:str, # System prompt for the summarizing LLM
    step:int=1, # Subsample every Nth frame
    cache:bool=False, # Whether to use HTTP cache
    context:str='', # Prior summary text to prepend as context
)->str: # Summary text for this window
```

*Summarize a window of frames from runframes. Returns summary text.*

<details open class="code-fold">
<summary>Exported source</summary>

``` python
async def summarize_window(
    db:Database, # Database object
    run_ids:int|list[int], # Single run ID or list of run IDs
    start:int, # Start frame second
    stop:int, # Stop frame second
    session:Callable, # Stream session partial
    sys_prompt:str, # System prompt for the summarizing LLM
    step:int=1, # Subsample every Nth frame
    cache:bool=False, # Whether to use HTTP cache
    context:str='', # Prior summary text to prepend as context
)->str: # Summary text for this window
    "Summarize a window of frames from runframes. Returns summary text."
    if not cache: disable_cachy()
    rfs = _get_runframes(db, run_ids, start, stop)
    window = _build_window(rfs, step)
    if context: window = f'PRIOR CONTEXT\n{"="*13}\n{context}\n\n{window}'
    win_tokens = len(encoding_for_model('gpt-4o').encode(window))
    t0 = datetime.now(tz)
    label = run_ids if isinstance(run_ids, int) else f'{run_ids[0]}+{len(run_ids)-1}more'
    _summary_header(label, start, stop, step, session.keywords['model'], window, win_tokens, cache, t0)
    r = await session([user(window)], system=sys_prompt)
    t1 = datetime.now(tz)
    if not cache: enable_cachy()
    summary = r.message.text
    _summary_footer(summary, win_tokens, t0, t1, r.usage.raw.get('cost', 0))
    return summary
```

</details>

Let me try this manually with two windows to see if the context chaining
produces a flowing narrative.

``` python
fnums = rfs.itemgot(0)
chunks = list(chunked(fnums, 300)); chunks[:3]
```

    [[3, 4]]

``` python
s1 = await summarize_window(db, run_id, chunks[0][0], chunks[0][-1], s, sys_prompt)
if len(chunks) > 1: s2 = await summarize_window(db, run_id, chunks[1][0], chunks[1][-1], s, sys_prompt, context=s1)
```

    ╭─ Summary Run #1 ═══════════════════════╮
    │ Frames   3–4 (step 1)
    │ Model    bytedance-seed/seed-2.0-lite
    │ Window   2072 chars / 451 tokens
    │ Cache    False
    │ Start    15:38:08
    ╰──────────────────────────────────────────────╯

    ╭─ Summary Complete ═══════════════════════════╮
    │ Finish   15:38:33
    │ Elapsed  0:00:25
    │ Summary  1578 chars / 323 tokens
    │ Reduced  28.4%
    │ Cost     $0.0039 (HKD 0.03)
    ╰──────────────────────────────────────────────╯

A single window gives me one passage. But a full video has many windows.
I want to feed each window to the LLM with the previous window’s summary
as context, so the narrative flows continuously. That means chunking
frames into `window_sec`-sized groups (default 300 seconds / 5 minutes),
summarizing each in turn, and concatenating the results. The final
summary gets written back to the database, onto the run record if I
started from a run ID, or onto the video record if I started from a
video ID (merging all its runs).

------------------------------------------------------------------------

### summarize_run

``` python
async def summarize_run(
    db:Database, # Database object
    session:Callable, # Stream session partial
    sys_prompt:str, # System prompt for the summarizing LLM
    perspective:str, # Key for storing this summary in description JSON
    run_id:int | None=None, # Single run ID to summarize
    video_id:int | None=None, # Summarize all runs for this video
    window_sec:int=300, # Rolling window size in seconds
    step:int=1, # Subsample every Nth frame within windows
    cache:bool=False, # Whether to use HTTP cache
)->str: # Full concatenated summary across all windows
```

*Summarize a single run or all frames for a video in rolling windows.*

<details open class="code-fold">
<summary>Exported source</summary>

``` python
from IPython.display import clear_output
```

</details>

<details open class="code-fold">
<summary>Exported source</summary>

``` python
async def summarize_run(
    db:Database, # Database object
    session:Callable, # Stream session partial
    sys_prompt:str, # System prompt for the summarizing LLM
    perspective:str, # Key for storing this summary in description JSON
    run_id:int|None=None, # Single run ID to summarize
    video_id:int|None=None, # Summarize all runs for this video
    window_sec:int=300, # Rolling window size in seconds
    step:int=1, # Subsample every Nth frame within windows
    cache:bool=False, # Whether to use HTTP cache
)->str: # Full concatenated summary across all windows
    "Summarize a single run or all frames for a video in rolling windows."
    if run_id is not None: rids = [run_id]
    elif video_id is not None: rids = L(db.t.run('video_id=?', (video_id,))).map(lambda r: r.id)
    else: raise ValueError('Either run_id or video_id required')
    rfs = _get_runframes(db, rids)
    if not rfs: return ''
    fnums = rfs.itemgot(0)
    full_summary = ''
    chunks = list(chunked(fnums, window_sec))
    for chunk in (mb:=master_bar(chunks)):
        mb.main_bar.comment = f'window {chunk[0]}–{chunk[-1]}s'
        summary = await summarize_window(db, rids, chunk[0], chunk[-1], session, sys_prompt, step=step, cache=cache, context=full_summary)
        heading = f'[{chunk[0]}–{chunk[-1]}s]'
        summary = summary.strip()
        full_summary = full_summary + f'\n\n{heading} {summary}' if full_summary else f'{heading} {summary}' if summary else ''
        clear_output(wait=True)
        print(full_summary)
    target = db.t.run[run_id] if run_id is not None else db.t.video[video_id]
    desc = loads(target.description) if target.description else {}
    desc[perspective or 'default'] = full_summary
    if run_id is not None: db.t.run.update(id=run_id, description=dumps(desc))
    else: db.t.video.update(id=video_id, description=dumps(desc))
    return full_summary
```

</details>

`vlm_monitor` supports all providers
[`fastllm`](https://github.com/AnswerDotAI/fastllm) supports.
`vlm_monitor` uses an older version of `fastllm` and thus requires the
`vendor_name` parameter amongst other differences.