Skip to content

Instantly share code, notes, and snippets.

@arwie

arwie/codesys_python.md

Last active February 19, 2025 08:55
Download ZIP

Revisions

  1. arwie revised this gist Jan 4, 2025. 1 changed file with 3 additions and 1 deletion.
    4 changes: 3 additions & 1 deletion codesys_python.md
    View file
    Original file line number Diff line number Diff line change
    @@ -1,4 +1,6 @@
    Fast variable exchange between CODESYS and PYTHON via shared memory
    Fast variable exchange between
    CODESYS and PYTHON
    via shared memory
    ===


  2. arwie revised this gist Oct 15, 2024. 1 changed file with 13 additions and 13 deletions.
    26 changes: 13 additions & 13 deletions codesys_python.md
    View file
    Original file line number Diff line number Diff line change
    @@ -18,18 +18,18 @@ TYPE AppCmd:
    STRUCT
    axis_power: BOOL;
    move_exec: BOOL;
    move_distance: LREAL;
    move_velocity: LREAL;
    io: ARRAY [0..128] OF BOOL;
    move_distance: LREAL;
    move_velocity: LREAL;
    io: ARRAY [0..128] OF BOOL;
    END_STRUCT
    END_TYPE
    TYPE AppFbk:
    STRUCT
    axis_powered: BOOL;
    axis_powered: BOOL;
    move_done: BOOL;
    move_error: BOOL;
    io: ARRAY [0..128] OF BOOL;
    io: ARRAY [0..128] OF BOOL;
    END_STRUCT
    END_TYPE
    ~~~
    @@ -81,27 +81,27 @@ Instantiate the structures as *cmd* and *fbk*.
    class AppCmd(Structure):
    axis_power: bool
    move_exec: bool
    move_distance: float
    move_velocity: float
    io: list[bool]
    move_distance: float
    move_velocity: float
    io: list[bool]
    _fields_ = [
    ('axis_power', c_bool),
    ('move_exec', c_bool),
    ('move_distance', c_double),
    ('move_velocity', c_double),
    ('io', c_bool * 129),
    ('io', c_bool * 129),
    ]

    class AppFbk(Structure):
    axis_powered: bool
    axis_powered: bool
    move_done: bool
    move_error: bool
    io: list[bool]
    io: list[bool]
    _fields_ = [
    ('axis_powered', c_bool),
    ('move_done', c_bool),
    ('move_error', c_bool),
    ('io', c_bool * 129),
    ('io', c_bool * 129),
    ]

    cmd = AppCmd()
    @@ -143,7 +143,7 @@ Assign the variables from *cmd* to the inputs of the function blocks and use the
    ~~~ iecst
    PROGRAM main
    VAR
    Power: MC_Power;
    Power: MC_Power;
    MoveRelative: MC_MoveRelative;
    END_VAR
  3. arwie revised this gist Oct 15, 2024. 1 changed file with 21 additions and 18 deletions.
    39 changes: 21 additions & 18 deletions codesys_python.md
    View file
    Original file line number Diff line number Diff line change
    @@ -3,15 +3,15 @@ Fast variable exchange between CODESYS and PYTHON via shared memory



    A video of this working example is on [YouTube](https://www.youtube.com/watch?v=UcGE2JFpIMA) and full PYTHON source code and CODESYS project is on [GitHub](https://github.com/arwie/controlOS_demo/tree/youtube_2024-10-03).
    A video of this working example is available on [YouTube](https://www.youtube.com/watch?v=UcGE2JFpIMA) and complete PYTHON source code and CODESYS project are available on [GitHub](https://github.com/arwie/controlOS_demo/tree/youtube_2024-10-03).



    Setup on CODESYS side
    Setup on the CODESYS side
    ---

    We begin with the CODESYS project and define variables that we want to exchange in two structures.
    *AppCmd* will be the data we receive from PYTHON and *AppFbk* contains variables we want to send back to PYTHON.
    Start with the CODESYS project and define the variables to be exchanged in two structures.
    *AppCmd* is the data received from PYTHON and *AppFbk* contains the variables to be sent back to PYTHON.

    ~~~ iecst
    TYPE AppCmd:
    @@ -34,8 +34,8 @@ END_STRUCT
    END_TYPE
    ~~~

    Next we instantiate our structures as *cmd* and *fbk* inside a program called *app*.
    On init we create a semaphore and a shared memory block big enogh to hold *cmd* and *fbk*.
    Next, instantiate the structures as *cmd* and *fbk* within a program named *app*.
    At init, create a semaphore and a shared memory block large enough to hold *cmd* and *fbk*.

    ~~~ iecst
    PROGRAM app
    @@ -54,7 +54,7 @@ sem := SysSemProcessCreate('codesys', ADR(iec_result));
    shm := SysSharedMemoryCreate('codesys', 0, ADR(shm_size), ADR(iec_result));
    ~~~

    Now we have to call the next code block cyclically to read *cmd* from and write *fbk* to shared memory.
    Now cyclically call the next block of code to read *cmd* from shared memory and write *fbk* to shared memory.

    ~~~ iecst
    IF SysSemProcessEnter(sem, 0) = 0 THEN
    @@ -66,14 +66,16 @@ ELSE
    END_IF
    ~~~

    Of course we protect the read/write operation with our semaphore. As we never want to delay our CODESYS task and wait for non-real-time PYTHON we set zero as timeout of *SysSemProcessEnter*.
    Protect the read/write operation with the semaphore. In order not to delay the CODESYS task and wait for non real-time PYTHON, set the timeout of *SysSemProcessEnter* to zero.



    Setup on PYTHON side
    Setup on the PYTHON side
    ---

    Fortunately CODESYS structures are binary compatible with C so we can use the *ctypes* module to recompose our *AppCmd* and *AppFbk* in PYTHON. Here only the *\_fields\_* list is used at runtime. The type annotation before are useful for static type checkers. We instantiate the structures as *cmd* and *fbk*.
    Since CODESYS structures are binary compatible with C, use the *ctypes* module to recompose *AppCmd* and *AppFbk* in PYTHON.
    Only the *\_fields\_* list is used at runtime; the type annotations above are useful for static type checkers.
    Instantiate the structures as *cmd* and *fbk*.

    ~~~ python
    class AppCmd(Structure):
    @@ -106,9 +108,10 @@ cmd = AppCmd()
    fbk = AppFbk()
    ~~~

    The creation of the obove structures can be automated using CODESYS scripting. Here is an [example](https://github.com/arwie/controlOS_codesys/blob/main/codesys/export_types.py).
    The creation of the above structures can be automated using CODESYS scripting. Check out the following [example](https://github.com/arwie/controlOS_codesys/blob/main/codesys/export_types.py).

    We use the extern *posix_ipc* module to open our semaphore and shared memory and create a mapfile using *mmap*. After calculating all source and destination addresses for the data transfer we periodically call *memmove* to copy *cmd* to and *fbk* from our shared memory. Of course we again protect the read/write operation with the semaphore.
    Use the external *posix_ipc* module to open the semaphore and shared memory and create a mapfile using *mmap*.
    After calculating all source and destination addresses for the data transfer, periodically call *memmove* to copy *cmd* to shared memory and *fbk* from shared memory. Again, protect the read/write operation with the semaphore.

    ~~~ python
    sem = posix_ipc.Semaphore('/codesys')
    @@ -127,15 +130,15 @@ with mmap(shm.fd, shm.size) as mapfile:
    await asyncio.sleep(period)
    ~~~

    Now we are able to communicate with a running CODESYS application in almost real-time.
    Now the communication with a running CODESYS application takes place almost in real time.



    Executing CODESYS SoftMotion function blocks from PYTHON
    Execution of CODESYS SoftMotion function blocks from PYTHON
    ---

    As an example we create the following program in our CODESYS project. It uses two function blocks to power and move an axis.
    We assign the the variables from *cmd* to the inputs of the function blocks and use the outputs to set *fbk* variables.
    Create the following program as an example in the CODESYS project. It uses two function blocks to power and move an axis.
    Assign the variables from *cmd* to the inputs of the function blocks and use the outputs to set the variables in *fbk*.

    ~~~ iecst
    PROGRAM main
    @@ -166,7 +169,7 @@ app.fbk.move_done := MoveRelative.Done;
    app.fbk.move_error := MoveRelative.Error;
    ~~~

    Now we can define PYTHON functions to control COSDESYS funtion blocks.
    Now define PYTHON functions to control COSDESYS function blocks.

    ~~~ python
    async def poll(condition, *, abort=None, timeout=None)
    @@ -199,7 +202,7 @@ async def move(distance, velocity):
    await poll(lambda: not (fbk.move_done or fbk.move_error))
    ~~~

    Finally we write a short demo program to move a CODESYS axis in a PYTHONic way.
    Finally, write a short demo program to move a CODESYS axis in a PYTHONic way.

    ~~~ python
    async def run():
  4. arwie created this gist Oct 3, 2024.
    211 changes: 211 additions & 0 deletions codesys_python.md
    View file
    Original file line number Diff line number Diff line change
    @@ -0,0 +1,211 @@
    Fast variable exchange between CODESYS and PYTHON via shared memory
    ===



    A video of this working example is on [YouTube](https://www.youtube.com/watch?v=UcGE2JFpIMA) and full PYTHON source code and CODESYS project is on [GitHub](https://github.com/arwie/controlOS_demo/tree/youtube_2024-10-03).



    Setup on CODESYS side
    ---

    We begin with the CODESYS project and define variables that we want to exchange in two structures.
    *AppCmd* will be the data we receive from PYTHON and *AppFbk* contains variables we want to send back to PYTHON.

    ~~~ iecst
    TYPE AppCmd:
    STRUCT
    axis_power: BOOL;
    move_exec: BOOL;
    move_distance: LREAL;
    move_velocity: LREAL;
    io: ARRAY [0..128] OF BOOL;
    END_STRUCT
    END_TYPE
    TYPE AppFbk:
    STRUCT
    axis_powered: BOOL;
    move_done: BOOL;
    move_error: BOOL;
    io: ARRAY [0..128] OF BOOL;
    END_STRUCT
    END_TYPE
    ~~~

    Next we instantiate our structures as *cmd* and *fbk* inside a program called *app*.
    On init we create a semaphore and a shared memory block big enogh to hold *cmd* and *fbk*.

    ~~~ iecst
    PROGRAM app
    VAR
    cmd: AppCmd;
    fbk: AppFbk;
    shm: RTS_IEC_HANDLE;
    sem: RTS_IEC_HANDLE;
    shm_size: __UXINT;
    iec_result: RTS_IEC_RESULT;
    conflicts: UDINT := 0;
    END_VAR
    shm_size := SIZEOF(fbk) + SIZEOF(cmd);
    sem := SysSemProcessCreate('codesys', ADR(iec_result));
    shm := SysSharedMemoryCreate('codesys', 0, ADR(shm_size), ADR(iec_result));
    ~~~

    Now we have to call the next code block cyclically to read *cmd* from and write *fbk* to shared memory.

    ~~~ iecst
    IF SysSemProcessEnter(sem, 0) = 0 THEN
    SysSharedMemoryRead (shm, 0, ADR(cmd), SIZEOF(cmd), ADR(iec_result));
    SysSharedMemoryWrite(shm, SIZEOF(cmd), ADR(fbk), SIZEOF(fbk), ADR(iec_result));
    SysSemProcessLeave(sem);
    ELSE
    conflicts := conflicts + 1;
    END_IF
    ~~~

    Of course we protect the read/write operation with our semaphore. As we never want to delay our CODESYS task and wait for non-real-time PYTHON we set zero as timeout of *SysSemProcessEnter*.



    Setup on PYTHON side
    ---

    Fortunately CODESYS structures are binary compatible with C so we can use the *ctypes* module to recompose our *AppCmd* and *AppFbk* in PYTHON. Here only the *\_fields\_* list is used at runtime. The type annotation before are useful for static type checkers. We instantiate the structures as *cmd* and *fbk*.

    ~~~ python
    class AppCmd(Structure):
    axis_power: bool
    move_exec: bool
    move_distance: float
    move_velocity: float
    io: list[bool]
    _fields_ = [
    ('axis_power', c_bool),
    ('move_exec', c_bool),
    ('move_distance', c_double),
    ('move_velocity', c_double),
    ('io', c_bool * 129),
    ]

    class AppFbk(Structure):
    axis_powered: bool
    move_done: bool
    move_error: bool
    io: list[bool]
    _fields_ = [
    ('axis_powered', c_bool),
    ('move_done', c_bool),
    ('move_error', c_bool),
    ('io', c_bool * 129),
    ]

    cmd = AppCmd()
    fbk = AppFbk()
    ~~~

    The creation of the obove structures can be automated using CODESYS scripting. Here is an [example](https://github.com/arwie/controlOS_codesys/blob/main/codesys/export_types.py).

    We use the extern *posix_ipc* module to open our semaphore and shared memory and create a mapfile using *mmap*. After calculating all source and destination addresses for the data transfer we periodically call *memmove* to copy *cmd* to and *fbk* from our shared memory. Of course we again protect the read/write operation with the semaphore.

    ~~~ python
    sem = posix_ipc.Semaphore('/codesys')
    shm = posix_ipc.SharedMemory('/codesys')
    with mmap(shm.fd, shm.size) as mapfile:

    cmd_addr, cmd_size = addressof(cmd), sizeof(cmd)
    fbk_addr, fbk_size = addressof(fbk), sizeof(fbk)
    shm_cmd_addr = addressof(c_byte.from_buffer(mapfile))
    shm_fbk_addr = addressof(c_byte.from_buffer(mapfile, cmd_size))

    while True:
    with sem:
    memmove(shm_cmd_addr, cmd_addr, cmd_size)
    memmove(fbk_addr, shm_fbk_addr, fbk_size)
    await asyncio.sleep(period)
    ~~~

    Now we are able to communicate with a running CODESYS application in almost real-time.



    Executing CODESYS SoftMotion function blocks from PYTHON
    ---

    As an example we create the following program in our CODESYS project. It uses two function blocks to power and move an axis.
    We assign the the variables from *cmd* to the inputs of the function blocks and use the outputs to set *fbk* variables.

    ~~~ iecst
    PROGRAM main
    VAR
    Power: MC_Power;
    MoveRelative: MC_MoveRelative;
    END_VAR
    Power(
    Axis := AxisMR,
    Enable := TRUE,
    bRegulatorOn := app.cmd.axis_power,
    bDriveStart := app.cmd.axis_power,
    );
    app.fbk.axis_powered := Power.Status;
    MoveRelative(
    Axis := AxisMR,
    Execute := app.cmd.move_exec,
    Distance := app.cmd.move_distance,
    BufferMode := MC_BUFFER_MODE.Aborting,
    Velocity := app.cmd.move_velocity,
    Acceleration := AxisMR.fSWMaxAcceleration,
    Deceleration := AxisMR.fSWMaxDeceleration,
    Jerk := AxisMR.fSWMaxJerk,
    );
    app.fbk.move_done := MoveRelative.Done;
    app.fbk.move_error := MoveRelative.Error;
    ~~~

    Now we can define PYTHON functions to control COSDESYS funtion blocks.

    ~~~ python
    async def poll(condition, *, abort=None, timeout=None)
    while True:
    if condition():
    return True
    if abort() or timeout(): #pseudocode
    return False
    await asyncio.sleep(poll_period)

    @asynccontextmanager
    async def power():
    cmd.axis_power = True
    try:
    if not await poll(lambda: fbk.axis_powered, timeout=1):
    raise Exception('Failed to power on axis')
    yield
    finally:
    cmd.axis_power = False

    async def move(distance, velocity):
    cmd.move_distance = distance
    cmd.move_velocity = velocity
    cmd.move_exec = True
    try:
    if not await poll(lambda: fbk.move_done, abort=lambda: fbk.move_error):
    raise Exception('Failed to move axis')
    finally:
    cmd.move_exec = False
    await poll(lambda: not (fbk.move_done or fbk.move_error))
    ~~~

    Finally we write a short demo program to move a CODESYS axis in a PYTHONic way.

    ~~~ python
    async def run():
    async with power():
    await move(+400, 300)
    await move(-700, 400)
    await move(+900, 500)
    ~~~