actfw_core package

Subpackages

Submodules

actfw_core.application module

class actfw_core.application.AppSettings(settings: Dict[str, Any], schema: Dict[str, actfw_core.application.SettingSchema])

Bases: object

class actfw_core.application.Application(stop_by_signals: typing.Iterable[signal.Signals] = (<Signals.SIGINT: 2>, <Signals.SIGTERM: 15>))

Bases: object

get_app_settings(path: str = 'setting_schema.json') → actfw_core.application.AppSettings

Get application settings from schema. :param path: path to schema file :type path: str

Returns

application settings

Return type

AppSettings

get_settings(default_settings: Dict[str, Any]) → Dict[str, Any]

Get given Act settings.

Parameters

default_settings (dict) – default settings

Returns

updated settings

Return type

dict

Notes

Copy default_settings and overwrite it by given Act settings.

register_task(task: actfw_core.task.task.Task) → None

Register the application task.

Parameters

task (Task) – task

run() → None

Start application

running: bool
settings: Optional[Dict[str, Any]]

Actcast Application

stop() → None

Stop application

tasks: List[actfw_core.task.task.Task]
class actfw_core.application.SettingSchema(title: str, description: str, type_: type, default: Optional[Any] = None, ui_type: Optional[str] = None)

Bases: object

static decoder(obj: Any) → Any
static infertype(typestring: str) → type

actfw_core.capture module

class actfw_core.capture.Frame(value: actfw_core.capture.T)

Bases: Generic[actfw_core.capture.T]

getvalue() → actfw_core.capture.T

Get frame data.

Returns

captured image data

Return type

bytes

value: actfw_core.capture.T

Captured Frame

class actfw_core.capture.V4LCameraCapture(device: typing.Union[str, actfw_core.system.DeviceInfo] = '/dev/video0', size: typing.Tuple[int, int] = (640, 480), framerate: int = 30, expected_format: actfw_core.v4l2.video.V4L2_PIX_FMT = V4L2_PIX_FMT.RGB24, fallback_formats: typing.Iterable[actfw_core.v4l2.video.V4L2_PIX_FMT] = (<V4L2_PIX_FMT.YUYV: 1448695129>, <V4L2_PIX_FMT.MJPEG: 1196444237>), format_selector: actfw_core.capture.FormatSelector = FormatSelector.DEFAULT)

Bases: actfw_core.task.producer.Producer[actfw_core.capture.Frame[bytes]]

Parameters

  • device (str | DeviceInfo) – v4l device path or info

  • size (int, int) – expected capture resolution

  • framerate (int) – expected capture framerate

  • expected_format (V4L2_PIX_FMT) – expected capture format

  • fallback_formats (list of V4L2_PIX_FMT) – fallback capture format

  • ( (format_selector) – class:`~actfw_core.capture.V4LCameraCapture.FormatSelector): how to select a format from listed formats supported by a camera. # noqa: B006 B950 DEFAULT selects the first format that meets the conditions. PROPER selects the smallest format that meets the conditions. MAXIMUM selects the largest resolution format as a camera can. MAXIMUM ignores framerate parameters (uses appropriate framerate for the selected resolution). If a camera lists [1280x720, 1920x1080, 640x480, 800x600] and an expected capture resolution is (512, 512), DEFAULT selects 1280x720, PROPER selects 800x600 and MAXIMUM selects 1920x1080.

Notes

If a camera doesn’t support the expected_format, try to capture one of the fallback_formats and convert it to expected_format.

class FormatSelector(value)

Bases: enum.Enum

Captured Frame Producer for Video4Linux

DEFAULT = 1
MAXIMUM = 3
PROPER = 2
capture_format: actfw_core.v4l2.video.V4L2_PIX_FMT
capture_height: int
capture_size() → Tuple[int, int]

Get configured capture resolution. A configured resolution may be more larger than expected one.

Returns

configured capture resolution (width, height)

Return type

(int, int)

capture_width: int
configure(configurator: Callable[[actfw_core.v4l2.video.Video], actfw_core.capture.CONFIGURATOR_RETURN]) → actfw_core.capture.CONFIGURATOR_RETURN

Run user defined video configurator.

Parameters

configurator – unary function (actfw_core.v4l2.video.Video -> a)

Returns

return type of configurator

Return type

object

run() → None

Run producer activity

video: actfw_core.v4l2.video.Video

actfw_core.command_server module

class actfw_core.command_server.CommandServer(sock_path: Optional[str] = None)

Bases: actfw_core.task.isolated.Isolated

img: Optional[PIL.Image.Image]

Actcast Command Server

This server handles these commands

  • ‘Take Photo’

    • responses cached image as png data

img_lock: _thread.allocate_lock
run() → None

Run and start the activity

sock_path: Optional[str]
update_image(image: PIL.Image.Image) → None

Update the cached ‘Take Photo’ command image.

Parameters

image (Image) – image

actfw_core.service_client module

class actfw_core.service_client.ServiceClient(socket_path: Optional[pathlib.Path] = None)

Bases: object

rs256(payload: bytes) → str

Sign a message with an actcast device specific secret key.

Parameters

payload (bytes) – message

Returns

signature (base64url encoded)

Return type

str

Exceptions:

RuntimeError

actfw_core.system module

class actfw_core.system.DeviceInfo(type: str, nodes: List[actfw_core.system.DeviceNode])

Bases: object

static from_json(json: Dict[str, Any]) → actfw_core.system.DeviceInfo
nodes: List[actfw_core.system.DeviceNode]
type: str
class actfw_core.system.DeviceNode(path: pathlib.Path, label: Optional[str] = None)

Bases: object

static from_json(json: Dict[str, Any]) → actfw_core.system.DeviceNode
label: Optional[str] = None
path: pathlib.Path
class actfw_core.system.DeviceSupply(devices: List[actfw_core.system.DeviceInfo])

Bases: object

devices: List[actfw_core.system.DeviceInfo]
static from_json(json: Dict[str, Any]) → actfw_core.system.DeviceSupply
exception actfw_core.system.EnvironmentVariableNotSet(name: str)

Bases: Exception

actfw_core.system.find_csi_camera_device() → Optional[str]

Path of CSI camera device. Since ACTCAST_PROTOCOL_VERSION 1.3.0.

actfw_core.system.find_usb_camera_device() → Optional[str]

Path of USB camera device. Since ACTCAST_PROTOCOL_VERSION 1.3.0.

actfw_core.system.get_act_settings_path() → str

Path of act settings. This is rather low level API. Use get_settings in Application to get act settings. Since ACTCAST_PROTOCOL_VERSION 1.0.0.

actfw_core.system.get_actcast_act_id() → int

Act ID of the actcast application. Since ACTCAST_PROTOCOL_VERSION 1.0.0.

actfw_core.system.get_actcast_agent_simulator() → Optional[str]

Fixed value “actsim” if on actsim environment. Not set otherwise. Since ACTCAST_PROTOCOL_VERSION 1.1.0.

actfw_core.system.get_actcast_command_sock() → str

Path of socket file to receive command from actcast agent. This is rather low level API. Use CommandServer to receive commands from actcast agent. Since ACTCAST_PROTOCOL_VERSION 1.0.0.

actfw_core.system.get_actcast_device_id() → str

Device ID of the device running the actcast application. Since ACTCAST_PROTOCOL_VERSION 1.0.0.

actfw_core.system.get_actcast_device_type() → str

Device type. Since ACTCAST_PROTOCOL_VERSION 1.1.0.

actfw_core.system.get_actcast_firmware_type() → str

Firmware type of the host. Since ACTCAST_PROTOCOL_VERSION 1.3.0.

actfw_core.system.get_actcast_group_id() → int

Group ID that the device is belonging to. Since ACTCAST_PROTOCOL_VERSION 1.2.0.

actfw_core.system.get_actcast_instance_id() → str

ID that identifies the launch of the actcast application, like PID. Since ACTCAST_PROTOCOL_VERSION 1.0.0.

actfw_core.system.get_actcast_protocol_version() → str

Protocol version of agent-app communication. Since ACTCAST_PROTOCOL_VERSION 1.0.0.

actfw_core.system.get_actcast_service_sock() → str

Path of socket file to send command to actcast agent. This is rather low level API. Use ServiceClient to send commands to actcast agent. Since ACTCAST_PROTOCOL_VERSION 1.0.0.

actfw_core.system.get_actcast_socks_server() → Optional[str]

URL of SOCKS5 proxy server. Since ACTCAST_PROTOCOL_VERSION 1.0.0.

actfw_core.system.get_camera_device_info(default_image_source: Optional[str] = None) → actfw_core.system.DeviceInfo

DeviceInfo for camera. Set default_image_source only if you write camera device path in manifesto files. Since ACTCAST_PROTOCOL_VERSION 1.3.0.

actfw_core.system.get_device_supply() → actfw_core.system.DeviceSupply

Device Supply from actcast agent. Since ACTCAST_PROTOCOL_VERSION 1.3.0.

actfw_core.system.get_device_supply_path() → str

Path of device supply file. This is rather low level API. Use get_device_supply to get DeviceSupply object. Since ACTCAST_PROTOCOL_VERSION 1.3.0.

actfw_core.unicam_isp_capture module

class actfw_core.unicam_isp_capture.Auto(value)

Bases: enum.Enum

An enumeration.

AUTO = 1
class actfw_core.unicam_isp_capture.CameraMode(size: Tuple[int, int], scale: Tuple[float, float], crop: Tuple[int, int])

Bases: object

crop: Tuple[int, int]
scale: Tuple[float, float]
size: Tuple[int, int]
class actfw_core.unicam_isp_capture.UnicamIspCapture(unicam: str = '/dev/video0', unicam_subdev: str = '/dev/v4l-subdev0', isp_in: str = '/dev/video13', isp_out_high: str = '/dev/video14', isp_out_metadata: str = '/dev/video16', size: Tuple[int, int] = (640, 480), unicam_size: Optional[Tuple[int, int]] = None, crop_size: Optional[Tuple[int, int, int, int]] = None, framerate: int = 30, expected_format: actfw_core.v4l2.video.V4L2_PIX_FMT = V4L2_PIX_FMT.RGB24, auto_whitebalance: bool = True, init_controls: List[str] = [], agc: bool = True, target_Y: float = 0.16, alsc: bool = True, default_color_temperature: int = 4500, contrast: bool = True, config: Optional[Dict[str, Any]] = None, vflip: bool = False, hflip: bool = False, shutter_time: Union[float, actfw_core.unicam_isp_capture.Auto] = Auto.AUTO, analogue_gain: Union[float, actfw_core.unicam_isp_capture.Auto] = Auto.AUTO)

Bases: actfw_core.task.producer.Producer[actfw_core.capture.Frame[bytes]]

capture_size() → Tuple[int, int]
run() → None

Run and start the activity

set_exposure_settings(shutter_time: Union[float, actfw_core.unicam_isp_capture.Auto], analogue_gain: Union[float, actfw_core.unicam_isp_capture.Auto]) → None

Set shutter_time and analogue_gain.

Parameters

  • shutter_time (float or Auto) – [μsec] (e.g. 60000). If set to Auto, shutter_time is set with agc algorithm .

  • analogue_gain (float or Auto) – (>=0) (e.g. 2.5). If set to Auto, analogue_gain is set with agc algorithm .

set_exposure_time(ms: Optional[int] = None) → bool

Set exposure time.

This function is no longer supported. Use set_exposure_settings instead.

Returns

False

Module contents

actfw_core.heartbeat(*args: Any, **kwargs: Any) → None

Execute heartbeat action.

Notes

Default action is ‘touch /root/heartbeat’.

actfw_core.notify(notification: List[Dict[str, Any]], *args: Any, **kwargs: Any) → None

Make a notification to Actcast.

Parameters

notification (list of dict) – dicts must be encodable to JSON.

Example

>>> import actfw_core
>>> actfw_core.notify([{'msg': 'Hello!'}])
[{"msg": "Hello!"}]
actfw_core.set_heartbeat_function(f: Callable[[...], None]) → None

Set heartbeat action.

Parameters

f (function) – function which execute heartbeat action

Example

>>> import actfw_core
>>> def heartbeat(): print("working!")
...
>>> actfw_core.set_heartbeat_function(heartbeat)
>>> actfw_core.heartbeat()
working!