Python library

shapeflow.__version__ = '0.4.4'

Library version

shapeflow

Common elements.

• Application settings

• Logging

• Caching

shapeflow.ROOTDIR = PosixPath('/home/docs/.local/share/shapeflow')

Root directory of the application.

Linux: /home/<user>/.local/share/shapeflow

Windows: C:\Users\<user>\AppData\Roaming\shapeflow

class shapeflow._Settings

Bases: pydantic.main.BaseModel

Abstract application settings

to_dict() → dict

Returns

Application settings as a dict

Return type

dict

override(overrides: dict)

Override some parameters of the settings in a context. Settings will only be modified within this context and restored to their previous values afterwards. Usage:

with settings.override({"parameter": "override value"}):
    <do something>

Parameters

overrides (dict) – A dict mapping field names to values with which to override those fields

classmethod schema(by_alias: bool = True, ref_template: str = '') → Dict[str, Any]

Inject title & description into pydantic schema.

These get lost due to some bug with Enum.

class shapeflow.FormatSettings

Bases: shapeflow._Settings

Formatting settings

datetime_format: str

Base datetime format string. Defaults to '%Y/%m/%d %H:%M:%S.%f'.

datetime_format_fs: str

Filesystem-safe datetime format string. Used to append date & time to file names. Defaults to '%Y-%m-%d_%H-%M-%S'.

class shapeflow.LoggingLevel(value)

Bases: str, enum.Enum

Logging level.

critical = 'critical'

Only log critical (unrecoverable) errors

error = 'error'

Only log errors

warning = 'warning'

Only log warnings (or errors)

info = 'info'

Log general information

debug = 'debug'

Log debugging information

vdebug = 'vdebug'

Log verbose debugging information

property level: int

Return the int logging level for compatibility with built-in logging library

class shapeflow.LogSettings

Bases: shapeflow._Settings

Logging settings

path: pydantic.types.FilePath

The application logs to this file

dir: pydantic.types.DirectoryPath

This is the log directory. Logs from previous runs are stored here.

keep: int

The applications stores a number of old logs.

When the amount of log files in shapeflow.LogSettings.dir exceeds this number, the oldest files are deleted.

lvl_console: shapeflow.LoggingLevel

The level at which the application logs to the Python console.

Defaults to info to keep the console from getting too spammy. Set to a lower level such as debug to show more detailed logs in the console.

lvl_file: shapeflow.LoggingLevel

The level at which the application logs to the log file at path.

Defaults to shapeflow.LoggingLevel.debug.

class shapeflow.CacheSettings

Bases: shapeflow._Settings

Caching settings

do_cache: bool

Enables the cache. Set to True by default.

Disabling the cache will make the application significantly slower.

dir: pydantic.types.DirectoryPath

Where to keep the cache

size_limit_gb: float

How big the cache is allowed to get

resolve_frame_number: bool

Whether to resolve frame numbers to the nearest requested frame numbers.

Increases seeking performance, but may make it a bit more ‘jumpy’ if a low number of frames is requested for an analysis.

block_timeout: float

How long to keep waiting for data that’s actively being committed to the cache before giving up and computing it instead.

In the rare case that two cachable data requests are being processed at the same time, the first request will block the cache for those specific data and cause the second request to wait until it can grab this data from the cache. This timeout prevents the second request from waiting forever until the first request finishes (for example, in case it crashes).

reset_on_error: bool

Clear the cache if it can’t be opened.

In rare cases, diskcache may cache data in a way it can’t read back. To recover from such an error, the cache will be cleared completely. The only downside to this is decreased performance for a short while.

class shapeflow.RenderSettings

Bases: shapeflow._Settings

Rendering settings

dir: pydantic.types.DirectoryPath

The directory where SVG files should be rendered to

keep: bool

Keep rendered images after they’ve been used.

Disabled by default, you may want to enable this if you want to inspect the renders.

class shapeflow.DatabaseSettings

Bases: shapeflow._Settings

Database settings.

path: pydantic.types.FilePath

The path to the database file

cleanup_interval: int

The database can get cluttered after a while, and will be cleaned at this interval

class shapeflow.ResultSaveMode(value)

Bases: str, enum.Enum

Where (or whether) to save the results of an analysis

skip = 'skip'

Don’t save results at all

next_to_video = 'next to video file'

Save results in the same directory as the video file that was analyzed

next_to_design = 'next to design file'

Save results in the same directory as the design file that was analyzed

directory = 'in result directory'

Save results in their own directory at shapeflow.ApplicationSettings.result_dir

class shapeflow.ApplicationSettings

Bases: shapeflow._Settings

Application settings.

save_state: bool

Whether to save the application state when exiting the application

load_state: bool

Whether to load the application state when starting the application

state_path: pydantic.types.FilePath

Where to save the application state

recent_files: int

The number of recent files to show in the user interface

video_pattern: str

Recognized video file extensions. Defaults to "*.mp4 *.avi *.mov *.mpv *.mkv".

design_pattern: str

Recognized design file extensions. Defaults to "*.svg".

save_result_auto: shapeflow.ResultSaveMode

Where or whether to save results after each run of an analysis

save_result_manual: shapeflow.ResultSaveMode

Where or whether to save results that are exported manually via the user interface

result_dir: pydantic.types.DirectoryPath

The path to the result directory

cancel_on_q_stop: bool

Whether to cancel the currently running analysis when stopping a queue.

Defaults to False, i.e. the currently running analysis will be completed first.

class shapeflow.Settings

Bases: shapeflow._Settings

shapeflow settings.

• app: ApplicationSettings

• log: LogSettings

• cache: CacheSettings

• render: RenderSettings

• format: FormatSettings

• db: DatabaseSettings

shapeflow.settings = shapeflow.Settings()

This global Settings object is used throughout the library

shapeflow.save_settings(path: str = '/home/docs/.local/share/shapeflow/settings.yaml')

Save settings to .yaml

shapeflow.update_settings(s: dict) → dict

Update the global settings object.

Note

Just doing settings = Settings(**new_settings_dict) would prevent other modules from accessing the updated settings!

Parameters

s (dict) – new settings to integrate into the global settings

Returns

the current global settings as a dict

Return type

dict

class shapeflow.Logger(name, level=0)

shapeflow logger.

• Adds a verbose debug logging level vdebug()

• Strips newlines from log output to keep each log event on its own line

shapeflow.get_logger(name: str) → shapeflow.Logger

Get a new Logger object

Parameters

name (str) – name of the logger

Returns

a fresh logging handle

Return type

Logger

shapeflow.get_cache(retry: bool = False) → diskcache.core.Cache

Get a new diskcache.Cache object In some rare cases this can fail due to a corrupt cache. If settings.cache.reset_on_error is on, and an exception is raised the cache directory is removed and get_cache() is called again with retry set to True.

Parameters

retry (bool) – Whether this call is a “retry call”. Defaults to False, i.e. a first call

Returns

a fresh cache handle

Return type

diskcache.Cache

util

General utility functions & classes.

class shapeflow.util.Timing(t0, t1, elapsed)

Bases: tuple

property elapsed

Alias for field number 2

property t0

Alias for field number 0

property t1

Alias for field number 1

shapeflow.util.timed(f, logger: logging.Logger)

Function decorator to measure elapsed time. Useful for profiling and debugging.

Parameters

• f – Any function or method

• logger (Logger) – A logger to log to

shapeflow.util.logged(f, logger: logging.Logger)

Function decorator to log before & after a function call. Useful for profiling and debugging.

Parameters

• f – Any function or method

• logger (Logger) – A logger to log to

class shapeflow.util.Timer(parent: object, logger: logging.Logger)

Bases: object

A timer context manager. Starts timing on __enter__ and stops on __exit__. Timing info can be retrieved with timing.

property timing: Optional[tuple]

start time, end time, elapsed time.

Type

Optional[tuple]

shapeflow.util.restrict(val, minval, maxval)

Clamp a value between a minimum and a maximum.

shapeflow.util.frame_number_iterator(total: int, Nf: Optional[int] = None, dt: Optional[float] = None, fps: Optional[float] = None) → Generator[int, None, None]

Get an iterator of frame numbers, based on either a number of frames Nf or a frame interval dt.

Parameters

• total (int) – The total number of frames.

• Nf (int) – The number of frames to return. Defaults to None.

• dt (float) – The frame interval to return. Defaults to None. If using dt, fps must also be provided.

• fps (float) – The framerate. Defaults to None

Raises

ValueError – When both Nf and dt are None

Returns

An iterator that returns the requested frame numbers.

Return type

Generator

shapeflow.util.before_version(version_a, version_b)

Check whether version_a precedes version_b.

Note

Only handles numerics, i.e. "1.25b.3v7" won’t work.

shapeflow.util.after_version(version_a, version_b)

Check whether version_a supercedes version_b.

Note

Only handles numerics, i.e. "1.25b.3v7" won’t work.

shapeflow.util.hash_file(path: str, blocksize: int = 1024) → queue.Queue

Start hashing a file without blocking the current thread.

Parameters

• path (str) – The path of the file to hash.

• blocksize (int) – The blocksize step to take when reading the file. Defaults to 1024.

Returns

A new queue.Queue object. Once the hash is ready, it will be pushed to this queue.

Return type

queue.Queue

shapeflow.util.suppress_stdout()

Suppress stdout within a context.

https://stackoverflow.com/questions/2125702/

shapeflow.util.sizeof_fmt(size, suffix='B')

Get a file size in bytes as a human-readable string. For example:

>>> sizeof_fmt(10**3)
"1 KB"
>>> sizeof_fmt(10**9)
"1 GB"

Parameters

• num – A file size in bytes

• suffix (str) – The suffix to use. Defaults to "B".

Returns

The file size as a human-readable string in decimal bytes.

Return type

str

class shapeflow.util.Singleton

Bases: type

A metaclass for singletons.

shapeflow.util.open_path(path: str) → None

Open a path in the file browser.

shapeflow.util.ensure_path(path: Union[str, pathlib.Path])

A hacky way to allow imports from arbitrary directories. Only use this for testing sf.py please :(

shapeflow.util.meta.describe_function(f) → str

Return a function description. More specific than regular old __qualname__.

Note

For nested functions, __qualname__ contains <locals>. diskcache.Cache can write keys containing < or > but crashes when trying to read these back. To prevent weird bugs, < and > are replaced with _

Parameters

f – Any function or method

Returns

A unique name for this method

Return type

str

shapeflow.util.meta.bases(c: type) → List[type]

Returns the bases of a class, including the bases of its bases.

Note

Just in general, don’t do list(set()) if the order is important!

Parameters

c (type) – Any class

Returns

A list of all bases of c

Return type

List[type]

shapeflow.util.meta.unbind(m)

Unbind a method from its instance.

Parameters

m – Any method

Returns

Return type

The method m, not bound to any instance

shapeflow.util.meta.bind(instance, m)

Bind a method to an instance.

https://stackoverflow.com/a/1015405/12259362

Parameters

• instance – An object

• m – Any bound or unbound method of the class of instance

Returns

The method m, bound to instance

Return type

Callable

shapeflow.util.from_venv.from_venv(env: str) → None

Re-run the current script from a virtual environment.

Gathers call arguments from sys.argv and starts a child process from the virtual environment.

Parameters

env (str) – The directory of the virtual environment.

shapeflow.util.filedialog.filedialog: shapeflow.util.filedialog._FileDialog = <shapeflow.util.filedialog._SubprocessTkinter object>

Cross-platform file dialog.

Tries to use zenity if available, because tkinter looks fugly in GNOME don’t @ me.

Defaults to tkinter to support basically any platform.

core

exception shapeflow.core.RootException(*args)

Bases: Exception

All shapeflow exceptions should be subclasses of this one. Automatically logs the exception class and message at the ERROR level.

msg = ''

The message to log

exception shapeflow.core.DispatchingError(*args)

Bases: shapeflow.core.RootException

An error dispatching a method call or exposing an endpoint.

class shapeflow.core.EnforcedStr(string: Optional[str] = None)

Bases: str

A string that is enforced to be one of several options. Works like a dynamic Enum – options can be added at runtime.

property options

The accepted options

property descriptions

The descriptions of each option

property describe

The description of the currently selected option

property default

The default option for this EnforcedStr

classmethod set_default(value: shapeflow.core.EnforcedStr) → None

Explicitly sets the default.

Parameters

value (EnforcedStr) – The default value to set

classmethod __modify_schema__(field_schema)

Modify pydantic schema to include default, descriptions and act as an Enum

class shapeflow.core.Endpoint(signature: _GenericAlias, streaming: shapeflow.core._Streaming = <_Streaming 'off'>)

Bases: object

An endpoint for an internal method.

compatible(method: Callable) → bool

Checks whether a method is compatible with the endpoint’s signature

Parameters

method (Callable) – Any method or function

Returns

True if the method is compatible, False if it isn’t.

Return type

bool

expose()

Expose a method at this endpoint. Used as a decorator:

@endpoint.expose()
def some_method():
    pass

property method: Optional[Callable]

The method exposed at this endpoint. Can be None

property signature: tuple

The signature of this endpoint.

property streaming: shapeflow.core._Streaming

What or whether this endpoint streams.

property registered: bool

Whether this endpoint is registered.

property name: str

The name of this endpoint. Taken from its attribute name in the object where it is registered.

register(name: str, callback: Callable[[shapeflow.core.Endpoint], None])

Register the endpoint in some other object.

class shapeflow.core.Dispatcher(instance: Optional[object] = None)

Bases: object

Dispatches requests to shapeflow.core.Endpoint objects.

property name: str

The name of this dispatcher.

property dispatchers: Tuple[shapeflow.core.Dispatcher, ...]

The dispatchers nested in this dispatcher.

property endpoints: Tuple[shapeflow.core.Endpoint, ...]

The endpoints contained in this dispatcher.

property address_space: Dict[str, Optional[Callable]]

The address-method mapping of this dispatcher.

dispatch(address: str, *args, **kwargs) → Any

Dispatch a request to a method.

Parameters

• address (str) – The address to dispatch to

• args – Any positional arguments to pass on to the method

• kwargs – Any keyword arguments to pass on to the method

Returns

Whatever the method returns.

Return type

Any

class shapeflow.core.Described

Bases: object

A class with a description.

This description is taken from the first line of the docstring if there is one or set to the name of the class if there isn’t.

class shapeflow.core.Lockable

Bases: object

Wrapper around threading.Lock & threading.Event

Defines a lock context to handle locking and unlocking along with a _cancel and _error events to communicate with Lockable objects from other threads.

Doesn’t need to initialize; lock & events are created when they’re needed.

lock()

Locking context.

If _lock event doesn’t exist yet it is instantiated first. Upon exiting the context, the threading.Lock object is compared to the original to ensure that no shenanigans took place.

cancel()

Sets the _cancel event. If _cancel event doesn’t exist yet it is instantiated first.

error()

Sets the _error event. If _error event doesn’t exist yet it is instantiated first.

property canceled: bool

Returns True if the _cancel event is set. If _cancel event doesn’t exist yet it is instantiated first.

property errored: bool

Returns True if the _error event is set. If _error event doesn’t exist yet it is instantiated first.

clear_cancel()

Clears the _cancel event. If _cancel event doesn’t exist yet it is instantiated first.

clear_error()

Clears the _error event. If _error event doesn’t exist yet it is instantiated first.

class shapeflow.core.RootInstance

Bases: shapeflow.core.Lockable

streaming

Streaming various data over Flask.

class shapeflow.core.streaming.BaseStreamer

Bases: abc.ABC

Abstract streamer.

push(value: Any) → None

Push something to the stream.

Parameters

value (Any) – Anything, really. If it doesn’t work, you’ll hear it.

stream() → Generator[Any, None, None]

Start a stream.

Returns

A generator that can be returned as a Flask response.

Return type

Generator[Any, None, None]

stop() → None

Stop the stream.

classmethod mime_type() → str

Get the streamer’s MIME type (for Flask response).

classmethod content_type()

Get the streamer’s content type (for Flask response).

property headers

Get the streamer’s headers (for Flask response).

class shapeflow.core.streaming.PlainFileStreamer(path: str)

Bases: shapeflow.core.streaming.BaseStreamer

Streams plaintext files. Used to stream logs to the frontend.

property headers

Get the streamer’s headers (for Flask response).

stream() → Generator[Any, None, None]

Start a stream.

Returns

A generator that can be returned as a Flask response.

Return type

Generator[Any, None, None]

class shapeflow.core.streaming.JsonStreamer

Bases: shapeflow.core.streaming.BaseStreamer

Streams JSON data.

class shapeflow.core.streaming.EventStreamer

Bases: shapeflow.core.streaming.JsonStreamer

Streams server-sent events with JSON data.

event(category: str, id: str, data: Any)

Push a JSON event

Parameters

• category – event category

• id – UUID of event source

• data – event data

Returns

stop()

Stop the stream.

class shapeflow.core.streaming.FrameStreamer

Bases: shapeflow.core.streaming.BaseStreamer

Streams images. Subclasses can define specific encodings, for now JpegStreamer seems to work best.

class shapeflow.core.streaming.JpegStreamer

Bases: shapeflow.core.streaming.FrameStreamer

Streams JPEG images.

class shapeflow.core.streaming.StreamHandler(*args, **kwargs)

Bases: shapeflow.core.Lockable

Handles streaming of method return values

register(instance: object, method) → shapeflow.core.streaming.BaseStreamer

Register an instance/method combination, start a streamer. If this combination has been registered already, return its streamer.

Parameters

• instance (object) – Any instance that contains method

• method – Any method of instance that’s mapped to a streaming Endpoint object

Returns

A streamer

Return type

BaseStreamer

is_registered(instance: object, method=None) → bool

Check whether an instance/method combo is registered.

push(instance: object, method, data) → None

Push some data, if this instance/method combination is registered.

Parameters

• instance (object) – Any instance that contains method

• method – Any method of instance that’s mapped to a streaming Endpoint object

• data (Any) – Anything, really. If it doesn’t work, you’ll hear it.

unregister(instance: object, method=None) → None

Unregister an instance/method combination and stop streaming.

Parameters

• instance (object) – Any instance that contains method

• method – Any method of instance that’s mapped to a streaming Endpoint object. If None, unregister all methods for this instance.

update() → None

Update all streams.

For all registered streamers, invoke their method against their instance and push the return value.

stop() → None

Unregister everything and stop streaming altogether.

shapeflow.core.streaming.streams = <shapeflow.core.streaming.StreamHandler object>

Global StreamHandler instance.

shapeflow.core.streaming.stream(method)

Decorator to mark streaming methods.

Only works for methods exposed at Endpoint objects, should be placed above the expose() decorator:

from shapeflow.core import Endpoint, _Streaming

some_endpoint = Endpoint(Callable[[...], ...], _Streaming())

class SomeClass:
    @stream
    @some_endpoint.expose()
    def some_method(self, ...):
        pass

To actually stream, the wrapped method and an instance should be registered in shapeflow.core.streaming.streams.

db

class shapeflow.core.db.SessionWrapper

Bases: object

Wrapper object for a SQLAlchemy session factory.

connect(session_wrapper: shapeflow.core.db.SessionWrapper)

Share the session factory of another SessionWrapper instance

session()

SQLAlchemy session context manager.

Opens a SQLAlchemy session and commits after the block is done. Changes are rolled back if an exception is raised. Usage:

with self.session() as s:
    # interact with the database here

class shapeflow.core.db.DbModel(**kwargs)

Bases: sqlalchemy.orm.decl_api.Base, shapeflow.core.db.SessionWrapper, shapeflow.core.Lockable

Abstract database model class.

Subclasses should

get(attr: str) → Any

Get attribute value from database

session(add: bool = True)

SQLAlchemy session context manager.

Opens a SQLAlchemy session and commits after the block is done. Changes are rolled back if an exception is raised. Usage:

with self.session() as s:
    # interact with the database here

Calls DbModel._pre() before yielding the session and DbModel._post() after the block is completed.

Parameters

add (bool) – add model(s) after opening the session

class shapeflow.core.db.FileModel(path: str)

Bases: shapeflow.core.db.DbModel

Abstract database model for files.

Files are hashed and resolved in order to keep a single entry per file.

property resolved: bool

Whether the FileModel has been resolved

resolve() → shapeflow.core.db.FileModel

Resolve the file by its SHA1 hash ~ hash_file().

If the computed hash is new, the file is committed to the database. Otherwise, the original entry is re-used.

Returns

The current instance if the file is new, or a new FileModel instance representing the original database entry.

Return type

FileModel

class shapeflow.core.db.BaseAnalysisModel(**kwargs)

Bases: shapeflow.core.db.DbModel

AnalysisModel interface.

abstract get_name() → str

Get the name of the analysis

abstract get_config_json() → Optional[str]

Get the current configuration in JSON

abstract load_config(video_path: str, design_path: Optional[str] = None, include: Optional[List[str]] = None) → Optional[dict]

Load configuration from the database

abstract get_undo_config(context: Optional[str] = None) → Tuple[Optional[dict], Optional[int]]

Undo configuration. If a context is supplied, ensure that the context field changes, but the other fields remain the same

abstract get_redo_config(context: Optional[str] = None) → Tuple[Optional[dict], Optional[int]]

Redo configuration. If a context is supplied, ensure that the context field changes, but the other fields remain the same

abstract store() → None

Store analysis information from wrapped BaseVideoAnalyzer to the database

abstract export_result(run: Optional[int] = None) → None

Export a result from the database

abstract get_runs() → int

Get th number of runs of this analysis

abstract get_id() → int

Get the id of this analysis

config

class shapeflow.core.config.Factory(string: Optional[str] = None)

Bases: shapeflow.core.EnforcedStr

An enforced string which maps its options to types.

Included types should be subclasses of Described in order to generate descriptions for all options.

get() → Type[shapeflow.core.Described]

Get the type associated with the current string.

classmethod get_str(mapped_value: Type[shapeflow.core.Described])

Get the string for a specific type.

property options: List[str]

The options for this factory.

property descriptions: Dict[str, str]

The descriptions for this factory.

property default: Optional[str]

The default for this factory.

classmethod extend(key: str, extension: Type[shapeflow.core.Described])

Add a new type to this factory. Used to dynamically add options e.g. for including plugins.

abstract config_schema() → dict

The pydantic configuration schema for the members of this factory

class shapeflow.core.config.extend(factory: Type[shapeflow.core.config.Factory], key: Optional[str] = None)

Bases: object

Decorator to extend Factory classes. Usage:

from shapeflow.core.config import extend

@extend(SomeFactory)
class SomeClass:
    pass

shapeflow.core.config.untag(d: dict) → dict

Remove the tags from a configuration dict

Parameters

d (dict) – Any configuration dict

Returns

The original configuration dict without class and version info

Return type

dict

class shapeflow.core.config.BaseConfig

Bases: pydantic.main.BaseModel, shapeflow.core.Described

Abstract configuration class. All other configuration classes should derive from this one.

Usage, where SomeConfig is a subclass of BaseConfig:

# instantiating
config = SomeConfig()
config = SomeConfig(field1=1.0, field2='text')
config = SomeConfig(**dict_with_fields_and_values)

# updating
config(field1=1.0, field2='text')
config(**dict_with_fields_and_values)

# saving
dict_with_fields_and_values = config.to_dict()

When writing BaseConfig subclasses, use the extend decorator to make your configuration class accessible through the ConfigType factory. Configuration fields

are declared as pydantic.Field instances and must be type-annotated for type resolution to work properly.

Example:

from pydantic import Field
from shapeflow.core.config import BaseConfig

@extend(ConfigType)
class SomeConfig(BaseConfig):
    field1: int = Field(default=42)
    field2: SomeNestedConfig = Field(default_factory=SomeOtherConfig)

class Config

Bases: object

pydantic configuration class

classmethod _resolve_enforcedstr(value, field)

Resolve EnforcedStr objects from regular str objects. To be used in pydantic validators.

classmethod _odd_add(value)

Make sure a value stays odd by incrementing even values. To be used in pydantic validators.

classmethod _int_limits(value, field)

Enforce pydantic field limits (le, lt, ge, gt) for int fields. To be used in pydantic validators.

classmethod _float_limits(value, field)

Enforce pydantic field limits (le, lt, ge, gt) for float fields. To be used in pydantic validators.

to_dict(do_tag: bool = False) → dict

Return the configuration as a serializable dict.

Parameters

do_tag (bool) – If True, add configuration class and version fields to the dict

Returns

A serializable representation of this configuration object.

Return type

dict

tag(d: dict) → dict

Tag a dict with this object’s class and the library version. This information is used to deserialize correctly later on.

class shapeflow.core.config.ConfigType(string: Optional[str] = None)

Bases: shapeflow.core.config.Factory

Configuration type factory

get() → Type[shapeflow.core.config.BaseConfig]

Return the configuration type.

config_schema() → dict

Return the configuration schema.

class shapeflow.core.config.Configurable

Bases: shapeflow.core.Described

A class with an associated configuration type.

_config_class

The configuration class as a class attribute. When subclassing, set this attribute to a specific BaseConfig type to associate it with this class.

alias of shapeflow.core.config.BaseConfig

classmethod config_class()

The configuration class.

classmethod config_schema()

The configuration schema.

class shapeflow.core.config.Instance(config: Optional[shapeflow.core.config.BaseConfig] = None)

Bases: shapeflow.core.config.Configurable

interface

class shapeflow.core.interface.InterfaceType(string: Optional[str] = None)

Bases: shapeflow.core.config.Factory

Interface type factory.

get() → Type[shapeflow.core.config.Configurable]

Get the interface type.

config_schema() → dict

Get the config schema for this interface type.

classmethod __modify_schema__(field_schema)

Modify pydantic schema to include the interface.

class shapeflow.core.interface.Handler

Bases: abc.ABC

Abstract class to wrap an interface implementation and its configuration

set_implementation(implementation: str) → str

Set the implementation.

get_implementation() → str

Get the current implementation.

class shapeflow.core.interface.HandlerConfig

Bases: shapeflow.core.config.BaseConfig, abc.ABC

Abstract handler configuration.

type: shapeflow.core.interface.InterfaceType

The handler’s interface implementation.

data: shapeflow.core.config.BaseConfig

The implementation’s configuration.

class shapeflow.core.interface.TransformConfig

Bases: shapeflow.core.config.BaseConfig

Abstract transform configuration.

class shapeflow.core.interface.TransformInterface

Bases: shapeflow.core.config.Configurable

Transform interface. Handles transforming frames based on ROI coordinates.

All transform plugins should derive from this class and implement its methods.

abstract validate(matrix: Optional[numpy.ndarray]) → bool

Check whether a transformation matrix is valid.

Parameters

matrix (Optional[np.ndarray]) – A transformation matrix or None if not set yet.

abstract from_coordinates(roi: shapeflow.maths.coordinates.Roi, from_shape: tuple) → numpy.ndarray

Make relative video-space coordinates absolute and compatible with numpy / OpenCV.

Parameters

• roi (Roi) – A region of interest in the video (in relative coordinates)

• from_shape (tuple) – The dimensions of the video (in pixels)

Returns

An array of absolute video-space coordinates

Return type

np.ndarray

abstract to_coordinates(to_shape: tuple) → numpy.ndarray

Get the edges of the design in design-space coordinates

Parameters

to_shape (tuple) – The dimensions of the design (in pixels)

Returns

An array of absolute design-space coordinates

Return type

np.ndarray

abstract estimate(roi: shapeflow.maths.coordinates.Roi, from_shape: tuple, to_shape: tuple) → Optional[numpy.ndarray]

Estimate the transformation matrix from from_coordinates() to to_coordinates()

Parameters

• roi (Roi) – A region of interest in the video (in relative coordinates)

• from_shape (tuple) – The dimensions of the video (in pixels)

• to_shape (tuple) – The dimensions of the design (in pixels)

Returns

The estimated transformation matrix. Returns None if one of the arguments is invalid or if some other exception occurs.

Return type

Optional[np.ndarry]

invert(matrix: Optional[numpy.ndarray]) → Optional[numpy.ndarray]

Invert a transformation matrix

Parameters

matrix (Optional[np.ndarray]) – The transformation matrix to invert. Can be None if not set yet.

Returns

The inverse transformation matrix. Returns None if matrix is None or if some exception occurs during inversion.

Return type

Optional[np.ndarray]

abstract transform(matrix: numpy.ndarray, img: numpy.ndarray, shape: Tuple[int, int]) → numpy.ndarray

Transform a frame from video-space to design-space.

Parameters

• matrix (np.ndarray) – The transformation matrix

• img (np.ndarray) – The frame to transform

• shape (tuple) – The shape of the design to transform to

Returns

The transformed frame

Return type

np.ndarray

abstract coordinate(matrix: numpy.ndarray, coordinate: shapeflow.maths.coordinates.ShapeCoo, shape: Tuple[int, int]) → shapeflow.maths.coordinates.ShapeCoo

Transform an (x,y) coordinate from video-space to design-space.

Parameters

• matrix (np.ndarray) – The transformation matrix

• coordinate (ShapeCoo) – The coordinate to transform

• shape (tuple) – The shape of the design to transform to

Returns

The transformed coordinate

Return type

ShapeCoo

class shapeflow.core.interface.FilterConfig

Bases: shapeflow.core.config.BaseConfig

Abstract filter configuration

property ready

Return true if filter can be applied ~ this configuration. Override for specific filter implementations

class shapeflow.core.interface.FilterInterface

Bases: shapeflow.core.config.Configurable

Filter interface. Handles filtering color images to binary.

All filter plugins should derive from this class and implement its methods.

abstract set_filter(filter, color: shapeflow.maths.colors.Color) → shapeflow.core.interface.FilterConfig

Set the filter to a specific color.

Parameters

• filter (FilterConfig) – The filter configuration

• color (Color) – The color to set the filter to

Returns

The updated filter configuration

Return type

FilterConfig

abstract mean_color(filter) → shapeflow.maths.colors.Color

Get the mean color from a filter configuration

Parameters

filter (FilterConfig) – The filter configuration

Returns

The configuration’s mean color

Return type

Color

abstract filter(filter, image: numpy.ndarray, mask: Optional[numpy.ndarray] = None) → numpy.ndarray

Filter a frame.

Parameters

• filter (FilterConfig) – The filter configuration

• image (np.ndarray) – The frame to filter

• mask (Optional[np.ndarray]) – The mask to apply to the frame

Returns

The filtered frame

Return type

np.ndarray

class shapeflow.core.interface.FilterType(string: Optional[str] = None)

Bases: shapeflow.core.interface.InterfaceType

Filter type factory.

get() → Type[shapeflow.core.interface.FilterInterface]

Get the filter type.

class shapeflow.core.interface.TransformType(string: Optional[str] = None)

Bases: shapeflow.core.interface.InterfaceType

Transform type factory.

get() → Type[shapeflow.core.interface.TransformInterface]

Get the transform type.

backend

exception shapeflow.core.backend.BackendSetupError(*args)

Bases: shapeflow.core.RootException

exception shapeflow.core.backend.BackendError(*args)

Bases: shapeflow.core.RootException

exception shapeflow.core.backend.CacheAccessError(*args)

Bases: shapeflow.core.RootException

class shapeflow.core.backend.PushEvent(value)

Bases: enum.Enum

Categories of server-pushed events.

class shapeflow.core.backend.AnalyzerState(value)

Bases: enum.IntEnum

The state of an analyzer

classmethod can_launch(state: int) → bool

Returns True if an analyzer can launch from this state

classmethod is_launched(state: int) → bool

Returns True if an analyzer is launched in this state

class shapeflow.core.backend.QueueState(value)

Bases: enum.IntEnum

The state of the analysis queue

class shapeflow.core.backend.CachingInstance(config: Optional[shapeflow.core.config.BaseConfig] = None)

Bases: shapeflow.core.config.Instance

Cache method results with diskcache.Cache

cached_call(method, *args, **kwargs)

Call a method or get the result from the cache if available.

Parameters

• method – The method to call

• args – Any positional arguments to pass on to the method

• kwargs – Any keyword arguments to pass on to the method

Returns

Whatever the method returns.

Return type

Any

class shapeflow.core.backend.FeatureConfig

Bases: shapeflow.core.config.BaseConfig

Abstract Feature parameters

class shapeflow.core.backend.Feature(elements: Tuple[shapeflow.core.config.Instance, ...], global_config: shapeflow.core.backend.FeatureConfig, config: Optional[dict] = None)

Bases: shapeflow.core.config.Configurable

A feature implements interactions between BackendElements to produce a certain value

_label: str = ''

Label string, to be used in exported data and the user interface

_unit: str = ''

Unit string, to be used in exported data and the user interface

calculate(frame: numpy.ndarray, state: Optional[numpy.ndarray] = None) → Tuple[Any, Optional[numpy.ndarray]]

Calculate the feature for the given frame

Parameters

• frame (np.ndarray) – A video frame

• state (Optional[np.ndarray]) – An np.ndarray for the state image. Should have the same dimensions as the frame

Returns

• Any – The calculated feature value

• Optional[np.ndarray] – If a state image was provided, return this state frame with the feature’s state frame added onto it. If not, return None.

classmethod label() → str

The label of this feature. Used in the user interface and results.

classmethod unit() → str

The unit of this feature. Used in the user interface and results.

abstract property skip: bool

Whether this feature should be skipped

abstract property ready: bool

Whether this feature is ready to be calculated

property color: shapeflow.maths.colors.Color

Color of the feature in figures.

A feature’s color is handled by its FeatureSet as not to overlap with any other features in that set.

abstract state(frame: numpy.ndarray, state: numpy.ndarray) → numpy.ndarray

Return the feature’s state image for a given frame

abstract value(frame: numpy.ndarray) → Any

Compute the value of the feature for a given frame

property config

The configuration of the feature. Default to the global configuration if no specific one is provided.

class shapeflow.core.backend.FeatureSet(features: Tuple[shapeflow.core.backend.Feature, ...])

Bases: shapeflow.core.config.Configurable

A set of Feature instances

resolve_colors() → Tuple[shapeflow.maths.colors.Color, ...]

Resolve the colors of all features in this set so that none of them overlap.

property colors: Tuple[shapeflow.maths.colors.Color, ...]

The resolved colors in this feature set

property features: Tuple[shapeflow.core.backend.Feature, ...]

The features in this feature set

calculate(frame: numpy.ndarray, state: Optional[numpy.ndarray]) → Tuple[List[Any], Optional[numpy.ndarray]]

Calculate all features in this set for a given frame

Parameters

• frame (np.ndarray) – An image

• state (Optional[np.ndarray]) – An empty np.ndarray for the state image. Should have the same dimensions as the frame

Returns

• List[Any] – The calculated feature values

• Optional[np.ndarray] – If a state image was provided, return the composite state image of this feature set. If not, return None.

class shapeflow.core.backend.FeatureType(string: Optional[str] = None)

Bases: shapeflow.core.interface.InterfaceType

Feature factory

get() → Type[shapeflow.core.backend.Feature]

Get the Feature for this feature type

config_schema() → dict

The pydantic configuration schema for this type of Feature

classmethod __modify_schema__(field_schema)

Modify pydantic schema to include units and labels.

class shapeflow.core.backend.BaseAnalyzerConfig

Bases: shapeflow.core.config.BaseConfig

Abstract analyzer configuration.

class shapeflow.core.backend.BaseAnalyzer(config: Optional[shapeflow.core.backend.BaseAnalyzerConfig] = None, eventstreamer: Optional[shapeflow.core.streaming.EventStreamer] = None)

Bases: shapeflow.core.config.Instance, shapeflow.core.RootInstance

Abstract analyzer.

property model

The database model associated with this analyzer.

property runs

The number of runs performed for this analysis.

property eventstreamer

Return the server-sent event streamer.

set_eventstreamer(eventstreamer: Optional[shapeflow.core.streaming.EventStreamer] = None)

Set the server-sent event streamer.

event(category: shapeflow.core.backend.PushEvent, data: dict) → None

Push an event.

Parameters

• category (PushEvent) – The category of event to push

• data (dict) – The data to push

notice(message: str, persist: bool = False) → None

Push a notice.

Parameters

• message (str) – The notice message to push

• persist (bool) – Whether the notice should be persistent (i.e. stay up on the user interface until dismissed manually)

commit() → bool

Save video analysis configuration to history database

shapeflow.api._VideoAnalyzerDispatcher.commit

abstract can_launch() → bool

Returns

Whether this analyzer can launch

Return type

bool

abstract can_filter() → bool

Returns

Whether this analyzer can filter

Return type

bool

abstract can_analyze() → bool

Returns

Whether this analyzer can analyze

Return type

bool

state_transition(push: bool = True) → int

Handle state transitions.

shapeflow.api._VideoAnalyzerDispatcher.state_transition

Parameters

push (bool) – Whether to push an event once the state is set.

Returns

The resulting state ~ AnalyzerState

Return type

int

cancel() → None

Cancel a running analysis.

shapeflow.api._VideoAnalyzerDispatcher.cancel

error() → None

Sets the _error event. If _error event doesn’t exist yet it is instantiated first.

abstract analyze() → bool

abstract property position: float

abstract property cached: bool

abstract property has_results: bool

status() → dict

Get the analyzer’s status.

shapeflow.api._VideoAnalyzerDispatcher.status

get_config(do_tag=False) → dict

Get the analyzer’s configuration.

shapeflow.api._VideoAnalyzerDispatcher.get_config

abstract set_config(config: dict, silent: bool = False) → dict

Set the analyzer’s configuration

shapeflow.api._VideoAnalyzerDispatcher.set_config

Parameters

• config (dict) – A configuration dict

• silent (bool) – If True, don’t push server-sent events. False by default.

Returns

The updated configuration dict. Some fields may have been changed due to incompatibility.

Return type

dict

launch() → bool

Launch the analyzer.

shapeflow.api._VideoAnalyzerDispatcher.launch

If the analyzer’s configuration is sufficiently filled out, the analyzer will instantiate any other objects it needs to start configuring the analysis further.

To launch, the analyzer needs at least

• A valid video file

• A valid design file

• At least one valid feature configuration

Returns

Whether the launch was successful or not.

Return type

bool

get_db_id() → int

Get the database id of this analyzer.

shapeflow.api._VideoAnalyzerDispatcher.get_db_id

time(message: str = '')

A timing context.

property timing: Optional[shapeflow.util.Timing]

Get the timing info from the latest run of time()

class shapeflow.core.backend.AnalyzerType(string: Optional[str] = None)

Bases: shapeflow.core.config.Factory

Analyzer type factory

get() → Type[shapeflow.core.backend.BaseAnalyzer]

Return the analyzer type.

config_schema() → dict

Return the config schema of the analyzer type.

maths

colors

Some basic tools for working with colors.

For now, only 8-bit integer colors are handled.

class shapeflow.maths.colors.Color

Bases: shapeflow.core.config.BaseConfig

An abstract color.

A pydantic data class; subclasses implementing specific colors should include their channels as three separate pydantic.Field attributes.

_colorspace: str = ''

The name of this colorspace.

_conversion_map: Dict[str, int] = {}

Conversion map from this color to other colors. Maps _colorspace strings to OpenCV conversion method identifiers.

property np3d: numpy.uint8

This color as a 3D numpy array. Used with OpenCV conversions and when multiplying with binary images to ‘color’ them.

property list: List[int]

This color as a list.

convert(colorspace: str) → tuple

Convert this color to another colorspace.

Parameters

colorspace (str) – The name of the colorspace to convert to. Will raise NotImplementedError if not in _conversion_map.

Returns

The converted color as a tuple.

Return type

tuple

classmethod from_str(color: str) → shapeflow.maths.colors.Color

Deserialize from a formatted string.

Parameters

color (str) – A color string formatted as "SomeColor(a=1,b=2,c=3)"

Returns

A new Color object

Return type

Color

classmethod normalize_channel(v: int) → int

Enforce 8-bit values [0-255] for a color channel. This is the default pydantic.validator for all channels of a color.

Parameters

v (int) – The original value of the channel

Returns

The value of the channel clamped to [0-255]

Return type

int

class shapeflow.maths.colors.HsvColor

Bases: shapeflow.maths.colors.Color

Hue-Saturation-Value color.

h: int

s: int

v: int

_colorspace: str = 'hsv'

The name of this colorspace.

_conversion_map: Dict[str, int] = {'bgr': 54, 'rgb': 55}

Conversion map from this color to other colors. Maps _colorspace strings to OpenCV conversion method identifiers.

classmethod hue_wraps(h: int) → int

Make sure the hue channel wraps at h=180.

Parameters

v (int) – The original value of the hue channel

Returns

The value of the hue channel wrapped around at h=180

Return type

int

class shapeflow.maths.colors.RgbColor

Bases: shapeflow.maths.colors.Color

Red-Green-Blue color.

r: int

g: int

b: int

_colorspace: str = 'rgb'

The name of this colorspace.

_conversion_map: Dict[str, int] = {'bgr': 4, 'hsv': 41}

Conversion map from this color to other colors. Maps _colorspace strings to OpenCV conversion method identifiers.

class shapeflow.maths.colors.BgrColor

Bases: shapeflow.maths.colors.Color

Blue-Green-Red color. This is the OpenCV default.

b: int

g: int

r: int

_colorspace: str = 'bgr'

The name of this colorspace.

_conversion_map: Dict[str, int] = {'hsv': 40, 'rgb': 4}

Conversion map from this color to other colors. Maps _colorspace strings to OpenCV conversion method identifiers.

shapeflow.maths.colors.convert(color: shapeflow.maths.colors.Color, to: Type[shapeflow.maths.colors.Color]) → shapeflow.maths.colors.Color

Convert a Color object to another Color type.

Parameters

• color (Color) – The original Color object

• to (Type[Color]) – The Color type to convert it to

Returns

The original Color object converted to the new Color type.

Return type

Color

shapeflow.maths.colors.as_hsv(color: shapeflow.maths.colors.Color) → shapeflow.maths.colors.HsvColor

Convert a color to HSV.

Parameters

color (Color) – Any Color object.

Returns

The original Color object as a HsvColor object.

Return type

HsvColor

shapeflow.maths.colors.as_bgr(color: shapeflow.maths.colors.Color) → shapeflow.maths.colors.BgrColor

Convert a color to BGR.

Parameters

color (Color) – Any Color object.

Returns

The original Color object as a BgrColor object.

Return type

BgrColor

shapeflow.maths.colors.as_rgb(color: shapeflow.maths.colors.Color) → shapeflow.maths.colors.RgbColor

Convert a color to RGB.

Parameters

color (Color) – Any Color object.

Returns

The original Color object as a RgbColor object.

Return type

RgbColor

shapeflow.maths.colors.complementary(color: shapeflow.maths.colors.Color) → shapeflow.maths.colors.Color

Get the complementary of a color. Takes a shortcut through HSV.

Parameters

color (Color) – Any Color object.

Returns

The complementary color in the same colorspace as the original one.

Return type

Color

shapeflow.maths.colors.css_hex(color: shapeflow.maths.colors.Color) → str

Get the color as a CSS-compatible hex RGB string:

>>> css_hex(RgbColor(r=170,g=187,b=204))
    "#aabbcc"

Parameters

color (Color) – Any Color object.

Returns

A hex RGB string

Return type

str

coordinates

Some basic tools for working with coordinates.

class shapeflow.maths.coordinates.Coo

Bases: shapeflow.core.config.BaseConfig

Coordinates of a point on a 2D image. Relative, and thus independent of the size of the image.

x: float

The x-coordinate.

y: float

The y coordinate.

property rel: Tuple[float, float]

Get the relative coordinates.

property list: list

Get the relative coordinates as a list.

class shapeflow.maths.coordinates.ShapeCoo

Bases: shapeflow.maths.coordinates.Coo

Coordinates of a point on a 2D image with a known size.

shape: Tuple[int, int]

The size of the image, from numpy.ndarray.shape.

property abs: Tuple[float, float]

Get the absolute coordinates.

property idx: Tuple[int, int]

Get (approximate) array indices ~ the absolute coordinate.

property cv2

Get the absolute coordinates for OpenCV.

Note

The order is flipped with respect to abs()

value(image: numpy.ndarray) → Any

Get the value (color) of an image at this coordinate (approximately).

Parameters

image (np.ndarray) – An image as a numpy array. Its shape should match shape.

Returns

The value of the image array at this coordinate.

Return type

Any

transform(matrix: numpy.ndarray, shape: Tuple[int, int]) → None

Transform this coordinate.

matrix: np.ndarray

Transformation matrix. Should be a 2D matrix with dimensions [2x2], [2x3], [3x3] or [4x4]

shape: Tuple[int, int]

class shapeflow.maths.coordinates.Roi

Bases: shapeflow.core.config.BaseConfig

A rectangular region of interest, composed of four points as Coo objects.

BL: shapeflow.maths.coordinates.Coo

The bottom-left corner.

TL: shapeflow.maths.coordinates.Coo

The top-left corner.

TR: shapeflow.maths.coordinates.Coo

The top-right corner.

BR: shapeflow.maths.coordinates.Coo

The bottom-right corner.

images

Some basic tools for working with images.

shapeflow.maths.images.ckernel(size: int) → numpy.ndarray

Circular/disk kernel.

Parameters

size (int) – The size or diameter of the kernel. Should be odd; if an even value is supplied, it will be decremented.

Returns

A disk kernel as a numpy array

Return type

np.ndarray

shapeflow.maths.images.overlay(frame: numpy.ndarray, overlay: numpy.ndarray, alpha: float = 0.5) → numpy.ndarray

Overlay frame image with overlay image. Both images should be in the BGR color space.

shapeflow.maths.images.crop_mask(mask: numpy.ndarray) → Tuple[numpy.ndarray, numpy.ndarray, Tuple[int, int]]

Crop a binary mask image array to its minimal (rectangular) size to exclude unnecessary regions.

shapeflow.maths.images.rect_contains(rect: numpy.ndarray, point: shapeflow.maths.coordinates.ShapeCoo) → bool

Check whether a point is contained in a rectangle.

Parameters

• rect (np.ndarray) – An ‘array rectangle’: [first_row, last_row, first_column, last_column]

• point (ShapeCoo) – A point on a 2D image with known size

Returns

Whether the point is contained in the rectangle or not.

Return type

bool

shapeflow.maths.images.mask(image: numpy.ndarray, mask: numpy.ndarray, rect: numpy.ndarray) → numpy.ndarray

Mask and crop off a part of an image.

Parameters

• image (np.ndarray) – The original image

• mask (np.ndarray) – The mask. Should be an OpenCV-compatible binary image, with False -> 0 and True -> 255

• rect (np.ndarray) – An ‘array rectangle’: [first_row, last_row, first_column, last_column]. Should correspond to the position of the mask in the original image.

Returns

The image cropped by rect and masked by mask

Return type

np.ndarray

shapeflow.maths.images.area_pixelsum(image: Optional[numpy.ndarray]) → Optional[int]

Get the total number of True pixels in a binary image.

Parameters

image (np.ndarray) – An OpenCV-compatible binary image, with False -> 0 and True -> 255.

Returns

The number of 255 pixels in the image

Return type

int

shapeflow.maths.images.to_mask(image: numpy.ndarray, kernel: Optional[numpy.ndarray] = None) → numpy.ndarray

Convert a PNG image to a binary mask array.

Parameters

• image (np.ndarray) – A numpy array representing a full-color PNG image without transparency

• kernel (Optional[np.ndarray]) – A smoothing kernel. If set to None, defaults to a ckernel() of size 7.

design

Design file handling

Adapted from https://github.com/ybnd/OnionSVG

shapeflow.design.check_design(file: Union[pathlib.Path, str]) → None

Check whether the file is a valid design file.

Parameters

file (Path) – Any file

Raises

SvgError – If the file can’t be read or parsed

shapeflow.design.peel(file: Union[pathlib.Path, str], dpi: int, dir: Union[pathlib.Path, str]) → None

Render a design layer per layer

Parameters

• file –

• dpi –

• dir –

onions

exception shapeflow.design.onions.DesignFileError(*args)

Bases: shapeflow.core.RootException

class shapeflow.design.onions.Layer(root: lxml.etree._Element, label: str)

Bases: object

Represents a layer within an SVG file

property label

hide() → None

Hide this layer

show() → None

Show this layer

class shapeflow.design.onions.Peeler(file: pathlib.Path)

Bases: object

Renders an SVG file into multiple PNG files layer-per-layer.

G = '{http://www.w3.org/2000/svg}g'

LABEL = '{http://www.inkscape.org/namespaces/inkscape}label'

NAMEDVIEW = '{http://sodipodi.sourceforge.net/DTD/sodipodi-0.dtd}namedview'

BG_COLOR = 'pagecolor'

BG_OPACITY = '{http://www.inkscape.org/namespaces/inkscape}pageopacity'

file: pathlib.Path

Path to the SVG file

peel(dpi: int, to_dir: pathlib.Path) → None

“Peel” the layers

render

exception shapeflow.design.render.RendererError(*args)

Bases: shapeflow.core.RootException

class shapeflow.design.render.Renderer

Bases: object

Renders SVG (as an XML string) to a PNG image. Multiple implementations are provided as a fallback for Windows systems where cairo may be unavailable.

Transparency is handled after importing PNG ~ OpenCV; SVGs are rendered with a white background. The background color of the original SVG file is ignored. Moreover, it’s no longer required to include a “hardcoded” background layer. Layers labelled with _* are ignored for backwards compatibility with legacy design files.

save(svg: bytes, dpi: int, to: pathlib.Path) → None

Save an SVG string to a PNG image. Checks whether this renderer works on the current system before trying.

Parameters

• svg (str) – An SVG image as an XML string

• dpi (int) – The DPI (details per inch) to render the SVG at

• to (Path) – The file to save to

property works: bool

class shapeflow.design.render.CairoRenderer

Bases: shapeflow.design.render.Renderer

First-choice renderer: fastest, cleanest interface.

Requires the Python package cairosvg to be installed, which in turn depends on cairo.

Installing cairo

Installing cairosvg

DEFAULT_DPI: int = 96

WHITE: str = '#ffffff'

cairosvg: Any

class shapeflow.design.render.WandRenderer

Bases: shapeflow.design.render.Renderer

Second-choice renderer: slower, but still a clean interface.

Requires the Python package Wand to be installed, which in turn depends on ImageMagick.

Installing ImageMagick

Installing Wand

Image: Any

background: Any

class shapeflow.design.render.InkscapeRenderer

Bases: shapeflow.design.render.Renderer

Fallback renderer: it works, but is hacky and bad.

Requires Inkscape to be installed.

shell: bool = False

class shapeflow.design.render.WindowsInkscapeRenderer

Bases: shapeflow.design.render.InkscapeRenderer

Fallback renderer for Windows: it works, but is hacky and bad.

Requires Inkscape to be installed.

INKSCAPE_DIR_CANDIDATES: List[pathlib.Path] = [PosixPath('C:\\Program Files (x86)\\Inkscape\\bin'), PosixPath('C:\\Program Files\\Inkscape\\bin')]

inkscape_dir: pathlib.Path

shell: bool = False

shapeflow.design.render.save_svg(svg: bytes, dpi: int, to: pathlib.Path) → None

Save SVG to a PNG image using the renderer selected for this system.

Parameters

• svg (bytes) – An SVG image as XML bytes

• dpi (int) – The DPI (dots per inch) to render the SVG at

• to (Path) – Where to save the PNG image

Raises

RendererError – if none of the available renderers work on this system

plugins

Transforms

PerspectiveTransform

class shapeflow.plugins.PerspectiveTransform._Config

Bases: shapeflow.core.interface.TransformConfig

class shapeflow.plugins.PerspectiveTransform._Transform

Bases: shapeflow.core.interface.TransformInterface

Wraps OpenCV’s getPerspectiveTransform function to estimate the transformation matrix and warpPerspective to apply it to a video frame or a coordinate.

validate(matrix: Optional[numpy.ndarray]) → bool

Check whether a transformation matrix is valid.

Parameters

matrix (Optional[np.ndarray]) – A transformation matrix or None if not set yet.

from_coordinates(roi: shapeflow.maths.coordinates.Roi, from_shape: tuple) → numpy.ndarray

Make relative video-space coordinates absolute and compatible with numpy / OpenCV.

Parameters

• roi (Roi) – A region of interest in the video (in relative coordinates)

• from_shape (tuple) – The dimensions of the video (in pixels)

Returns

An array of absolute video-space coordinates

Return type

np.ndarray

to_coordinates(to_shape: tuple) → numpy.ndarray

Get the edges of the design in design-space coordinates

Parameters

to_shape (tuple) – The dimensions of the design (in pixels)

Returns

An array of absolute design-space coordinates

Return type

np.ndarray

estimate(roi: shapeflow.maths.coordinates.Roi, from_shape: tuple, to_shape: tuple) → Optional[numpy.ndarray]

Estimate the transformation matrix from from_coordinates() to to_coordinates()

Parameters

• roi (Roi) – A region of interest in the video (in relative coordinates)

• from_shape (tuple) – The dimensions of the video (in pixels)

• to_shape (tuple) – The dimensions of the design (in pixels)

Returns

The estimated transformation matrix. Returns None if one of the arguments is invalid or if some other exception occurs.

Return type

Optional[np.ndarry]

transform(matrix: numpy.ndarray, img: numpy.ndarray, shape: tuple) → numpy.ndarray

Transform a frame from video-space to design-space.

Parameters

• matrix (np.ndarray) – The transformation matrix

• img (np.ndarray) – The frame to transform

• shape (tuple) – The shape of the design to transform to

Returns

The transformed frame

Return type

np.ndarray

coordinate(inverse: numpy.ndarray, coordinate: shapeflow.maths.coordinates.ShapeCoo, shape: Tuple[int, int]) → shapeflow.maths.coordinates.ShapeCoo

Transform an (x,y) coordinate from video-space to design-space.

Parameters

• matrix (np.ndarray) – The transformation matrix

• coordinate (ShapeCoo) – The coordinate to transform

• shape (tuple) – The shape of the design to transform to

Returns

The transformed coordinate

Return type

ShapeCoo

Filters

HsvRangeFilter

class shapeflow.plugins.HsvRangeFilter._Config

Bases: shapeflow.core.interface.FilterConfig

Configuration for shapeflow.plugins.HsvRangeFilter._Filter

color: shapeflow.maths.colors.HsvColor

The center color.

range: shapeflow.maths.colors.HsvColor

The range around the center color.

The default setting of HsvColor(h=10,s=75,v=75) works well for most cases, but if you notice false positive or false negative regions you can try adjusting the range.

Mixing or separating colors can be handled by increasing the h value (allowing more hues through) and uneven lighting/shadows can be compensated for by increasing the v value (lightness of the color).

close: int

Kernel size (circular) of a morphological closing operation.

If close is set to 0 (the default), no closing will be performed. This attribute will be coerced to an odd integer below 200 in order to keep performance somewhat reasonable.

You may want to configure a higher close if you notice that the state image of the corresponding frame includes noise (small objects or colored pixels) outside of its main area.

open: int

Kernel size (circular) of a morphological opening operation.

If open is set to 0 (the default), no opening will be performed. This attribute will be coerced to an odd integer below 200 in order to keep performance somewhat reasonable.

You may want to configure a higher open if you notice that the state image of the corresponding frame includes noise (small ‘holes’ or non-colored pixels) inside of its main area.

property ready: bool

Return true if filter can be applied ~ this configuration. Override for specific filter implementations

property c0: shapeflow.maths.colors.HsvColor

The center color minus the range.

property c1: shapeflow.maths.colors.HsvColor

The center color plus the range.

class shapeflow.plugins.HsvRangeFilter._Filter

Bases: shapeflow.core.interface.FilterInterface

Filters out colors outside of a HsvColor radius around a center color.

set_filter(filter: shapeflow.plugins.HsvRangeFilter._Config, color: shapeflow.maths.colors.Color) → shapeflow.plugins.HsvRangeFilter._Config

Set the filter to a specific color.

Parameters

• filter (FilterConfig) – The filter configuration

• color (Color) – The color to set the filter to

Returns

The updated filter configuration

Return type

FilterConfig

mean_color(filter: shapeflow.plugins.HsvRangeFilter._Config) → shapeflow.maths.colors.Color

Get the mean color from a filter configuration

Parameters

filter (FilterConfig) – The filter configuration

Returns

The configuration’s mean color

Return type

Color

filter(filter: shapeflow.plugins.HsvRangeFilter._Config, img: numpy.ndarray, mask: Optional[numpy.ndarray] = None) → numpy.ndarray

Filter a frame.

Parameters

• filter (FilterConfig) – The filter configuration

• image (np.ndarray) – The frame to filter

• mask (Optional[np.ndarray]) – The mask to apply to the frame

Returns

The filtered frame

Return type

np.ndarray

BackgroundFilter

class shapeflow.plugins.BackgroundFilter._Config

Bases: shapeflow.core.interface.FilterConfig

Configuration for shapeflow.plugins.BackgroundFilter._Filter

color: shapeflow.maths.colors.HsvColor

See shapeflow.plugins.HsvRangeFilter._Config.color

range: shapeflow.maths.colors.HsvColor

See shapeflow.plugins.HsvRangeFilter._Config.range

close: int

shapeflow.plugins.HsvRangeFilter._Config.close

open: int

shapeflow.plugins.HsvRangeFilter._Config.open

property ready: bool

Return true if filter can be applied ~ this configuration. Override for specific filter implementations

property c0: shapeflow.maths.colors.HsvColor

See shapeflow.plugins.HsvRangeFilter._Config.c0()

property c1: shapeflow.maths.colors.HsvColor

See shapeflow.plugins.HsvRangeFilter._Config.c1()

class shapeflow.plugins.BackgroundFilter._Filter

Bases: shapeflow.core.interface.FilterInterface

Filters out colors outside of a HsvColor radius around a center color and inverts the resulting image.

set_filter(filter: shapeflow.plugins.BackgroundFilter._Config, color: shapeflow.maths.colors.Color) → shapeflow.plugins.BackgroundFilter._Config

Set the filter to a specific color.

Parameters

• filter (FilterConfig) – The filter configuration

• color (Color) – The color to set the filter to

Returns

The updated filter configuration

Return type

FilterConfig

mean_color(filter: shapeflow.plugins.BackgroundFilter._Config) → shapeflow.maths.colors.Color

Get the mean color from a filter configuration

Parameters

filter (FilterConfig) – The filter configuration

Returns

The configuration’s mean color

Return type

Color

filter(filter: shapeflow.plugins.BackgroundFilter._Config, img: numpy.ndarray, mask: Optional[numpy.ndarray] = None) → numpy.ndarray

Filter a frame.

Parameters

• filter (FilterConfig) – The filter configuration

• image (np.ndarray) – The frame to filter

• mask (Optional[np.ndarray]) – The mask to apply to the frame

Returns

The filtered frame

Return type

np.ndarray

Features

PixelSum

class shapeflow.plugins.PixelSum._Feature(mask: shapeflow.video.Mask, global_config: shapeflow.core.backend.FeatureConfig, config: Optional[dict] = None)

Bases: shapeflow.video.MaskFunction

The most basic feature: it just returns the number of True pixels the filtered frame.

Area_mm2

class shapeflow.plugins.Area_mm2._Feature(mask: shapeflow.video.Mask, global_config: shapeflow.core.backend.FeatureConfig, config: Optional[dict] = None)

Bases: shapeflow.video.MaskFunction

Convert PixelSum to an area in mm², taking into account the DPI of the design file.

Volume_uL

class shapeflow.plugins.Volume_uL._Config

Bases: shapeflow.core.backend.FeatureConfig

Configuration for shapeflow.plugins.Volume_uL._Feature

h: float

The channel height of the chip.

class shapeflow.plugins.Volume_uL._Feature(mask: shapeflow.video.Mask, global_config: shapeflow.core.backend.FeatureConfig, config: Optional[dict] = None)

Bases: shapeflow.video.MaskFunction

Multiply Area_mm2 by a channel height in mm to estimate the volume in µL.

api

class shapeflow.api._VideoAnalyzerDispatcher(instance: Optional[object] = None)

Bases: shapeflow.core.Dispatcher

Dispatches /api/va/<id>/<endpoint>

status = <shapeflow.core.Endpoint object>

Get the analyzer’s status

shapeflow.core.backend.BaseAnalyzer.status()

state_transition = <shapeflow.core.Endpoint object>

Trigger a state transition

shapeflow.core.backend.BaseAnalyzer.state_transition()

can_launch = <shapeflow.core.Endpoint object>

Returns True if the analyzer has enough of its configuration set up to launch

shapeflow.video.VideoAnalyzer.can_launch()

can_analyze = <shapeflow.core.Endpoint object>

Returns True if the analyzer has enough of its configuration set up to analyze

shapeflow.video.VideoAnalyzer.can_analyze()

launch = <shapeflow.core.Endpoint object>

Launch the analyzer

shapeflow.core.backend.BaseAnalyzer.launch()

commit = <shapeflow.core.Endpoint object>

Commit the analyzer to the database

shapeflow.core.backend.BaseAnalyzer.commit()

analyze = <shapeflow.core.Endpoint object>

Run an analysis

shapeflow.video.VideoAnalyzer.analyze()

cancel = <shapeflow.core.Endpoint object>

Cancel an analysis

shapeflow.core.backend.BaseAnalyzer.cancel()

get_config = <shapeflow.core.Endpoint object>

Return the analyzer’s configuration

shapeflow.core.backend.BaseAnalyzer.get_config()

set_config = <shapeflow.core.Endpoint object>

Set the analyzer’s configuration

shapeflow.video.VideoAnalyzer.set_config()

undo_config = <shapeflow.core.Endpoint object>

Undo the latest change to the analyzer’s configuration

shapeflow.video.VideoAnalyzer.undo_config()

redo_config = <shapeflow.core.Endpoint object>

Redo the latest undone change to the analyzer’s configuration

shapeflow.video.VideoAnalyzer.redo_config()

estimate_transform = <shapeflow.core.Endpoint object>

Estimate a transform based on the provided ROI

shapeflow.video.VideoAnalyzer.estimate_transform()

turn_cw = <shapeflow.core.Endpoint object>

Turn the ROI clockwise

shapeflow.video.VideoAnalyzer.turn_cw()

turn_ccw = <shapeflow.core.Endpoint object>

Turn the ROI counter-clockwise

shapeflow.video.VideoAnalyzer.turn_ccw()

flip_h = <shapeflow.core.Endpoint object>

Flip the ROI horizontally

shapeflow.video.VideoAnalyzer.flip_h()

flip_v = <shapeflow.core.Endpoint object>

Flip the ROI vertically

shapeflow.video.VideoAnalyzer.flip_v()

clear_roi = <shapeflow.core.Endpoint object>

Clear the ROI

shapeflow.video.VideoAnalyzer.clear_roi()

get_overlay_png = <shapeflow.core.Endpoint object>

Return the overlay image

shapeflow.video.VideoAnalyzer.get_overlay_png()

get_frame = <shapeflow.core.Endpoint object>

Return the transformed frame at the provided frame number (or the current frame number if None)

shapeflow.video.VideoAnalyzer.get_transformed_frame()

set_filter_click = <shapeflow.core.Endpoint object>

Configure a filter based on a click position (in ROI-relative coordinates)

shapeflow.video.VideoAnalyzer.set_filter_click()

get_inverse_transformed_overlay = <shapeflow.core.Endpoint object>

Return the inverse transformed overlay image

shapeflow.video.VideoAnalyzer.get_inverse_transformed_overlay()

get_inverse_overlaid_frame = <shapeflow.core.Endpoint object>

Return the inverse overlaid frame at the provided frame number (or the current frame number if None)

shapeflow.video.VideoAnalyzer.get_inverse_overlaid_frame()

get_state_frame = <shapeflow.core.Endpoint object>

Return the state frame at the provided frame number (or the current frame number if None)

shapeflow.video.VideoAnalyzer.get_state_frame()

get_colors = <shapeflow.core.Endpoint object>

Return the color list

shapeflow.video.VideoAnalyzer.get_colors()

get_db_id = <shapeflow.core.Endpoint object>

Return the database ID associated with this analyzer

shapeflow.core.backend.BaseAnalyzer.get_db_id()

clear_filters = <shapeflow.core.Endpoint object>

Clear all filter configuration

shapeflow.video.VideoAnalyzer.clear_filters()

seek = <shapeflow.core.Endpoint object>

Seek to the provided position (relative, 0-1)

shapeflow.video.VideoAnalyzer.seek()

get_relative_roi = <shapeflow.core.Endpoint object>

Return the current ROI in relative coordinates

shapeflow.video.VideoAnalyzer.get_relative_roi()

get_coordinates = <shapeflow.core.Endpoint object>

Return the current relative coordinates

shapeflow.video.VideoAnalyzer.get_coordinates()

get_time = <shapeflow.core.Endpoint object>

Return the current time in the video in seconds

shapeflow.video.VideoAnalyzer.get_time()

get_total_time = <shapeflow.core.Endpoint object>

Return the total time of the video file in seconds

shapeflow.video.VideoAnalyzer.get_total_time()

get_fps = <shapeflow.core.Endpoint object>

Return the framerate of the video file

shapeflow.video.VideoAnalyzer.get_fps()

get_raw_frame = <shapeflow.core.Endpoint object>

Return the raw frame at the provided frame number (or the current frame number if None)

shapeflow.video.VideoAnalyzer.read_frame()

get_seek_position = <shapeflow.core.Endpoint object>

Return the current seek position (relative, 0-1)

shapeflow.video.VideoAnalyzer.get_seek_position()

class shapeflow.api._VideoAnalyzerManagerDispatcher(instance: Optional[object] = None)

Bases: shapeflow.core.Dispatcher

Dispatches /api/va/<endpoint>, active analyzers are handled by dispatchers at /api/va/<id>

init = <shapeflow.core.Endpoint object>

Initialize a new analyzer

shapeflow.main._VideoAnalyzerManager.init()

close = <shapeflow.core.Endpoint object>

Close an analyzer

shapeflow.main._VideoAnalyzerManager.close()

start = <shapeflow.core.Endpoint object>

Start analyzing the queue provided as a list of ID strings

shapeflow.main._VideoAnalyzerManager.q_start()

stop = <shapeflow.core.Endpoint object>

Stop the queue

shapeflow.main._VideoAnalyzerManager.q_stop()

cancel = <shapeflow.core.Endpoint object>

Cancel the queue

shapeflow.main._VideoAnalyzerManager.q_cancel()

state = <shapeflow.core.Endpoint object>

Return the application state

shapeflow.main._VideoAnalyzerManager.state()

save_state = <shapeflow.core.Endpoint object>

Save the application state to disk

shapeflow.main._VideoAnalyzerManager.save_state()

load_state = <shapeflow.core.Endpoint object>

Load the application state from disk

shapeflow.main._VideoAnalyzerManager.load_state()

stream = <shapeflow.core.Endpoint object>

Open a new stream for a given analyzer ID and endpoint

shapeflow.main._VideoAnalyzerManager.stream()

stream_stop = <shapeflow.core.Endpoint object>

Close the stream for a given analyzer ID and endpoint

shapeflow.main._VideoAnalyzerManager.stream_stop()

class shapeflow.api._DatabaseDispatcher(instance: Optional[object] = None)

Bases: shapeflow.core.Dispatcher

Dispatches /api/db/<endpoint>

get_recent_paths = <shapeflow.core.Endpoint object>

Get a list of recent video and design files

shapeflow.db.History.get_paths()

get_result_list = <shapeflow.core.Endpoint object>

Get a list of result IDs for a given analyzer ID

shapeflow.db.History.get_result_list()

get_result = <shapeflow.core.Endpoint object>

Get the result for a given analyzer ID and result ID

shapeflow.db.History.get_result()

export_result = <shapeflow.core.Endpoint object>

Export the result for a given analyzer ID and result ID

shapeflow.db.History.export_result()

clean = <shapeflow.core.Endpoint object>

Clean the database

shapeflow.db.History.clean()

forget = <shapeflow.core.Endpoint object>

Remove all entries from the database

shapeflow.db.History.forget()

class shapeflow.api._FilesystemDispatcher(instance: Optional[object] = None)

Bases: shapeflow.core.Dispatcher

Dispatches /api/fs/<endpoint>

select_video = <shapeflow.core.Endpoint object>

Open a dialog to select a video file

shapeflow.main._Filesystem.select_video()

select_design = <shapeflow.core.Endpoint object>

Open a dialog to select a design file

shapeflow.main._Filesystem.select_design()

check_video = <shapeflow.core.Endpoint object>

Check whether the given video file is valid

shapeflow.main._Filesystem.check_video()

check_design = <shapeflow.core.Endpoint object>

Check whether the given video file is valid

shapeflow.main._Filesystem.check_design()

open_root = <shapeflow.core.Endpoint object>

Open the application’s root directory in a file explorer window

shapeflow.main._Filesystem.open_root()

class shapeflow.api._CacheDispatcher(instance: Optional[object] = None)

Bases: shapeflow.core.Dispatcher

Dispatches /api/cache/<endpoint>

clear = <shapeflow.core.Endpoint object>

Clear the cache

shapeflow.main._cache.clear()

size = <shapeflow.core.Endpoint object>

Get the size of the cache on disk

shapeflow.main._Cache.size()

class shapeflow.api.ApiDispatcher(instance: Optional[object] = None)

Bases: shapeflow.core.Dispatcher

Invoked by Flask server for requests to /api/

ping = <shapeflow.core.Endpoint object>

Ping the backend

shapeflow.main._Main.ping()

map = <shapeflow.core.Endpoint object>

Get API map

shapeflow.main._Main.map()

schemas = <shapeflow.core.Endpoint object>

Get all schemas

shapeflow.main._Main.schemas()

normalize_config = <shapeflow.core.Endpoint object>

Normalize the given config to the current version of the application

shapeflow.main._Main.normalize_config()

get_settings = <shapeflow.core.Endpoint object>

Get the application settings

shapeflow.main._Main.get_settings()

set_settings = <shapeflow.core.Endpoint object>

Set the application settings

shapeflow.main._Main.set_settings()

events = <shapeflow.core.Endpoint object>

Open an event stream

shapeflow.main._Main.events()

stop_events = <shapeflow.core.Endpoint object>

Stop the event stream

shapeflow.main._Main.stop_events()

log = <shapeflow.core.Endpoint object>

Open a log stream

shapeflow.main._Main.log()

stop_log = <shapeflow.core.Endpoint object>

Stop the log stream

shapeflow.main._Main.stop_log()

unload = <shapeflow.core.Endpoint object>

Unload the application. In order to support page reloading, the backend will wait for some time and quit if no further requests come in.

shapeflow.main._Main.unload()

quit = <shapeflow.core.Endpoint object>

Quit without waiting for incoming requests

shapeflow.main._Main.quit()

restart = <shapeflow.core.Endpoint object>

Restart the server

shapeflow.main._Main.restart()

pid_hash = <shapeflow.core.Endpoint object>

Get the hashed process ID. Used to confirm server restart without exposing the actual PID.

shapeflow.main._Main.pid_hash()

fs = <shapeflow.api._FilesystemDispatcher object>

Nested _FilesystemDispatcher

db = <shapeflow.api._DatabaseDispatcher object>

Nested _DatabaseDispatcher

va = <shapeflow.api._VideoAnalyzerManagerDispatcher object>

Nested _VideoAnalyzerManagerDispatcher

cache = <shapeflow.api._CacheDispatcher object>

Nested _CacheDispatcher

shapeflow.api.api = <shapeflow.api.ApiDispatcher object>

Global ApiDispatcher instance. Endpoints should be exposed against this object. API calls should be dispatched from this object.

URLs match object structure: "/api/va/abc123/analyze" is equivalent to api.va.abc123.analyze.

_VideoAnalyzerManagerDispatcher.__id__ = <shapeflow.api._VideoAnalyzerDispatcher object>

Prototype _VideoAnalyzerDispatcher instance.

Endpoint instances exposed against this object will be propagated to all analyzers and bound to their respective instances.

config

class shapeflow.config.VideoFileHandlerConfig

Bases: shapeflow.core.config.BaseConfig

class shapeflow.config.FlipConfig

Bases: shapeflow.core.config.BaseConfig

class shapeflow.config.TransformHandlerConfig

Bases: shapeflow.core.interface.HandlerConfig

Transform handler configuration

roi: Optional[shapeflow.maths.coordinates.Roi]

The region of interest (ROI).

This is the position of the chip in the video in relative coordinates with respect to its resolution.

flip: shapeflow.config.FlipConfig

Flip the selected ROI horizontally/vertically before estimating the transform

turn: int

Turn the ROI a number of clockwise 90° turns before estimating the transform

class shapeflow.config.FilterHandlerConfig

Bases: shapeflow.core.interface.HandlerConfig

Filter handler configuration

class shapeflow.config.MaskConfig

Bases: shapeflow.core.config.BaseConfig

Mask configuration

name: str

The name of this mask.

Taken from the name of its layer in the design file.

skip: bool

Whether to skip this mask when analyzing

filter: shapeflow.config.FilterHandlerConfig

Filter configuration for this mask.

parameters: Tuple[Any, ...]

Feature parameter overrides for this mask.

If None, the global feature parameters are used when computing the feature value.

property ready

Whether this mask is ready to be analyzed

class shapeflow.config.DesignFileHandlerConfig

Bases: shapeflow.core.config.BaseConfig

Design file handler configuration

dpi: int

The details per inch (DPI) at which to render the design

overlay_alpha: float

Transparency of the overlay

smoothing: int

The smoothing kernel size used when generating masks from renders

class shapeflow.config.VideoAnalyzerConfig

Bases: shapeflow.core.backend.BaseAnalyzerConfig

Video analyzer configuration

frame_interval_setting: shapeflow.config.FrameIntervalSetting

Determines which parameter to use when requesting a set of frames to analyze.

• Nf: request shapeflow.video.VideoAnalyzerConfig.Nf frames in total

• dt: request a frame every seconds shapeflow.video.VideoAnalyzerConfig.dt seconds

dt: Optional[float]

Frame interval in seconds

Nf: Optional[int]

Total number of frames

video: shapeflow.config.VideoFileHandlerConfig

Video file handler configuration

design: shapeflow.config.DesignFileHandlerConfig

Design file handler configuration

transform: shapeflow.config.TransformHandlerConfig

Transform handler configuration

features: Tuple[shapeflow.core.backend.FeatureType, ...]

The features to extract

feature_parameters: Tuple[shapeflow.core.backend.FeatureConfig, ...]

The feature parameters to use

masks: Tuple[shapeflow.config.MaskConfig, ...]

Mask configuration

shapeflow.config.schemas() → Dict[str, dict]

Get the JSON schemas of

• shapeflow.video.VideoAnalyzerConfig

• shapeflow.Settings

• shapeflow.core.backend.AnalyzerState

• shapeflow.core.backend.QueueState

shapeflow.config.loads(config: str) → shapeflow.core.config.BaseConfig

Load a configuration object from a JSON string.

Parameters

config (str) – JSON string configuration

Returns

A configuration object

Return type

BaseConfig

shapeflow.config.normalize_config(d: dict) → dict

Normalize a configuration dictionary to match the current __version__

Parameters

d (dict) – A configuration dict

Returns

A normalized configuration dict

Return type

dict

db

class shapeflow.db.VideoFileModel(path)

Bases: shapeflow.core.db.FileModel

Database model of a video file.

resolve() → shapeflow.db.VideoFileModel

Resolve the file by its SHA1 hash ~ hash_file().

If the computed hash is new, the file is committed to the database. Otherwise, the original entry is re-used.

Returns

The current instance if the file is new, or a new FileModel instance representing the original database entry.

Return type

FileModel

class shapeflow.db.DesignFileModel(path)

Bases: shapeflow.core.db.FileModel

Database model of a design file.

resolve() → shapeflow.db.DesignFileModel

Resolve the file by its SHA1 hash ~ hash_file().

If the computed hash is new, the file is committed to the database. Otherwise, the original entry is re-used.

Returns

The current instance if the file is new, or a new FileModel instance representing the original database entry.

Return type

FileModel

class shapeflow.db.ConfigModel(**kwargs)

Bases: shapeflow.core.db.DbModel

Database model of a configuration.

class shapeflow.db.ResultModel(**kwargs)

Bases: shapeflow.core.db.DbModel

Database model of a result.

feature

The feature that was analyzed

data

Results of the analysis. In JSON, formatted ~ pandas.DataFrame.to_json(orient='split')

class shapeflow.db.AnalysisModel(*args, **kwargs)

Bases: shapeflow.core.db.BaseAnalysisModel

Database model of an analysis.

Contains a reference to a BaseAnalyzer instance.

get_name() → str

Returns

Name of the analysis from the database. Empty names are reset to ‘#{id}’

Return type

str

get_runs() → int

Returns

Number of completed runs for the analysis

Return type

int

get_id() → int

Returns

The database id of the analysis

Return type

int

store()

Store analysis information from the BaseAnalyzer to the database.

export_result(run: Optional[int] = None, manual: bool = False)

Export a result to disk

Parameters

• run (int) – The run to export

• manual (bool) – Whether this export request is manual (i.e. explicitly requested by the user). This setting determines whether to follow settings.app.save_result_manual or settings.app.save_result_auto when choosing where or whether to actually save.

load_config(video_path: Optional[str] = None, design_path: Optional[str] = None, include: Optional[List[str]] = None) → Optional[dict]

Load configuration from the database.

Parameters

• video_path (str) – Path to video file

• design_path (str) – Path to design file

• include (List[str]) – List of fields which must be included in the configuration. If a matching ConfigModel doesn’t provide all of these, the other matches will be parsed to complete it.

Returns

Configuration dict, if a matching config is found. Otherwise, returns None

Return type

dict

get_config_json() → Optional[str]

Get the current configuration in JSON

get_undo_config(context: Optional[str] = None) → Tuple[Optional[dict], Optional[int]]

Undo configuration. If a context is supplied, ensure that the context field changes, but the other fields remain the same

Parameters

context (str) – Name of a VideoAnalyzerConfig field

Raises

ValueError – If context is not a VideoAnalyzer field

get_redo_config(context: Optional[str] = None) → Tuple[Optional[dict], Optional[int]]

Redo configuration. If a context is supplied, ensure that the context field changes, but the other fields remain the same

Parameters

context (str) – Name of a VideoAnalyzerConfig field

Raises

ValueError – If context is not a VideoAnalyzer field

class shapeflow.db.History(path: Optional[pathlib.Path] = None)

Bases: shapeflow.core.db.SessionWrapper, shapeflow.core.RootInstance

Interface to the history database

add_video_file(path: str) → shapeflow.db.VideoFileModel

Add a video file to the database. Duplicate files are resolved to their original entry.

Parameters

path (str) – The path of the file to add

Returns

A database model of the file. Will reference the original entry if the user tried to add a previously used file again.

Return type

VideoFileModel

add_design_file(path: str) → shapeflow.db.DesignFileModel

Add a design file to the database. Duplicate files are resolved to their original entry.

Parameters

path (str) – The path of the file to add

Returns

A database model of the file. Will reference the original entry if the user tried to add a previously used file again.

Return type

DesignFileModel

add_analysis(analyzer: shapeflow.core.backend.BaseAnalyzer, model: Optional[shapeflow.db.AnalysisModel] = None) → shapeflow.db.AnalysisModel

Add a new analysis to the database.

Parameters

• analyzer (BaseAnalyzer) – The analyzer object to add to the database

• model (AnalysisModel) – Optionally, an existing model can be specified. Defaults to None

Returns

If a model is provided, no new AnalysisModel will be created and the analyzer will be linked to the existing model instead.

Return type

AnalysisModel

fetch_analysis(id: int) → Optional[shapeflow.db.AnalysisModel]

Fetch an analysis model from the database.

Parameters

id (int) – Database id of the analysis to fetch

get_paths() → Dict[str, List[str]]

Fetch the latest video and design file paths from the database.

shapeflow.api._DatabaseDispatcher.get_recent_paths

Number of paths is limited by settings.app.recent_files

get_result_list(analysis: int) → dict

Fetch the result list for a given analysis

shapeflow.api._DatabaseDispatcher.get_result_list

Parameters

analysis (int) – Database id of the analysis

Returns

A dict mapping run id int to result id int

Return type

dict

get_result(analysis: int, run: int) → dict

Fetch the result for a given analysis and run

shapeflow.api._DatabaseDispatcher.get_result

Parameters

• analysis (int) – Database id of the analysis

• run (int) – Run number of the result to fetch

Returns

Analysis results, as a dict formatted ~ pandas.DataFrame.to_json(orient='split')

Return type

dict

export_result(analysis: int, run: Optional[int] = None) → bool

Export the result for a given analysis and run

shapeflow.api._DatabaseDispatcher.export_result

Parameters

• analysis (int) – Database id of the analysis

• run (int) – Run number of the result to fetch

Returns

True if exported, False if something went wrong.

Return type

bool

check() → bool

Check the database’s integrity (somewhat). Makes sure that the required tables exist and that their columns match.

Returns

True if everything’s fine, False if the database is messed up

Return type

bool

clean() → None

Clean the database.

shapeflow.api._DatabaseDispatcher.clean

• remove ‘video_file & ‘design_file’ entries with <null> path

  • resolve entries with <null> hash

• remove ‘analysis’ entries with <null> config

• remove ‘config’ entries with <null> json

• for ‘analysis’ entries older than settings.db.cleanup_interval

  • remove all non-primary ‘config’ entries

  • remove all non-primary ‘results’ entries

forget() → None

Remove everything.

shapeflow.api._DatabaseDispatcher.forget

video

exception shapeflow.video.VideoFileTypeError(*args)

Bases: shapeflow.core.backend.BackendSetupError

class shapeflow.video.VideoFileHandler(video_path, config: Optional[shapeflow.config.VideoFileHandlerConfig] = None)

Bases: shapeflow.core.backend.CachingInstance, shapeflow.core.Lockable

Interface to video files ~ OpenCV

path: str

Path to the video file

set_requested_frames(requested_frames: List[int]) → None

Add a list of requested frames

Limiting the set of possible frames decreases the disk space cost needed to get a performance advantage through caching.

read_frame(frame_number: Optional[int] = None) → numpy.ndarray

Read a frame from cv2.VideoCapture.

seek(position: Optional[float] = None) → float

Seek to a relative position in the video ~ [0,1]

get_seek_position() → float

Get current relative position in the video ~ [0,1]

exception shapeflow.video.NotEstimatedYet(*args)

Bases: shapeflow.core.RootException

class shapeflow.video.TransformHandler(video_shape, design_shape, config: shapeflow.config.TransformHandlerConfig)

Bases: shapeflow.core.config.Instance, shapeflow.core.interface.Handler

Handles coordinate transforms

set_implementation(implementation: Optional[str] = None) → str

Set the implementation.

set(matrix: Optional[numpy.ndarray])

Set the transform matrix

property is_set: bool

Whether the transform matrix is set

get_relative_roi() → dict

Get the relative ROI as a dict

estimate(roi: Optional[shapeflow.maths.coordinates.Roi] = None) → None

Estimate the transform matrix from a set of coordinates.

Coordinates should correspond to the corners of the outline of the design, relative to the video frame size: * x in [0,1] ~ width * y in [0,1] ~ height

clear() → None

Clear the ROI

coordinate(coordinate: shapeflow.maths.coordinates.ShapeCoo) → Optional[shapeflow.maths.coordinates.ShapeCoo]

Transform a design coordinate to a video coordinate

inverse(img: numpy.ndarray) → numpy.ndarray

Inverse transform an image

class shapeflow.video.FilterHandler(config: Optional[shapeflow.config.FilterHandlerConfig] = None)

Bases: shapeflow.core.config.Instance, shapeflow.core.interface.Handler

Handles filters

set_config(config: dict) → None

Set the configuration of this filter

set(color: Optional[shapeflow.maths.colors.HsvColor] = None) → shapeflow.core.interface.FilterConfig

Set the filter to a color.

set_implementation(implementation: Optional[str] = None) → str

Set the filter implementation

class shapeflow.video.Mask(design: shapeflow.video.DesignFileHandler, mask: numpy.ndarray, name: str, config: Optional[shapeflow.config.MaskConfig] = None, filter: Optional[shapeflow.video.FilterHandler] = None)

Bases: shapeflow.core.config.Instance

Handles masks in the context of a video file

filter: shapeflow.video.FilterHandler

The filter associated with this mask

set_config(config: dict) → None

Set the configuration of this mask

set_filter(color: Optional[shapeflow.maths.colors.HsvColor])

Set the filter of this mask

contains(coordinate: shapeflow.maths.coordinates.ShapeCoo) → bool

Whether a coordinate is contained within this mask

clear_filter()

Clear this mask’s filter

property rows

This mask’s row slice

To crop an image to this mask:

crop = img[mask.rows, mask.cols]

property cols

This mask’s column slice

To crop an image to this mask:

crop = img[mask.rows, mask.cols]

class shapeflow.video.DesignFileHandler(path: str, config: Optional[shapeflow.config.DesignFileHandlerConfig] = None, mask_config: Optional[Tuple[shapeflow.config.MaskConfig, ...]] = None)

Bases: shapeflow.core.backend.CachingInstance

Handles design files

peel_design(design_path: str, dpi: int) → numpy.ndarray

Render out all of the layers in a design

read_masks(design_path: str, dpi: int) → Tuple[List[numpy.ndarray], List[str]]

Load masks from the rendered files

property shape

The numpy.ndarray.shape of the design

overlay() → numpy.ndarray

The overlay of this design

overlay_frame(frame: numpy.ndarray) → numpy.ndarray

Apply this design’s overlay to a frame

Parameters

frame (np.ndarray) – An image

Returns

The image with the overlay laid over it

Return type

np.ndarray

property masks: List[shapeflow.video.Mask]

The masks of this design

class shapeflow.video.MaskFunction(mask: shapeflow.video.Mask, global_config: shapeflow.core.backend.FeatureConfig, config: Optional[dict] = None)

Bases: shapeflow.core.backend.Feature

An abstract feature based on a Mask

property ready: bool

Whether this feature is ready to be calculated

property skip

Whether this feature should be skipped

px2mm(value)

Convert design-space pixels to mm

Parameters

value (float) – Distance in # of pixels

Returns

Distance in mm

Return type

float

pxsq2mmsq(value)

Convert design-space pixels to mm²

Parameters

value (float) – Area in pixels

Returns

Area in mm²

Return type

float

value(frame) → Any

The value of this feature for a given frame

Parameters

frame – An image

Returns

Some value

Return type

Any

state(frame: numpy.ndarray, state: numpy.ndarray) → numpy.ndarray

Generate a state image (BGR)

class shapeflow.video.VideoAnalyzer(config: Optional[shapeflow.config.VideoAnalyzerConfig] = None)

Bases: shapeflow.core.backend.BaseAnalyzer

Main video handling class

• Load frames from video files

• Load mask files

• Load/save measurement metadata

video: shapeflow.video.VideoFileHandler

Handles video operations for this analyzer

• Reads frames from the video file

design: shapeflow.video.DesignFileHandler

Handles design operations for this analyzer.

• Renders the design

• Reads in the overlay and mask images

• Initializes the Mask instances

transform: shapeflow.video.TransformHandler

Handles the transform for this analyzer

features: Tuple[shapeflow.core.backend.Feature, ...]

The features used in this analysis

results: Dict[str, pandas.core.frame.DataFrame]

The results of the current run of this analysis

property cached: bool

property has_results: bool

can_launch() → bool

Returns

Whether this analyzer can launch

Return type

bool

can_filter() → bool

Returns

Whether this analyzer can filter

Return type

bool

can_analyze() → bool

Returns

Whether this analyzer can analyze

Return type

bool

property position

set_config(config: dict, silent: bool = False) → dict

Set the analyzer’s configuration

shapeflow.api._VideoAnalyzerDispatcher.set_config

Parameters

• config (dict) – A configuration dict

• silent (bool) – If True, don’t push server-sent events. False by default.

Returns

The updated configuration dict. Some fields may have been changed due to incompatibility.

Return type

dict

get_transformed_frame(frame_number: Optional[int] = None) → numpy.ndarray

Get a video frame transformed to design-space

shapeflow.api._VideoAnalyzerDispatcher.get_transformed_frame

Can be streamed to the user interface.

Parameters

frame_number (Optional[int]) – The frame number to get. If None, get the current frame number.

Returns

An image (design-space)

Return type

np.ndarray

get_inverse_transformed_overlay() → numpy.ndarray

Get the design overlay image transformed to video-space

shapeflow.api._VideoAnalyzerDispatcher.get_inverse_transformed_overlay

Can be streamed tothe user interface.

Returns

An image (in video-space)

Return type

np.ndarray

get_colors() → Tuple[str, ...]

Get the list of colors to use for each mask

shapeflow.api._VideoAnalyzerDispatcher.get_colors

Returns

A tuple of CSS-compatible hex RGB strings

Return type

Tuple[str, …]

frame_numbers() → Generator[int, None, None]

Get the requested frame numbers

Returns

An iterator that returns the requested frame numbers

Return type

Generator[int, None, None]

get_inverse_overlaid_frame(frame_number: Optional[int] = None) → numpy.ndarray

Get a raw video frame overlaid with the inverse transformed design overlay. Used to evaluate video-design alignment.

shapeflow.api._VideoAnalyzerDispatcher.get_inverse_overlaid_frame

Can be streamed tothe user interface.

Parameters

frame_number (Optional[int]) – The frame number to get. If None, get the current frame number.

Returns

An image (video-space)

Return type

np.ndarray

seek(position: Optional[float] = None) → float

Seek the video to a relative position

shapeflow.api._VideoAnalyzerDispatcher.seek

Parameters

position (float) – A relative position in the video, with 0 the start and 1 the end.

Returns

The resulting relative position. This may differ from the original requested position due to being resolved to the nearest requested frame number. This improves performance if cached frames can be reused.

Return type

float

estimate_transform(roi: Optional[dict] = None) → Optional[dict]

Estimate the video-design transform from a ROI

shapeflow.api._VideoAnalyzerDispatcher.estimate_transform

Parameters

roi (Optional[dict]) – A region of interest as a dict. Must be compatible with shapeflow.maths.coordinates.Roi. If None, the current shapeflow.config.TransformHandlerConfig.roi will be used.

Returns

The region of interest that was provided, or the current region of interest if None.

Return type

Optional[dict]

clear_roi() → None

Clear the current region of interest and video-design transform

shapeflow.api._VideoAnalyzerDispatcher.clear_roi

turn_cw() → None

Add a clockwise 90° turn to this analyzer’s shapeflow.config.TransformHandlerConfig.turn

shapeflow.api._VideoAnalyzerDispatcher.turn_cw

turn_ccw() → None

Add a counter-clockwise 90° turn to this analyzer’s shapeflow.config.TransformHandlerConfig.turn

shapeflow.api._VideoAnalyzerDispatcher.turn_ccw

flip_h() → None

Toggle this analyzer’s shapeflow.config.TransformHandlerConfig.flip’s horizontal

shapeflow.api._VideoAnalyzerDispatcher.flip_h

flip_v() → None

Toggle this analyzer’s shapeflow.config.TransformHandlerConfig.flip’s vertical

shapeflow.api._VideoAnalyzerDispatcher.flip_v

undo_config(context: Optional[str] = None) → dict

Undo this analyzer’s last configuration change

shapeflow.api._VideoAnalyzerDispatcher.undo_config

Parameters

context (str) – The context in which to undo. If context is None, the previous configuration will be applied. If context is e.g. "transform", the latest configuration with a transform different from the current one will be applied.

Returns

A configuration dict

Return type

dict

redo_config(context: Optional[str] = None) → dict

Redo this analyzer’s last undone configuration change

shapeflow.api._VideoAnalyzerDispatcher.redo_config

Parameters

context (str) – The context in which to redo. If context is None, the previous configuration will be applied. If context is e.g. "transform", the latest configuration with a transform different from the current one will be applied.

Returns

A configuration dict

Return type

dict

set_filter_click(relative_x: float, relative_y: float) → None

Configure a filter by clicking an “in”-pixel on the image

shapeflow.api._VideoAnalyzerDispatcher.set_filter_click

Parameters

• relative_x (float) – The ‘x’-position of an “in”-pixel in relative video-space

• relative_y (float) – The ‘y’-position of an “in”-pixel in relative video-space

clear_filters() → bool

Clear this analyzer’s filter configuration

shapeflow.api._VideoAnalyzerDispatcher.clear_filters

get_state_frame(frame_number: Optional[int] = None, featureset: Optional[int] = None) → numpy.ndarray

Get a state frame. Used to evaluate filter configuration.

shapeflow.api._VideoAnalyzerDispatcher.get_state_frame

Can be streamed tothe user interface.

Parameters

• frame_number (Optional[int]) – The frame number to get. If None, get the current frame number.

• featureset (Optional[int]) – The index of the FeatureSet for which to get the state frame. If None, use 0. Feature sets are taken from this analyzer’s features.

Returns

An image (design-space)

Return type

np.ndarray

get_overlay_png() → bytes

Get this analyzer’s design overlay

shapeflow.api._VideoAnalyzerDispatcher.get_overlay_png

Returns

The design overlay encoded as PNG

Return type

bytes

calculate(frame_number: int) → None

Calculate this analyzer’s features for a frame.

Parameters

frame_number (int) – The frame to calculate for

Returns

Results are stored in results

Return type

None

analyze() → bool

Run the configured analysis

shapeflow.api._VideoAnalyzerDispatcher.analyze

Returns

Whether the analysis was successful

Return type

bool

load_config() → None

Load a configuration from the database

get_time(frame_number: Optional[int] = None) → float

Get the time corresponding to a video frame number

shapeflow.api._VideoAnalyzerDispatcher.get_time

Parameters

frame_number (Optional[int]) – A frame number. If None, use the current frame number.

Returns

A time in seconds

Return type

float

get_fps() → float

Get the framerate of the video

shapeflow.api._VideoAnalyzerDispatcher.get_fps

get_total_time() → float

Get the total time of the video in seconds

shapeflow.api._VideoAnalyzerDispatcher.get_total_time

read_frame(frame_number: Optional[int] = None) → numpy.ndarray

Get a raw video frame.

shapeflow.api._VideoAnalyzerDispatcher.get_raw_frame

Can be streamed tothe user interface.

Parameters

frame_number (Optional[int]) – The frame number to get. If None, get the current frame number.

Returns

An image (video-space)

Return type

np.ndarray

get_seek_position() → float

Get the current relative seek position in the video

shapeflow.api._VideoAnalyzerDispatcher.get_seek_posititon

get_relative_roi() → dict

Get the current region of interest in relative video-space

shapeflow.api._VideoAnalyzerDispatcher.get_relative_roi

get_coordinates() → Optional[list]

Get the current coordinates

shapeflow.api._VideoAnalyzerDispatcher.get_coordinates

server

class shapeflow.server.ServerThread(*args, **kwargs)

Bases: threading.Thread

A thread running a Flask app over a waitress server.

run()

Serve until interrupted or stopped.

If the current address is already in use, the server errors out & stops the current process with shapeflow.server.ServerThread.stop().

stop()

Stop the thread & process with exit(0)

class shapeflow.server.ShapeflowServer

Bases: shapeflow.main.ShapeflowServerInterface

Wrapper for a Flask server

serve(host: str, port: int, open: bool) → None

Serve the application.

Starts a new ServerThread and opens a new browser window/tab if requested.

This method keeps serving until either

• quit() is called

• unload() is called and no incoming traffic is received for 5 seconds.

• The user interrupts the process with Ctrl+C

Parameters

• host (str) – Host address

• port (int) – Host port

• open (bool) – Whether to open in a browser window/tab after starting the server

get_file(file: str)

Serve frontend files

Parameters

file (str) – The file to send

call_api(address: str) → flask.wrappers.Response

Dispatch request to the API.

Arguments are gathered from Flask’s request.data or request.args

Handles multiple types of return data:

• bytes data are handled by flask.make_response

• for shapeflow.core.streaming.BaseStreamer instances, custom flask.Response objects are made from their shapeflow.core.streaming.BaseStreamer.stream() generator

• all other data are handled by flask.jsonify

Parameters

address (str) – The address of the endpoint to dispatch to

unload() → None

Mark for unload.

quit() → None

Mark for quit.

restart() → None

Restart the server.

active() → None

If the _unload has been set, cancel it. Should be called for incoming traffic.

property api: shapeflow.api.ApiDispatcher

Get a reference to shapeflow.api.api and ensure it has been initialized properly with shapeflow.main and bound to this ShapeflowServer instance.

This property is used to lazy-load shapeflow.api.api on the first request. As this takes some time, requests received during loading will be locked out until initialization completes. Subsequent requests bypass this lock.

Returns

A reference to api

Return type

ApiDispatcher

property eventstreamer: shapeflow.core.streaming.EventStreamer

A reference to the server’s shapeflow.core.streaming.EventStreamer instance

main

Main functionality of the shapeflow server.

Implements shapeflow.api endpoints related to the global application and analyzer management.

The classes defined below should not be instantiated individually, but as a whole using shapeflow.main.load(), because their setup process is not very intuitive.

class shapeflow.main._Main(server: shapeflow.main.ShapeflowServerInterface)

Bases: object

Implements root-level api endpoints.

ping() → bool

Ping the server

shapeflow.api.ApiDispatcher.ping

Returns

If the server is up, True. If the server is down, a /api/ping request will not get any response.

Return type

bool

map() → Dict[str, List[str]]

Get the URL map of the API

shapeflow.api.ApiDispatcher.map

Returns

A flat dict mapping each available URL to a list of accepted HTTP methods

Return type

Dict[str, List[str]]

schemas() → dict

Get the application schemas

shapeflow.api.ApiDispatcher.schemas

Returns

A dict with schemas

Return type

dict

normalize_config(config: dict) → dict

Normalize a configuration dict with shapeflow.config.normalize_config()

shapeflow.api.ApiDispatcher.normalize_config

Parameters

config (dict) – A configuration dict

Returns

A normalized configuration dict

Return type

dict

get_settings() → dict

Get the application settings

shapeflow.api.ApiDispatcher.get_settings

Returns

The application settings as a dict

Return type

dict

set_settings(settings: dict) → dict

Set the application settings

shapeflow.api.ApiDispatcher.set_settings

Parameters

settings (dict) – New application settings as a dict

Returns

The new application settings dict, which may have been modified.

Return type

dict

events() → shapeflow.core.streaming.EventStreamer

Get a server-sent event stream

shapeflow.api.ApiDispatcher.events

Returns

A BaseStreamer object, to be streamed by Flask

Return type

EventStreamer

stop_events() → None

Stop streaming server-sent events

shapeflow.api.ApiDispatcher.stop_events

log() → shapeflow.core.streaming.PlainFileStreamer

Start streaming log file

shapeflow.api.ApiDispatcher.log

Returns

A BaseStreamer object, to be streamed by Flask

Return type

PlainFileStreamer

stop_log() → None

Stop streaming log file

shapeflow.api.ApiDispatcher.stop_log

unload() → bool

Unload the application. Called when the user closes or refreshes a tab with the user interface.

shapeflow.api.ApiDispatcher.unload

quit() → bool

Quit the API server.

shapeflow.api.ApiDispatcher.quit

restart() → bool

Restart the API server

shapeflow.api.ApiDispatcher.restart

pid_hash() → str

Get the current pid hash of the API server.

shapeflow.api.ApiDispatcher.pid_hash

Returns

The hash of the current pid. The actual pid is not given to avoid cheekiness.

Return type

str

class shapeflow.main._Cache

Bases: object

Implements cache endpoints in api.

clear_cache() → None

Clear the cache

shapeflow.api._CacheDispatcher.clear

cache_size() → str

Get the size of the cache

shapeflow.api._CacheDispatcher.size

Returns

The size of the cache in human-readable form

Return type

str

class shapeflow.main._Filesystem

Bases: object

Implements fs endpoints in api.

select_video() → Optional[str]

Open a video selection dialog

shapeflow.api._FilesystemDispatcher.select_video

Returns

The path of the selected video file, if any.

Return type

Optional[str]

select_design() → Optional[str]

Open a design selection dialog

shapeflow.api._FilesystemDispatcher.select_design

Returns

The path of the selected design file, if any.

Return type

Optional[str]

check_video(path: str) → bool

Check if a video file path is valid:

• Whether it exists

• Whether it’s a valid video file that can be opened with OpenCV

shapeflow.api._FilesystemDispatcher.check_video

Parameters

path (str) – A path

Returns

Whether the path is a valid video file

Return type

bool

check_design(path: str) → bool

Check if a design file path is valid:

• Whether it exists

• Whether it’s a valid design file

shapeflow.api._FilesystemDispatcher.check_design

Parameters

path (str) – A path

Returns

Whether the path is a valid design file

Return type

bool

open_root() → None

Open shapeflow.ROOTDIR in the file explorer

shapeflow.api._FilesystemDispatcher.open_root

class shapeflow.main._VideoAnalyzerManager(server: shapeflow.main.ShapeflowServerInterface)

Bases: object

Implements va endpoints in api.

Manages BaseAnalyzer instances

• Adds / removes instances

• Handles saving / loading of application state

• Handles analysis queueing

• Handles analyzer-specific streams

ID_LENGTH = 6

Length of id strings. Kept relatively short for readable URLs.

notice(message: str, persist: bool = False)

Push a notice to the server’s EventStreamer

Parameters

message (str) – The message to push

init() → str

Initialize a new analyzer.

shapeflow.api._VideoAnalyzerManagerDispatcher.init

A short unique ìd will be generated. For example, a new analyzer with rG7bgH as its id can be addressed via /api/va/rG7bgH.

Returns

The id string of the newly added analyzer

Return type

str

close(id: str) → bool

Close an analyzer.

shapeflow.api._VideoAnalyzerManagerDispatcher.close

Parameters

id (str) – The id string of the analyzer to remove

Returns

Whether the analyzer was removed successfully

Return type

bool

q_start(queue: List[str]) → bool

Start analyzing a queue.

shapeflow.api._VideoAnalyzerManagerDispatcher.start

Parameters

queue (List[str]) – List of analyzer id to queue.

q_stop() → None

Stop analyzing the current queue.

shapeflow.api._VideoAnalyzerManagerDispatcher.stop

q_cancel() → None

Cancel analyzing the current queue.

shapeflow.api._VideoAnalyzerManagerDispatcher.cancel

state() → dict

Get the queue state and the status of all analyzers.

shapeflow.api._VideoAnalyzerManagerDispatcher.state

Example:

{
    "q_state": 0,                   # QueueState
    "ids": ["abc123", "def456"],
    "status": {
        "abc123": {
            "state": 7,             # AnalyzerState
            "busy": True,
            "cached": True,
            "results": False,
            "position": 0.7,
            "progress": 0.75,
        },
        "def456": {
            "state": 6,             # AnalyzerState
            "busy": False,
            "cached": True,
            "results": False,
            "position": 0.0,
            "progress": 0.0,
        },
    }
}

With the state int values according to the Enum classes QueueState and AnalyzerState.

Returns

A state dict

Return type

dict

save_state() → None

Save application state to shapeflow.settings.app.state_path

shapeflow.api._VideoAnalyzerManagerDispatcher.save_state

load_state() → None

Load application state from shapeflow.settings.app.state_path

shapeflow.api._VideoAnalyzerManagerDispatcher.load_state

stream(id: str, endpoint: str) → shapeflow.core.streaming.BaseStreamer

Stream something.

shapeflow.api._VideoAnalyzerManagerDispatcher.stream

Parameters

• id (str) – The id of an analyzer

• endpoint (str) – The endpoint to stream

Returns

A stream handler object

Return type

BaseStreamer

stream_stop(id: str, endpoint: str) → None

Stop streaming something.

shapeflow.api._VideoAnalyzerManagerDispatcher.stream_stop

Parameters

• id (str) – The id of an analyzer

• endpoint (str) – The endpoint to stop streaming

shapeflow.main.load(server: shapeflow.main.ShapeflowServerInterface) → shapeflow.api.ApiDispatcher

Initialize api and return a reference to it.

Parameters

server (ShapeflowServer) – The shapeflow server object

Returns

A reference to api

Return type

ApiDispatcher

cli

Tiny commands to be called from sf.py

Calling from the commandline:

python sf.py --do <command name> <arguments>

exception shapeflow.cli.CliError

Bases: Exception

class shapeflow.cli.IterCommand(name, bases, namespace, **kwargs)

Bases: abc.ABCMeta

Command iterator metaclass.

Iterates over its subclasses, skipping any without a __command__. If any of these should remain abstract, they shouldn’t define one.

__command__: str

Command name. This is how the command is addressed from the commandline.

property sub

Returns True if this class is a subcommand

property dict: dict

Get a dict mapping all defined command names to their respective class

__getitem__(item: str) → shapeflow.cli.IterCommand

Get a subcommand by its __command__

abstract __usage__() → str

Usage string

abstract __help__() → str

Help string

class shapeflow.cli.Command(args: Optional[List[str]] = None)

Bases: abc.ABC

Abstract command.

• handles argument parsing & execution

• subclasses can implement their functionality in Command.command()

classmethod __help__() → str

Cleaned-up help string

classmethod __usage__() → str

Cleaned-up usage string

class shapeflow.cli.Sf(args: Optional[List[str]] = None)

Bases: shapeflow.cli.Command

Commandline entry point. This is the command that gets called first and calls any subcommands if requested.

class shapeflow.cli.Serve(args: Optional[List[str]] = None)

Bases: shapeflow.cli.Command

Starts the shapeflow server.

__command__ = 'serve'

HOST = '127.0.0.1'

PORT = 7951

class shapeflow.cli.Dump(args: Optional[List[str]] = None)

Bases: shapeflow.cli.Command

Dump application schemas and settings to JSON

__command__ = 'dump'

class shapeflow.cli.GitMixin

Bases: abc.ABC

Metaclass for commands interacting with the git repository

URL = 'https://github.com/ybnd/shapeflow/'

property repo: git.repo.base.Repo

property latest: str

is_up_to_date(tag) → bool

property tag: str

is_at_release(tag: str) → bool

property ui_url: str

class shapeflow.cli.Update(args: Optional[List[str]] = None)

Bases: shapeflow.cli.Command, shapeflow.cli.GitMixin

Update the application

__command__ = 'update'

class shapeflow.cli.Checkout(args: Optional[List[str]] = None)

Bases: shapeflow.cli.Command, shapeflow.cli.GitMixin

Check out a specific version of the application. Please not you will not have access to this command if you check out a version before 0.4.4

__command__ = 'checkout'

class shapeflow.cli.GetCompiledUi(args: Optional[List[str]] = None)

Bases: shapeflow.cli.Command, shapeflow.cli.GitMixin

Get the compiled UI for the current version

__command__ = 'get-compiled-ui'

class shapeflow.cli.SetupCairo(args: Optional[List[str]] = None)

Bases: shapeflow.cli.Command

Set up cairo DLLs (Windows) ~ https://github.com/preshing/cairo-windows/

__command__ = 'setup-cairo'

URL = 'https://github.com/preshing/cairo-windows/releases/download/with-tee/cairo-windows-1.17.2.zip'

property env: pathlib.Path

class shapeflow.cli.Declutter(args: Optional[List[str]] = None)

Bases: shapeflow.cli.Command

Hide clutter files from the repository

CLUTTER = ['mypy.ini', 'tox.ini', '__pycache__']

__command__ = 'declutter'