#!/usr/bin/env python3
__notes__ = r"""
.. :code: bash
# Make an animated gif for specified bands (use "," to separate)
# Requires a CD
CHANNELS="red|green|blue"
mapfile -td \, _BANDS < <(printf "%s\0" "$CHANNELS")
items=$(jq -r '.videos[] | .name' $OUTPUT_COCO_FPATH)
for item in ${items[@]}; do
echo "item = $item"
for bandname in ${_BANDS[@]}; do
echo "_BANDS = $_BANDS"
BAND_DPATH="$VIZ_DPATH/${item}/_anns/${bandname}/"
GIF_FPATH="$VIZ_DPATH/${item}_anns_${bandname}.gif"
python -m kwplot.cli.gifify --frames_per_second .7 \
--input "$BAND_DPATH" --output "$GIF_FPATH"
done
done
"""
[docs]
def animate_visualizations(viz_dpath, channels=None, video_names=None,
frames_per_second=0.7, draw_anns=True,
draw_imgs=True, workers=0, zoom_to_tracks=False,
verbose=0):
r"""
Helper that roughly does the same thing as this bash script:
Args:
viz_dpath (str): the path where visualizations were dumped with the
coco_visualize_videos script.
zoom_to_tracks (bool):
if specified uses "track" based-logic find paths to animate
Example:
>>> # xdoctest: +SKIP
>>> # xdoctest: +REQUIRES(--ffmpeg-test')
>>> import ubelt as ub
>>> dpath = ub.Path.appdir('geowatch/tests/ani_video').delete().ensuredir()
>>> import kwcoco
>>> from geowatch.utils import kwcoco_extensions
>>> dset = kwcoco.CocoDataset.demo('vidshapes2-msi', num_frames=5)
>>> img = dset.dataset['images'][0]
>>> coco_img = dset.coco_image(img['id'])
>>> channel_chunks = list(ub.chunks(coco_img.channels.fuse().parsed, chunksize=3))
>>> channels = ','.join(['|'.join(p) for p in channel_chunks])
>>> kwargs = {
>>> 'src': dset.fpath,
>>> 'viz_dpath': dpath,
>>> 'space': 'video',
>>> 'channels': channels,
>>> 'zoom_to_tracks': False,
>>> }
>>> from geowatch.cli.coco_visualize_videos import main
>>> cmdline = False
>>> main(cmdline=cmdline, **kwargs)
>>> viz_dpath = dpath
>>> channels = None
>>> video_names = None
>>> frame_per_second = 0.7
>>> from geowatch.cli.animate_visualizations import * # NOQA
>>> animate_visualizations(viz_dpath, verbose=1, workers=0)
"""
from kwplot.cli import gifify
import ubelt as ub
import kwcoco
from kwutil import util_parallel
if channels is not None:
channels = kwcoco.ChannelSpec.coerce(channels)
workers = util_parallel.coerce_num_workers(workers)
viz_dpath = ub.Path(viz_dpath)
ffmpeg_exe = ub.find_exe('ffmpeg')
if ffmpeg_exe is None:
raise Exception('Cannot find ffmpeg, which is required to run animation')
if video_names is None:
video_dpaths = [p for p in viz_dpath.glob('*') if p.is_dir()]
else:
if len(video_names) == 1:
workers = 0
video_dpaths = [viz_dpath / n for n in video_names]
pool = ub.JobPool(mode='thread', max_workers=workers)
types = []
if draw_imgs:
types.append('_imgs')
if draw_anns:
types.append('_anns')
if workers == 1:
workers = 0
verbose_worker = verbose and workers <= 1
# We make heavy reliance on a known directory structure here.
# In general I don't like this, but this is not a system-critical part
# so we can leave refactoring as a todo.
from kwutil import util_progress
pman = util_progress.ProgressManager()
pman.__enter__()
# prog = ub.ProgIter(desc='submit video jobs', verbose=3)
prog = pman.progiter(desc='submit video jobs')
prog.begin()
with_gif = 'auto'
with_mp4 = True
outputs = []
for type_ in types:
for video_dpath in video_dpaths:
prog.set_extra('type_={!r} video_dpath={!r}'.format(type_, video_dpath))
prog.step()
video_name = video_dpath.name
if zoom_to_tracks:
track_subdpath = video_dpath / '_tracks'
track_dpaths = list(track_subdpath.glob('*'))
for track_dpath in track_dpaths:
track_name = track_dpath.name
type_dpath = track_dpath / type_
if channels is None:
channel_dpaths = [p for p in type_dpath.glob('*') if p.is_dir()]
else:
channel_dpaths = [type_dpath / c.path_sanitize()
for c in channels.streams()]
for chan_dpath in channel_dpaths:
frame_fpaths = sorted(chan_dpath.glob('*'))
if len(frame_fpaths):
if len(frame_fpaths) < 300:
gif_fname = '{}{}_{}.gif'.format(track_name, type_, chan_dpath.name)
gif_fpath = track_subdpath / gif_fname
pool.submit(
gifify.ffmpeg_animate_frames, frame_fpaths,
gif_fpath, in_framerate=frames_per_second,
verbose=verbose_worker)
outputs.append({
'fpath': gif_fpath,
'type': 'gif',
})
ani_fname = '{}{}_{}.mp4'.format(track_name, type_, chan_dpath.name)
ani_fpath = track_subdpath / ani_fname
pool.submit(
gifify.ffmpeg_animate_frames, frame_fpaths,
ani_fpath, in_framerate=frames_per_second,
verbose=verbose_worker)
outputs.append({
'fpath': ani_fpath,
'type': 'mp4',
})
else:
type_dpath = video_dpath / type_
if channels is None:
channel_dpaths = [p for p in type_dpath.glob('*') if p.is_dir()]
else:
channel_dpaths = [type_dpath / c.path_sanitize()
for c in channels.streams()]
for chan_dpath in channel_dpaths:
frame_fpaths = sorted(chan_dpath.glob('*'))
if len(frame_fpaths):
with_gif_resolved = with_gif
if with_gif == 'auto':
with_gif_resolved = len(frame_fpaths) < 300
if with_gif_resolved:
gif_fname = '{}{}_{}.gif'.format(video_name, type_, chan_dpath.name)
gif_fpath = video_dpath / gif_fname
pool.submit(
gifify.ffmpeg_animate_frames, frame_fpaths, gif_fpath,
in_framerate=frames_per_second, verbose=verbose_worker)
outputs.append({
'fpath': gif_fpath,
'type': 'gif',
})
if with_mp4:
ani_fname = '{}{}_{}.mp4'.format(video_name, type_, chan_dpath.name)
ani_fpath = video_dpath / ani_fname
pool.submit(
gifify.ffmpeg_animate_frames, frame_fpaths, ani_fpath,
in_framerate=frames_per_second, verbose=verbose_worker)
outputs.append({
'fpath': ani_fpath,
'type': 'mp4',
})
prog.end()
failed = []
# for job in ub.ProgIter(pool.as_completed(), total=len(pool), desc='collect animate jobs'):
for job in pman.progiter(pool.as_completed(), total=len(pool), desc='collect animate jobs'):
try:
job.result()
except Exception as ex:
failed.append(ex)
pass
if failed:
print('Animation jobs failed with the following errors:')
print('failed = {}'.format(ub.urepr(failed, nl=1)))
raise Exception(f'{len(failed)} / {len(pool)} animations failed')
pman.__exit__(None, None, None)
print('Wrote animations to viz_dpath = {!r}'.format(viz_dpath))
# The animation jobs can do something weird to the tty, so we should try
# and fix it.
ub.cmd('stty sane')
return outputs
if __name__ == '__main__':
"""
CommandLine:
python ~/code/watch/geowatch/cli/animate_visualizations.py
"""
import fire
fire.Fire(animate_visualizations)