isimple.core package

Submodules

isimple.core.backend module

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

Bases: isimple.core.SetupError

msg = 'Error while setting up backend'

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

Bases: isimple.core.RootException

msg = 'Error in backend'

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

Bases: isimple.core.RootException

msg = 'Trying to access cache out of context'

class isimple.core.backend.CachingInstance(config: isimple.core.config.BaseConfig = None)

Bases: isimple.core.config.Instance

Interface to diskcache.Cache

is_caching() → bool

cancel_caching() → None

caching(override: bool = False)

class isimple.core.backend.Handler

Bases: abc.ABC

set_implementation(implementation: str) → str

get_implementation() → str

implementation_config() → isimple.core.config.BaseConfig

class isimple.core.backend.FeatureConfig

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

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

Bases: abc.ABC, isimple.core.config.Configurable

A feature implements interactions between BackendElements to produce a certain value

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

Calculate Feature for given frame and update state image (optional)

classmethod label() → str

classmethod unit() → str

skip

ready

set_color(color: isimple.maths.colors.Color)

color

Color of the Feature in figures.

A Feature’s color must be set as not to overlap with other Features in the same FeatureSet. Therefore, <Feature>._color must be determined by FeatureSet!

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

Return the Feature instance’s state image for a given frame

value(frame: numpy.ndarray) → Any

Compute the value of the Feature instance for a given frame

classmethod description() → str

config

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

Bases: object

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

colors

features

class isimple.core.backend.FeatureType(string: str = None)

Bases: isimple.core.config.Factory

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

config_schema() → dict

class isimple.core.backend.BaseAnalyzerConfig

Bases: isimple.core.config.BaseConfig

class isimple.core.backend.AnalyzerEvent

Bases: enum.Enum

An enumeration.

STATUS = 'status'

CONFIG = 'config'

RESULT = 'result'

RMETAD = 'result_metadata'

class isimple.core.backend.AnalyzerState

Bases: enum.IntEnum

An enumeration.

UNKNOWN = 0

INCOMPLETE = 1

CAN_LAUNCH = 2

LAUNCHED = 3

CAN_FILTER = 4

CAN_ANALYZE = 5

ANALYZING = 6

DONE = 7

CANCELED = 8

ERROR = 9

can_launch = <bound method AnalyzerState.can_launch of <enum 'AnalyzerState'>>

is_launched = <bound method AnalyzerState.is_launched of <enum 'AnalyzerState'>>

class isimple.core.backend.BaseVideoAnalyzer(config: isimple.core.backend.BaseAnalyzerConfig = None, eventstreamer: isimple.core.streaming.EventStreamer = None)

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

set_model(model: isimple.core.db.BaseAnalysisModel)

model

eventstreamer

set_eventstreamer(eventstreamer: isimple.core.streaming.EventStreamer = None)

event(category: isimple.core.backend.AnalyzerEvent, data: dict)

Push an event

Parameters:

• category – event category
• data – event data

Returns:

commit() → bool

Save video analysis configuration to history database

can_launch() → bool

can_filter() → bool

can_analyze() → bool

launched

set_state(state: int, push: bool = True)

state

done

state_transition(push: bool = True) → int

Handle state transitions

set_busy(busy: bool, push: bool = True)

busy

busy_context(busy_state: isimple.core.backend.AnalyzerState = None, done_state: isimple.core.backend.AnalyzerState = None)

cancel() → None

error() → None

set_progress(progress: float, push: bool = True)

progress

config

analyze() → bool

position

cached

has_results

status() → dict

push_status()

get_config(do_tag=False) → dict

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

launch() → bool

caching()

Caching context: propagated context to every object in _instances that implements caching

cache_open()

cache_close()

time(message: str = '', logger=<Logger isimple.core.backend (DEBUG)>)

timing

export()

description

class isimple.core.backend.AnalyzerType(string: str = None)

Bases: isimple.core.config.Factory

get() → Type[isimple.core.backend.BaseVideoAnalyzer]

config_schema() → dict

isimple.core.config module

class isimple.core.config.Factory(string: str = None)

Bases: isimple.core.EnforcedStr

get() → type

classmethod get_str(mapped_value)

options

descriptions

default

config_schema() → dict

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

Bases: object

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

class isimple.core.config.NpArray

Bases: numpy.ndarray

classmethod validate(v: Any) → str

class isimple.core.config.BaseConfig

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

Abstract class for configuration data.

• 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()

    ```

• Writing BaseConfig subclasses:

  • Use the @extends(ConfigType) decorator to make your configuration

    class accessible from the ConfigType Factory (defined below)

  • Configuration keys are declared as pydantic Field instances

    • Must be type-annotated for type resolution to work properly!

```

from pydantic import Field from isimple.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

arbitrary_types_allowed = False

use_enum_value = True

validate_assignment = True

json_encoders = {<class 'numpy.ndarray'>: <class 'list'>}

to_dict(do_tag: bool = False) → dict

Return the configuration as a serializable dict. :param do_tag: if True, add configuration class and version fields to the dict :return: dict

tag(d: dict) → dict

class isimple.core.config.ConfigType(string: str = None)

Bases: isimple.core.config.Factory

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

config_schema() → dict

class isimple.core.config.Configurable

Bases: isimple.core.Described

classmethod config_class()

classmethod config_schema()

class isimple.core.config.Instance(config: isimple.core.config.BaseConfig = None)

Bases: isimple.core.config.Configurable

config

isimple.core.db module

class isimple.core.db.SessionWrapper

Bases: object

Wrapper object for a SQLAlchemy session factory.

connect(session_wrapper: isimple.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 isimple.core.db.DbModel(**kwargs)

Bases: sqlalchemy.ext.declarative.api.Base, isimple.core.db.SessionWrapper

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 isimple.core.db.FileModel(path: str)

Bases: isimple.core.db.DbModel

Abstrat database model for files.

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

id = Column(None, Integer(), table=None, primary_key=True, nullable=False)

hash = Column(None, String(), table=None)

path = Column(None, String(), table=None)

used = Column(None, DateTime(), table=None)

resolved

Whether the FileModel has been resolved.

resolve() → isimple.core.db.FileModel

Resolve the file by its SHA1 hash. todo: reference to util.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 isimple.core.db.BaseAnalysisModel(**kwargs)

Bases: isimple.core.db.DbModel

AnalysisModel interface

get_name() → str

get_config_json() → Optional[str]

Get the current configuration in JSON

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

Load configuration from the database

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

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

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

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

store() → None

Store analysis information from wrapped BaseVideoAnalyzer to the database

isimple.core.interface module

class isimple.core.interface.InterfaceFactory(string: str = None)

Bases: isimple.core.config.Factory

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

config_schema() → dict

class isimple.core.interface.HandlerConfig

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

class isimple.core.interface.TransformConfig

Bases: isimple.core.config.BaseConfig

Undefined transform

class isimple.core.interface.TransformInterface

Bases: isimple.core.config.Configurable, abc.ABC

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

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

to_coordinates(to_shape: tuple) → numpy.ndarray

estimate(roi: isimple.maths.coordinates.Roi, from_shape: tuple, to_shape: tuple) → numpy.ndarray

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

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

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

class isimple.core.interface.FilterConfig

Bases: isimple.core.config.BaseConfig

Undefined filter

ready

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

class isimple.core.interface.FilterInterface

Bases: isimple.core.config.Configurable, abc.ABC

Handles pixel filtering operations

set_filter(filter, color: isimple.maths.colors.Color)

mean_color(filter) → isimple.maths.colors.Color

filter(filter, image: numpy.ndarray, mask: numpy.ndarray = None) → numpy.ndarray

class isimple.core.interface.FilterType(string: str = None)

Bases: isimple.core.interface.InterfaceFactory

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

class isimple.core.interface.TransformType(string: str = None)

Bases: isimple.core.interface.InterfaceFactory

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

isimple.core.streaming module

class isimple.core.streaming.BaseStreamer

Bases: abc.ABC

push(value: Any)

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

stop()

classmethod mime_type() → str

classmethod content_type()

class isimple.core.streaming.JsonStreamer

Bases: isimple.core.streaming.BaseStreamer

class isimple.core.streaming.EventStreamer

Bases: isimple.core.streaming.JsonStreamer

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

Push a JSON event

Parameters:

• category – event category
• id – UUID of event source
• data – event data

Returns:

class isimple.core.streaming.FrameStreamer

Bases: isimple.core.streaming.BaseStreamer

class isimple.core.streaming.JpegStreamer

Bases: isimple.core.streaming.FrameStreamer

class isimple.core.streaming.StreamHandler

Bases: isimple.core.Lockable

A singleton object to handle streaming frames from methods

register(instance: object, method, stream_type: Type[isimple.core.streaming.BaseStreamer] = None) → isimple.core.streaming.BaseStreamer

Register method, start a streamer. If method has been registered already, return its streamer.

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

push(instance: object, method, data)

If method is registered, push data to its streamer.

unregister(instance: object, method=None)

Unregister method: stop its streamer & delete

update()

stop()

isimple.core.streaming.stream(method)

Decorator for streaming methods. To stream frames, the wrapped method should be registered

in the global StreamHandler streams.

Module contents

exception isimple.core.RootException(*args)

Bases: Exception

msg = ''

exception isimple.core.SetupError(*args)

Bases: isimple.core.RootException

class isimple.core.EnforcedStr(string: str = None)

Bases: str

options

descriptions

describe

default

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

Bases: object

compatible(method: Callable) → bool

signature

streaming

registered

name

add(method)

register(name: str)

class isimple.core.EndpointRegistry

Bases: object

class isimple.core.InstanceRegistry

Bases: isimple.core.EndpointRegistry

This one is global, collects callables that expose endpoints

expose(endpoint: isimple.core.Endpoint)

exposes(endpoint: isimple.core.Endpoint)

endpoints

class isimple.core.ImmutableRegistry(endpoints: isimple.core.InstanceRegistry = None)

Bases: isimple.core.EndpointRegistry

expose(endpoint: isimple.core.Endpoint)

exposes(endpoint: isimple.core.Endpoint)

endpoints

class isimple.core.Described

Bases: object

class isimple.core.Lockable

Bases: abc.ABC

lock()

cancel()

error()

canceled

errored

clear_cancel()

clear_error()

class isimple.core.RootInstance

Bases: isimple.core.Lockable

get_id()

id

instance_mapping

get(endpoint: isimple.core.Endpoint, index: int = None) → Callable