"""

# this maybe simplified if / when we adopt consistent rounding for

# pixel size across the whole library

def correct_roundoff(x, dpi, n):

if int(x*dpi) % n != 0:

if int(np.nextafter(x, np.inf)*dpi) % n == 0:

x = np.nextafter(x, np.inf)

elif int(np.nextafter(x, -np.inf)*dpi) % n == 0:

x = np.nextafter(x, -np.inf)

return x

wnew = int(w * dpi / n) * n / dpi

hnew = int(h * dpi / n) * n / dpi

"""Registry of available writer classes by human readable name."""

def __init__(self):

self._registered = dict()

def register(self, name):

"""

def wrapper(writer_cls):

self._registered[name] = writer_cls

return writer_cls

return wrapper

def is_available(self, name):

"""

try:

cls = self._registered[name]

except KeyError:

return False

return cls.isAvailable()

def __iter__(self):

"""Iterate over names of available writer class."""

for name in self._registered:

if self.is_available(name):

yield name

def list(self):

"""Get a list of available MovieWriters."""

return [*self]

def __getitem__(self, name):

"""Get an available writer class from its name."""

if self.is_available(name):

return self._registered[name]

"""

def __init__(self, fps=5, metadata=None, codec=None, bitrate=None):

self.fps = fps

self.metadata = metadata if metadata is not None else {}

self.codec = mpl._val_or_rc(codec, 'animation.codec')

self.bitrate = mpl._val_or_rc(bitrate, 'animation.bitrate')

@abc.abstractmethod

def setup(self, fig, outfile, dpi=None):

"""

# Check that path is valid

Path(outfile).parent.resolve(strict=True)

self.outfile = outfile

self.fig = fig

if dpi is None:

dpi = self.fig.dpi

self.dpi = dpi

@property

def frame_size(self):

"""A tuple ``(width, height)`` in pixels of a movie frame."""

w, h = self.fig.get_size_inches()

return int(w * self.dpi), int(h * self.dpi)

def _supports_transparency(self):

"""

return False

@abc.abstractmethod

def grab_frame(self, **savefig_kwargs):

"""

@abc.abstractmethod

def finish(self):

"""Finish any processing for writing the movie."""

@contextlib.contextmanager

def saving(self, fig, outfile, dpi, *args, **kwargs):

"""

if mpl.rcParams['savefig.bbox'] == 'tight':

_log.info("Disabling savefig.bbox = 'tight', as it may cause "

"frame size to vary, which is inappropriate for "

"animation.")

# This particular sequence is what contextlib.contextmanager wants

self.setup(fig, outfile, dpi, *args, **kwargs)

with mpl.rc_context({'savefig.bbox': None}):

try:

yield self

finally:

"""

# Builtin writer subclasses additionally define the _exec_key and _args_key

# attributes, which indicate the rcParams entries where the path to the

# executable and additional command-line arguments to the executable are

# stored. Third-party writers cannot meaningfully set these as they cannot

# extend rcParams with new keys.

# Pipe-based writers only support RGBA, but file-based ones support more

# formats.

supported_formats = ["rgba"]

def __init__(self, fps=5, codec=None, bitrate=None, extra_args=None,

metadata=None):

"""

if type(self) is MovieWriter:

# TODO MovieWriter is still an abstract class and needs to be

# extended with a mixin. This should be clearer in naming

# and description. For now, just give a reasonable error

# message to users.

raise TypeError(

'MovieWriter cannot be instantiated directly. Please use one '

'of its subclasses.')

super().__init__(fps=fps, metadata=metadata, codec=codec,

bitrate=bitrate)

self.frame_format = self.supported_formats[0]

self.extra_args = extra_args

def _adjust_frame_size(self):

if self.codec == 'h264':

wo, ho = self.fig.get_size_inches()

w, h = adjusted_figsize(wo, ho, self.dpi, 2)

if (wo, ho) != (w, h):

self.fig.set_size_inches(w, h, forward=True)

_log.info('figure size in inches has been adjusted '

'from %s x %s to %s x %s', wo, ho, w, h)

else:

w, h = self.fig.get_size_inches()

_log.debug('frame size in pixels is %s x %s', *self.frame_size)

return w, h

def setup(self, fig, outfile, dpi=None):

# docstring inherited

super().setup(fig, outfile, dpi=dpi)

self._w, self._h = self._adjust_frame_size()

# Run here so that grab_frame() can write the data to a pipe. This

# eliminates the need for temp files.

self._run()

def _run(self):

# Uses subprocess to call the program for assembling frames into a

# movie file. *args* returns the sequence of command line arguments

# from a few configuration options.

command = self._args()

_log.info('MovieWriter._run: running command: %s',

cbook._pformat_subprocess(command))

PIPE = subprocess.PIPE

self._proc = subprocess.Popen(

command, stdin=PIPE, stdout=PIPE, stderr=PIPE,

creationflags=subprocess_creation_flags)

def finish(self):

"""Finish any processing for writing the movie."""

out, err = self._proc.communicate()

# Use the encoding/errors that universal_newlines would use.

out = TextIOWrapper(BytesIO(out)).read()

err = TextIOWrapper(BytesIO(err)).read()

if out:

_log.log(

logging.WARNING if self._proc.returncode else logging.DEBUG,

"MovieWriter stdout:\n%s", out)

if err:

_log.log(

logging.WARNING if self._proc.returncode else logging.DEBUG,

"MovieWriter stderr:\n%s", err)

if self._proc.returncode:

raise subprocess.CalledProcessError(

self._proc.returncode, self._proc.args, out, err)

def grab_frame(self, **savefig_kwargs):

# docstring inherited

_validate_grabframe_kwargs(savefig_kwargs)

_log.debug('MovieWriter.grab_frame: Grabbing frame.')

# Readjust the figure size in case it has been changed by the user.

# All frames must have the same size to save the movie correctly.

self.fig.set_size_inches(self._w, self._h)

# Save the figure data to the sink, using the frame format and dpi.

self.fig.savefig(self._proc.stdin, format=self.frame_format,

dpi=self.dpi, **savefig_kwargs)

def _args(self):

"""Assemble list of encoder-specific command-line arguments."""

return NotImplementedError("args needs to be implemented by subclass.")

@classmethod

def bin_path(cls):

"""

return str(mpl.rcParams[cls._exec_key])

@classmethod

def isAvailable(cls):

"""Return whether a MovieWriter subclass is actually available."""

if sys.platform == 'emscripten':

return False

"""

def __init__(self, *args, **kwargs):

super().__init__(*args, **kwargs)

self.frame_format = mpl.rcParams['animation.frame_format']

def setup(self, fig, outfile, dpi=None, frame_prefix=None):

"""

# Check that path is valid

Path(outfile).parent.resolve(strict=True)

self.fig = fig

self.outfile = outfile

if dpi is None:

dpi = self.fig.dpi

self.dpi = dpi

self._adjust_frame_size()

if frame_prefix is None:

self._tmpdir = TemporaryDirectory()

self.temp_prefix = str(Path(self._tmpdir.name, 'tmp'))

else:

self._tmpdir = None

self.temp_prefix = frame_prefix

self._frame_counter = 0 # used for generating sequential file names

self._temp_paths = list()

self.fname_format_str = '%s%%07d.%s'

def __del__(self):

if hasattr(self, '_tmpdir') and self._tmpdir:

self._tmpdir.cleanup()

@property

def frame_format(self):

"""

return self._frame_format

@frame_format.setter

def frame_format(self, frame_format):

if frame_format in self.supported_formats:

self._frame_format = frame_format

else:

_api.warn_external(

f"Ignoring file format {frame_format!r} which is not "

f"supported by {type(self).__name__}; using "

f"{self.supported_formats[0]} instead.")

self._frame_format = self.supported_formats[0]

def _base_temp_name(self):

# Generates a template name (without number) given the frame format

# for extension and the prefix.

return self.fname_format_str % (self.temp_prefix, self.frame_format)

def grab_frame(self, **savefig_kwargs):

# docstring inherited

# Creates a filename for saving using basename and counter.

_validate_grabframe_kwargs(savefig_kwargs)

path = Path(self._base_temp_name() % self._frame_counter)

self._temp_paths.append(path) # Record the filename for later use.

self._frame_counter += 1 # Ensures each created name is unique.

_log.debug('FileMovieWriter.grab_frame: Grabbing frame %d to path=%s',

self._frame_counter, path)

with open(path, 'wb') as sink: # Save figure to the sink.

self.fig.savefig(sink, format=self.frame_format, dpi=self.dpi,

**savefig_kwargs)

def finish(self):

# Call run here now that all frame grabbing is done. All temp files

# are available to be assembled.

try:

self._run()

super().finish()

finally:

if self._tmpdir:

_log.debug(

'MovieWriter: clearing temporary path=%s', self._tmpdir

)

def _supports_transparency(self):

return True

@classmethod

def isAvailable(cls):

return True

def setup(self, fig, outfile, dpi=None):

super().setup(fig, outfile, dpi=dpi)

self._frames = []

def grab_frame(self, **savefig_kwargs):

_validate_grabframe_kwargs(savefig_kwargs)

buf = BytesIO()

self.fig.savefig(

buf, **{**savefig_kwargs, "format": "rgba", "dpi": self.dpi})

im = Image.frombuffer(

"RGBA", self.frame_size, buf.getbuffer(), "raw", "RGBA", 0, 1)

if im.getextrema()[3][0] < 255:

# This frame has transparency, so we'll just add it as is.

self._frames.append(im)

else:

# Without transparency, we switch to RGB mode, which converts to P mode a

# little better if needed (specifically, this helps with GIF output.)

self._frames.append(im.convert("RGB"))

def finish(self):

self._frames[0].save(

self.outfile, save_all=True, append_images=self._frames[1:],

"""

_exec_key = 'animation.ffmpeg_path'

_args_key = 'animation.ffmpeg_args'

def _supports_transparency(self):

suffix = Path(self.outfile).suffix

if suffix in {'.apng', '.avif', '.gif', '.webm', '.webp'}:

return True

# This list was found by going through `ffmpeg -codecs` for video encoders,

# running them with _support_transparency() forced to True, and checking that

# the "Pixel format" in Kdenlive included alpha. Note this is not a guarantee

# that transparency will work; you may also need to pass `-pix_fmt`, but we

# trust the user has done so if they are asking for these formats.

return self.codec in {

'apng', 'avrp', 'bmp', 'cfhd', 'dpx', 'ffv1', 'ffvhuff', 'gif', 'huffyuv',

'jpeg2000', 'ljpeg', 'png', 'prores', 'prores_aw', 'prores_ks', 'qtrle',

'rawvideo', 'targa', 'tiff', 'utvideo', 'v408', }

@property

def output_args(self):

args = []

suffix = Path(self.outfile).suffix

if suffix in {'.apng', '.avif', '.gif', '.webm', '.webp'}:

self.codec = suffix[1:]

else:

args.extend(['-vcodec', self.codec])

extra_args = (self.extra_args if self.extra_args is not None

else mpl.rcParams[self._args_key])

# For h264, the default format is yuv444p, which is not compatible

# with quicktime (and others). Specifying yuv420p fixes playback on

# iOS, as well as HTML5 video in firefox and safari (on both Windows and

# macOS). Also fixes internet explorer. This is as of 2015/10/29.

if self.codec == 'h264' and '-pix_fmt' not in extra_args:

args.extend(['-pix_fmt', 'yuv420p'])

# For GIF, we're telling FFmpeg to split the video stream, to generate

# a palette, and then use it for encoding.

elif self.codec == 'gif' and '-filter_complex' not in extra_args:

args.extend(['-filter_complex',

'split [a][b];[a] palettegen [p];[b][p] paletteuse'])

# For AVIF, we're telling FFmpeg to split the video stream, extract the alpha,

# in order to place it in a secondary stream, as needed by AVIF-in-FFmpeg.

elif self.codec == 'avif' and '-filter_complex' not in extra_args:

args.extend(['-filter_complex',

'split [rgb][rgba]; [rgba] alphaextract [alpha]',

'-map', '[rgb]', '-map', '[alpha]'])

if self.bitrate > 0:

args.extend(['-b', '%dk' % self.bitrate]) # %dk: bitrate in kbps.

for k, v in self.metadata.items():

args.extend(['-metadata', f'{k}={v}'])

args.extend(extra_args)

"""

def _args(self):

# Returns the command line parameters for subprocess to use

# ffmpeg to create a movie using a pipe.

args = [self.bin_path(), '-f', 'rawvideo', '-vcodec', 'rawvideo',

'-s', '%dx%d' % self.frame_size, '-pix_fmt', self.frame_format,

'-framerate', str(self.fps)]

# Logging is quieted because subprocess.PIPE has limited buffer size.

# If you have a lot of frames in your animation and set logging to

# DEBUG, you will have a buffer overrun.

if _log.getEffectiveLevel() > logging.DEBUG:

args += ['-loglevel', 'error']

args += ['-i', 'pipe:'] + self.output_args

"""

supported_formats = ['png', 'jpeg', 'tiff', 'raw', 'rgba']

def _args(self):

# Returns the command line parameters for subprocess to use

# ffmpeg to create a movie using a collection of temp images

args = []

# For raw frames, we need to explicitly tell ffmpeg the metadata.

if self.frame_format in {'raw', 'rgba'}:

args += [

'-f', 'image2', '-vcodec', 'rawvideo',

'-video_size', '%dx%d' % self.frame_size,

'-pixel_format', 'rgba',

]

args += ['-framerate', str(self.fps), '-i', self._base_temp_name()]

if not self._tmpdir:

args += ['-frames:v', str(self._frame_counter)]

# Logging is quieted because subprocess.PIPE has limited buffer size.

# If you have a lot of frames in your animation and set logging to

# DEBUG, you will have a buffer overrun.

if _log.getEffectiveLevel() > logging.DEBUG:

args += ['-loglevel', 'error']

"""

_exec_key = 'animation.convert_path'

_args_key = 'animation.convert_args'

def _supports_transparency(self):

suffix = Path(self.outfile).suffix

return suffix in {'.apng', '.avif', '.gif', '.webm', '.webp'}

def _args(self):

# ImageMagick does not recognize "raw".

fmt = "rgba" if self.frame_format == "raw" else self.frame_format

extra_args = (self.extra_args if self.extra_args is not None

else mpl.rcParams[self._args_key])

return [

self.bin_path(),

"-size", "%ix%i" % self.frame_size,

"-depth", "8",

"-delay", str(100 / self.fps),

"-loop", "0",

f"{fmt}:{self.input_names}",

*extra_args,

self.outfile,

]

@classmethod

def bin_path(cls):

binpath = super().bin_path()

if binpath == 'convert':

binpath = mpl._get_executable_info('magick').executable

return binpath

@classmethod

def isAvailable(cls):

try:

return super().isAvailable()

except mpl.ExecutableNotFoundError as _enf:

# May be raised by get_executable_info.

_log.debug('ImageMagick unavailable due to: %s', _enf)

"""

"""

supported_formats = ['png', 'jpeg', 'tiff', 'raw', 'rgba']

input_names = property(

"""frame_list should be a list of base64-encoded png files"""

if frame_format == 'svg':

# Fix MIME type for svg

frame_format = 'svg+xml'

template = ' frames[{0}] = "data:image/{1};base64,{2}"\n'

return "\n" + "".join(

template.format(i, frame_format, frame_data.replace('\n', '\\\n'))

"""Writer for JavaScript-based HTML movies."""

supported_formats = ['png', 'jpeg', 'tiff', 'svg']

@classmethod

def isAvailable(cls):

return True

def __init__(self, fps=30, codec=None, bitrate=None, extra_args=None,

metadata=None, embed_frames=False, default_mode='loop',

embed_limit=None):

if extra_args:

_log.warning("HTMLWriter ignores 'extra_args'")

extra_args = () # Don't lookup nonexistent rcParam[args_key].

self.embed_frames = embed_frames

self.default_mode = default_mode.lower()

_api.check_in_list(['loop', 'once', 'reflect'],

default_mode=self.default_mode)

# Save embed limit, which is given in MB

self._bytes_limit = mpl._val_or_rc(embed_limit, 'animation.embed_limit')

# Convert from MB to bytes

self._bytes_limit *= 1024 * 1024

super().__init__(fps, codec, bitrate, extra_args, metadata)

def setup(self, fig, outfile, dpi=None, frame_dir=None):

outfile = Path(outfile)

_api.check_in_list(['.html', '.htm'], outfile_extension=outfile.suffix)

self._saved_frames = []

self._total_bytes = 0

self._hit_limit = False

if not self.embed_frames:

if frame_dir is None:

frame_dir = outfile.with_name(outfile.stem + '_frames')

frame_dir.mkdir(parents=True, exist_ok=True)

frame_prefix = frame_dir / 'frame'

else:

frame_prefix = None

super().setup(fig, outfile, dpi, frame_prefix)

self._clear_temp = False

def grab_frame(self, **savefig_kwargs):

_validate_grabframe_kwargs(savefig_kwargs)

if self.embed_frames:

# Just stop processing if we hit the limit

if self._hit_limit:

return

f = BytesIO()

self.fig.savefig(f, format=self.frame_format,

dpi=self.dpi, **savefig_kwargs)

imgdata64 = base64.encodebytes(f.getvalue()).decode('ascii')

self._total_bytes += len(imgdata64)

if self._total_bytes >= self._bytes_limit:

_log.warning(

"Animation size has reached %s bytes, exceeding the limit "

"of %s. If you're sure you want a larger animation "

"embedded, set the animation.embed_limit rc parameter to "

"a larger value (in MB). This and further frames will be "

"dropped.", self._total_bytes, self._bytes_limit)

self._hit_limit = True

else:

self._saved_frames.append(imgdata64)

else:

return super().grab_frame(**savefig_kwargs)

def finish(self):

# save the frames to an html file

if self.embed_frames:

fill_frames = _embedded_frames(self._saved_frames,

self.frame_format)

frame_count = len(self._saved_frames)

else:

# temp names is filled by FileMovieWriter

frame_count = len(self._temp_paths)

fill_frames = _included_frames(

frame_count, self.frame_format,

self._temp_paths[0].parent.relative_to(self.outfile.parent))

mode_dict = dict(once_checked='',

loop_checked='',

reflect_checked='')

mode_dict[self.default_mode + '_checked'] = 'checked'

interval = 1000 // self.fps

with open(self.outfile, 'w') as of:

of.write(JS_INCLUDE + STYLE_INCLUDE)

of.write(DISPLAY_TEMPLATE.format(id=uuid.uuid4().hex,

Nframes=frame_count,

fill_frames=fill_frames,

interval=interval,

**mode_dict))

# Duplicate the temporary file clean up logic from

# FileMovieWriter.finish. We cannot call the inherited version of

# finish because it assumes that there is a subprocess that we either

# need to call to merge many frames together or that there is a

# subprocess call that we need to clean up.

if self._tmpdir:

_log.debug('MovieWriter: clearing temporary path=%s', self._tmpdir)

"""

def __init__(self, fig, event_source, blit=False):

self._draw_was_started = False

self._fig = fig

# Disables blitting for backends that don't support it. This

# allows users to request it if available, but still have a

# fallback that works if it is not.

self._blit = blit and fig.canvas.supports_blit

# These are the basics of the animation. The frame sequence represents

# information for each frame of the animation and depends on how the

# drawing is handled by the subclasses. The event source fires events

# that cause the frame sequence to be iterated.

self.frame_seq = self.new_frame_seq()

self.event_source = event_source

self.event_source.add_callback(self._step)

# Instead of starting the event source now, we connect to the figure's

# draw_event, so that we only start once the figure has been drawn.

self._first_draw_id = fig.canvas.mpl_connect('draw_event', self._start)

# Connect to the figure's close_event so that we don't continue to

# fire events and try to draw to a deleted figure.

self._close_id = self._fig.canvas.mpl_connect('close_event',

self._stop)

if self._blit:

self._setup_blit()

def __del__(self):

if not getattr(self, '_draw_was_started', True):

warnings.warn(

'Animation was deleted without rendering anything. This is '

'most likely not intended. To prevent deletion, assign the '

'Animation to a variable, e.g. `anim`, that exists until you '

'output the Animation using `plt.show()` or '

'`anim.save()`.'

)

def _start(self, *args):

"""

# Do not start the event source if saving() it.

if self._fig.canvas.is_saving():

return

# First disconnect our draw event handler

self._fig.canvas.mpl_disconnect(self._first_draw_id)

# Now do any initial draw

self._init_draw()

# Actually start the event_source.

self.event_source.start()

def _stop(self, *args):

# On stop we disconnect all of our events.

if self._blit:

self._fig.canvas.mpl_disconnect(self._resize_id)

self._fig.canvas.mpl_disconnect(self._close_id)

self.event_source.remove_callback(self._step)

self.event_source = None

def save(self, filename, writer=None, fps=None, dpi=None, codec=None,

bitrate=None, extra_args=None, metadata=None, extra_anim=None,

savefig_kwargs=None, *, progress_callback=None):

"""