Python与Fortran非文件式实时数据交互方案咨询及示例请求

你可以通过TCP/IP套接字实现两者的实时数据交互，无需依赖文件或第三方中间件，以下是两种角色分配的完整实现示例：

示例1：Python作为服务器，Fortran作为客户端 #

Python服务器代码 #

import socket

# 初始化TCP套接字
server_sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
server_sock.bind(('localhost', 12345))
server_sock.listen(1)
print("等待Fortran客户端连接...")

# 接受连接
conn, addr = server_sock.accept()
print(f"已建立连接：{addr}")

# 接收Fortran发送的初始c值
c_data = conn.recv(4)
c = int.from_bytes(c_data, byteorder='little')
print(f"收到Fortran的c值：{c}")

# 计算a = c + 1
a = c + 1
print(f"计算得到a值：{a}")

# 将a发送回Fortran
conn.sendall(a.to_bytes(4, byteorder='little'))

# 关闭连接
conn.close()
server_sock.close()

Fortran客户端代码 #

program fortran_client
    use, intrinsic :: iso_c_binding
    implicit none

    ! C套接字相关常量与类型定义
    integer(c_int), parameter :: AF_INET = 2, SOCK_STREAM = 1
    integer(c_int), parameter :: INADDR_ANY = 0

    type, bind(c) :: sockaddr_in
        integer(c_short) sin_family
        integer(c_short) sin_port
        integer(c_int) sin_addr
        character(c_char) sin_zero(8)
    end type sockaddr_in

    ! 绑定C套接字函数
    interface
        function socket(domain, type, protocol) bind(c, name='socket')
            import :: c_int
            integer(c_int) :: socket
            integer(c_int), value :: domain, type, protocol
        end function socket

        function connect(sockfd, addr, addrlen) bind(c, name='connect')
            import :: c_int, c_ptr
            integer(c_int) :: connect
            integer(c_int), value :: sockfd
            type(c_ptr), value :: addr
            integer(c_int), value :: addrlen
        end function connect

        function send(sockfd, buf, len, flags) bind(c, name='send')
            import :: c_int, c_ptr
            integer(c_int) :: send
            integer(c_int), value :: sockfd
            type(c_ptr), value :: buf
            integer(c_int), value :: len, flags
        end function send

        function recv(sockfd, buf, len, flags) bind(c, name='recv')
            import :: c_int, c_ptr
            integer(c_int) :: recv
            integer(c_int), value :: sockfd
            type(c_ptr), value :: buf
            integer(c_int), value :: len, flags
        end function recv

        function close(sockfd) bind(c, name='close')
            import :: c_int
            integer(c_int) :: close
            integer(c_int), value :: sockfd
        end function close

        function htons(hostshort) bind(c, name='htons')
            import :: c_short
            integer(c_short) :: htons
            integer(c_short), value :: hostshort
        end function htons
    end interface

    integer(c_int) :: sockfd, status
    type(sockaddr_in) :: server_addr
    integer :: c, a_recv, new_c
    integer(c_int), target :: c_bytes, a_recv_bytes

    ! 初始计算c值
    c = 1 + 2
    print *, "[Fortran] 初始计算c = ", c

    ! 创建套接字
    sockfd = socket(AF_INET, SOCK_STREAM, 0_c_int)
    if (sockfd == -1_c_int) then
        print *, "[Fortran] 创建套接字失败"
        stop
    end if

    ! 配置服务器地址
    server_addr%sin_family = AF_INET
    server_addr%sin_port = htons(12345_c_short)
    server_addr%sin_addr = transfer(INADDR_ANY, server_addr%sin_addr)
    server_addr%sin_zero = ['\0', '\0', '\0', '\0', '\0', '\0', '\0', '\0']

    ! 连接Python服务器
    status = connect(sockfd, c_loc(server_addr), sizeof(server_addr))
    if (status == -1_c_int) then
        print *, "[Fortran] 连接服务器失败"
        call close(sockfd)
        stop
    end if

    ! 发送c值到Python
    c_bytes = int(c, c_int)
    status = send(sockfd, c_loc(c_bytes), 4_c_int, 0_c_int)
    if (status == -1_c_int) then
        print *, "[Fortran] 发送c值失败"
        call close(sockfd)
        stop
    end if

    ! 接收Python返回的a值
    status = recv(sockfd, c_loc(a_recv_bytes), 4_c_int, 0_c_int)
    if (status == -1_c_int) then
        print *, "[Fortran] 接收a值失败"
        call close(sockfd)
        stop
    end if
    a_recv = int(a_recv_bytes, integer)
    print *, "[Fortran] 从Python收到a = ", a_recv

    ! 计算新的c值
    new_c = a_recv + 2
    print *, "[Fortran] 新计算的c = ", new_c

    ! 关闭套接字
    call close(sockfd)

end program fortran_client

运行步骤 #

1. 先执行Python服务器脚本，等待连接；
2. 编译Fortran代码（示例命令：gfortran fortran_client.f90 -o fortran_client），然后运行编译后的可执行文件；
3. 两者会自动建立连接，完成数据双向传输与计算。

示例2：Fortran作为服务器，Python作为客户端 #

如果需要Fortran主动监听连接，可使用以下代码：

Fortran服务器代码 #

program fortran_server
    use, intrinsic :: iso_c_binding
    implicit none

    integer(c_int), parameter :: AF_INET = 2, SOCK_STREAM = 1
    integer(c_int), parameter :: INADDR_ANY = 0

    type, bind(c) :: sockaddr_in
        integer(c_short) sin_family
        integer(c_short) sin_port
        integer(c_int) sin_addr
        character(c_char) sin_zero(8)
    end type sockaddr_in

    interface
        function socket(domain, type, protocol) bind(c, name='socket')
            import :: c_int
            integer(c_int) :: socket
            integer(c_int), value :: domain, type, protocol
        end function socket

        function bind(sockfd, addr, addrlen) bind(c, name='bind')
            import :: c_int, c_ptr
            integer(c_int) :: bind
            integer(c_int), value :: sockfd
            type(c_ptr), value :: addr
            integer(c_int), value :: addrlen
        end function bind

        function listen(sockfd, backlog) bind(c, name='listen')
            import :: c_int
            integer(c_int) :: listen
            integer(c_int), value :: sockfd, backlog
        end function listen

        function accept(sockfd, addr, addrlen) bind(c, name='accept')
            import :: c_int, c_ptr
            integer(c_int) :: accept
            integer(c_int), value :: sockfd
            type(c_ptr), value :: addr
            type(c_ptr), value :: addrlen
        end function accept

        function send(sockfd, buf, len, flags) bind(c, name='send')
            import :: c_int, c_ptr
            integer(c_int) :: send
            integer(c_int), value :: sockfd
            type(c_ptr), value :: buf
            integer(c_int), value :: len, flags
        end function send

        function recv(sockfd, buf, len, flags) bind(c, name='recv')
            import :: c_int, c_ptr
            integer(c_int) :: recv
            integer(c_int), value :: sockfd
            type(c_ptr), value :: buf
            integer(c_int), value :: len, flags
        end function recv

        function close(sockfd) bind(c, name='close')
            import :: c_int
            integer(c_int) :: close
            integer(c_int), value :: sockfd
        end function close

        function htons(hostshort) bind(c, name='htons')
            import :: c_short
            integer(c_short) :: htons
            integer(c_short), value :: hostshort
        end function htons
    end interface

    integer(c_int) :: sockfd, connfd, status
    type(sockaddr_in) :: server_addr, client_addr
    integer(c_int) :: addr_len
    integer :: c, a_recv, new_c
    integer(c_int), target :: c_bytes, a_recv_bytes

    ! 创建套接字
    sockfd = socket(AF_INET, SOCK_STREAM, 0_c_int)
    if (sockfd == -1_c_int) then
        print *, "[Fortran] 创建套接字失败"
        stop
    end if

    ! 配置服务器地址
    server_addr%sin_family = AF_INET
    server_addr%sin_port = htons(12345_c_short)
    server_addr%sin_addr = transfer(INADDR_ANY, server_addr%sin_addr)
    server_addr%sin_zero = ['\0', '\0', '\0', '\0', '\0', '\0', '\0', '\0']

    ! 绑定地址
    status = bind(sockfd, c_loc(server_addr), sizeof(server_addr))
    if (status == -1_c_int) then
        print *, "[Fortran] 绑定地址失败"
        call close(sockfd)
        stop
    end if

    ! 监听连接
    status = listen(sockfd, 1_c_int)
    if (status == -1_c_int) then
        print *, "[Fortran] 监听失败"
        call close(sockfd)
        stop
    end if
    print *, "[Fortran] 等待Python客户端连接..."

    ! 接受连接
    addr_len = sizeof(client_addr)
    connfd = accept(sockfd, c_loc(client_addr), c_loc(addr_len))
    if (connfd == -1_c_int) then
        print *, "[Fortran] 接受连接失败"
        call close(sockfd)
        stop
    end if
    print *, "[Fortran] Python客户端已连接"

    ! 初始计算c并发送
    c = 1 + 2
    print *, "[Fortran] 初始计算c = ", c
    c_bytes = int(c, c_int)
    status = send(connfd, c_loc(c_bytes), 4_c_int, 0_c_int)
    if (status == -1_c_int) then
        print *, "[Fortran] 发送c值失败"
        call close(connfd)
        call close(sockfd)
        stop
    end if

    ! 接收Python返回的a值
    status = recv(connfd, c_loc(a_recv_bytes), 4_c_int, 0_c_int)
    if (status == -1_c_int) then
        print *, "[Fortran] 接收a值失败"
        call close(connfd)
        call close(sockfd)
        stop
    end if
    a_recv = int(a_recv_bytes, integer)
    print *, "[Fortran] 从Python收到a = ", a_recv

    ! 计算新的c值
    new_c = a_recv + 2
    print *, "[Fortran] 新计算的c = ", new_c

    ! 关闭连接
    call close(connfd)
    call close(sockfd)

end program fortran_server

Python客户端代码 #

import socket

# 初始化TCP套接字并连接
client_sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
client_sock.connect(('localhost', 12345))
print("已连接到Fortran服务器")

# 接收Fortran发送的c值
c_data = client_sock.recv(4)
c = int.from_bytes(c_data, byteorder='little')
print(f"收到Fortran的c值：{c}")

# 计算a = c + 1
a = c + 1
print(f"计算得到a值：{a}")

# 发送a值到Fortran
client_sock.sendall(a.to_bytes(4, byteorder='little'))

# 关闭连接
client_sock.close()

内容的提问来源于stack exchange，提问作者Yurii Yarema