Python Script Service (f8.pyscript)¶

Standalone python script runtime service with lifecycle/tick/command hooks.

• Service class: f8.pyscript
• Version: 0.0.1
• Source directory: f8/pyscript
• Tags: python, script, service

When to Use¶

• Use f8.pyscript for Python logic that has outgrown a single expression but is not yet formalized as dedicated operators.
• It is useful for multi-line scripts, custom experimentation, and intermediate prototype logic.
• It works well as a temporary home for evolving graph behavior.

Common Wiring Patterns¶

• Use it as a bounded custom logic layer while the design is still changing.
• Once the behavior stabilizes, consider moving the logic into PyEngine operators or a dedicated service.
• Keep inputs and outputs explicit so the script block does not become an unreadable center of gravity.

Pitfalls / Gotchas¶

• If the script keeps getting longer and harder to test, it is time to refactor.
• Clear input/output schemas help far more than "we will clean it up later".
• If the behavior should be reused across graphs, promote it into a more explicit module or operator.

Service Reference¶

How to Run¶

pixi run -e default f8pyscript

• Workdir: ../../../
• Environment overrides: none

Typical Inputs / Outputs¶

• Data inputs: in
• Data outputs: out, monitor
• Commands: grant_local_exec, revoke_local_exec

Service State Fields¶

Key Fields That Matter¶

• code (Code, rw): Python source code. Schema: `string / default=# Hooks template (uncomment what you need):

- onStart(ctx)¶

- onStop(ctx)¶

- onPause(ctx, meta=None)¶

- onResume(ctx, meta=None)¶

- onState(ctx, field, value, ts_ms=None)¶

- onData(ctx, port, value, ts_ms=None)¶

- onTick(ctx, tick)¶

- onCommand(ctx, name, args, meta=None)¶

¶

Useful context helpers:¶

- ctx.states. reads cached rw/ro/wo state snapshot¶

- example: ctx.states.tickEnabled / ctx.states.lastError¶

- await ctx.read_state(field) # fresh runtime read¶

- ctx.states.get(field) # cached snapshot¶

- ctx.set_state(field, value)¶

- await ctx.set_state_async(field, value)¶

- ctx.emit(port, value)¶

- ctx.permission.local_exec_granted / ctx.permission.expires_ts_ms¶

- TypeGuard helpers are available from f8_dynamic_inputs¶

- example: from f8_dynamic_inputs import is_port_in¶

- optional: from f8_dynamic_inputs import *¶

- then: if is_port_in(value, port): ...¶

- State TypeGuard helpers are available from f8_dynamic_states¶

- example: from f8_dynamic_states import is_state_tickEnabled¶

- then: if is_state_tickEnabled(value, field): ...¶

¶

from typing import TYPE_CHECKING, Any if TYPE_CHECKING: from f8_script_api import F8PyScriptContext, F8States, F8Tick

¶

def onStart(ctx: 'F8PyScriptContext') -> None: ctx.log('pyscript started')

def onStop(ctx: 'F8PyScriptContext') -> None:¶

ctx.log('pyscript stopped')¶

¶

def onPause(ctx: 'F8PyScriptContext', meta: dict[str, Any] | None = None) -> None:¶

ctx.log(f'paused: {meta}')¶

¶

def onResume(ctx: 'F8PyScriptContext', meta: dict[str, Any] | None = None) -> None:¶

ctx.log(f'resumed: {meta}')¶

¶

def onState(¶

ctx: 'F8PyScriptContext',¶

field: str,¶

value: Any,¶

ts_ms: int | None = None,¶

) -> None:¶

ctx.log(f'state {field}={value} ts_ms={ts_ms}')¶

¶

def onData(¶

ctx: 'F8PyScriptContext',¶

port: str,¶

value: Any,¶

ts_ms: int | None = None,¶

) -> None:¶

ctx.log(f'data port={port} value={value} ts_ms={ts_ms}')¶

¶

def onTick(ctx: 'F8PyScriptContext', tick: 'F8Tick') -> None:¶

ctx.log(f'tick seq={tick.seq} tsMs={tick.tsMs} deltaMs={tick.deltaMs}')¶

¶

def onCommand(¶

ctx: 'F8PyScriptContext',¶

name: str,¶

args: dict[str, Any],¶

meta: dict[str, Any] | None = None,¶

) -> dict[str, Any]:¶

if name == 'ping':¶

return {'ok': True, 'result': {'pong': True}}¶

return {'ok': False, 'error': f'unknown command: {name}'}¶

. -lastError(Last Error,ro): Last script compile/runtime error. Schema:string / default=. -tickEnabled(Tick Enabled,rw): Enable onTick scheduler. Schema:boolean / default=False. -tickMs(Tick Interval (ms),rw): onTick interval in milliseconds. Schema:integer / default=100. -active(Active,rw): Service lifecycle state (activate/deactivate). Schema:boolean / default=True. -svcId(Service Id,ro): Readonly: current service instance id (svcId). Schema:string`.

Service Commands¶

grant_local_exec¶

Grant local execution for this script session.

• Show on node: true

revoke_local_exec¶

Revoke local execution grant.

• Show on node: true
• Params: none

Service Data Input Ports¶

Service Data Output Ports¶

Operators¶

None

Related Scenarios¶

• No bundled scenario references this node yet.