1

#!/usr/bin/env python

     2

# -*- coding: utf-8 -*-

     3

     4

# This file is part of Beremiz Runtime

     5

#

     6

# Copyright (C) 2013: Laurent BESSARD

     7

# Copyright (C) 2017: Andrey Skvortsov

     8

# Copyright (C) 2025: Edouard Tisserant

     9

#

    10

# See COPYING file for copyrights details.

    11

    12

import csv

    13

from collections import OrderedDict

    14

    15

csv_int_files = {}

    16

cvs_int_changed = set()

    17

csv_str_files = {}

    18

cvs_str_changed = set()

    19

    20

class Entry():

    21

    def __init__(self, *args):

    22

        self.args = args

    23

    def __call__(self):

    24

        return self.args

    25

    26

def _CSV_int_Load(fname):

    27

    global csv_int_files

    28

    entry = csv_int_files.get(fname, None)

    29

    if entry is None:

    30

        data = list()

    31

        csvfile = open(fname, 'rt', encoding='utf-8')

    32

    33

        try:

    34

            dialect = csv.Sniffer().sniff(csvfile.read(1024))

    35

            csvfile.seek(0)

    36

            reader = csv.reader(csvfile, dialect)

    37

            for row in reader:

    38

                data.append(row)

    39

        finally:

    40

            csvfile.close()

    41

        entry = Entry(fname, dialect, data)

    42

        csv_int_files[fname] = entry

    43

    return entry

    44

    45

    46

def _CSV_str_Load(fname):

    47

    global csv_str_files

    48

    entry = csv_str_files.get(fname, None)

    49

    if entry is None:

    50

        data = []

    51

        csvfile = open(fname, 'rt', encoding='utf-8')

    52

    53

        try:

    54

            dialect = csv.Sniffer().sniff(csvfile.read(1024))

    55

            csvfile.seek(0)

    56

            reader = csv.reader(csvfile, dialect)

    57

            first_row = reader.next()

    58

            data.append(first_row)

    59

            col_headers = OrderedDict([(name, index+1) for index, name 

    60

                                        in enumerate(first_row[1:])])

    61

            max_row_len = len(first_row)

    62

            row_headers = OrderedDict()

    63

            for index, row in enumerate(reader):

    64

                row_headers[row[0]] = index+1

    65

                data.append(row)

    66

                max_row_len = max(max_row_len, len(row))

    67

        finally:

    68

            csvfile.close()

    69

        entry = Entry(fname, dialect, col_headers, row_headers, max_row_len, data)

    70

        csv_str_files[fname] = entry

    71

    return entry

    72

    73

    74

def _CSV_str_Create(fname):

    75

    global csv_str_files

    76

    data = [[]] # start with an empty row, acounting for header row

    77

    dialect = None

    78

    col_headers = OrderedDict()

    79

    row_headers = OrderedDict()

    80

    max_row_len = 1  # set to one initialy, accounting for header column

    81

    entry = Entry(fname, dialect, col_headers, row_headers, max_row_len, data)

    82

    csv_str_files[fname] = entry

    83

    return entry

    84

    85

    86

def _CSV_Save_data(fname, dialect, data):

    87

    try:

    88

        wfile = open(fname, 'wt')

    89

        writer = csv.writer(wfile) if not(dialect) else csv.writer(wfile, dialect)

    90

        for row in data:

    91

            writer.writerow(row)

    92

    finally:

    93

        wfile.close()

    94

    95

def _CSV_int_Save(entry):

    96

    fname, dialect, data = entry()

    97

    _CSV_Save_data(fname, dialect, data)

    98

    99

   100

def _CSV_str_Save(entry):

   101

    fname, dialect, col_headers, row_headers, max_row_len, data = entry()

   102

    _CSV_Save_data(fname, dialect, data)

   103

   104

   105

_already_registered_cb = False

   106

def _CSV_OnIdle_callback():

   107

    global _already_registered_cb, cvs_int_changed, cvs_str_changed

   108

    _already_registered_cb = False

   109

    while len(cvs_int_changed):

   110

        entry = cvs_int_changed.pop()

   111

        _CSV_int_Save(entry)

   112

   113

    while len(cvs_str_changed):

   114

        entry = cvs_str_changed.pop()

   115

        _CSV_str_Save(entry)

   116

   117

   118

def _CSV_register_OnIdle_callback():

   119

    global _already_registered_cb

   120

    if not _already_registered_cb:

   121

        OnIdle.append(_CSV_OnIdle_callback)

   122

        _already_registered_cb = True

   123

   124

   125

def _CSV_int_modified(entry):

   126

    global cvs_int_changed

   127

    cvs_int_changed.add(entry)

   128

    _CSV_register_OnIdle_callback()

   129

   130

   131

def _CSV_str_modified(entry):

   132

    global cvs_str_changed

   133

    cvs_str_changed.add(entry)

   134

    _CSV_register_OnIdle_callback()

   135

   136

   137

def CSVRdInt(fname, rowidx, colidx):

   138

    """

   139

    Return value at row/column pointed by integer indexes

   140

    Assumes data starts at first row and first column, no headers.

   141

    """

   142

   143

    try:

   144

        _fname, _dialect, data = _CSV_int_Load(fname)()

   145

    except IOError:

   146

        return "#FILE_NOT_FOUND"

   147

    except csv.Error as e:

   148

        return "#CSV_ERROR"

   149

   150

    try:

   151

        row = data[rowidx]

   152

        if not row and rowidx == len(data)-1:

   153

            raise IndexError

   154

    except IndexError:

   155

        return "#ROW_NOT_FOUND"

   156

   157

    try:

   158

        return row[colidx]

   159

    except IndexError:

   160

        return "#COL_NOT_FOUND"

   161

   162

   163

def CSVRdStr(fname, rowname, colname):

   164

    """

   165

    Return value at row/column pointed by a pair of names as string

   166

    Assumes first row is column headers and first column is row name.

   167

    """

   168

   169

    if not rowname:

   170

        return "#INVALID_ROW"

   171

    if not colname:

   172

        return "#INVALID_COLUMN"

   173

   174

    try:

   175

        fname, dialect, col_headers, row_headers, max_row_len, data = _CSV_str_Load(fname)()

   176

    except IOError:

   177

        return "#FILE_NOT_FOUND"

   178

    except csv.Error:

   179

        return "#CSV_ERROR"

   180

   181

    try:

   182

        rowidx = row_headers[rowname]

   183

    except KeyError:

   184

        return "#ROW_NOT_FOUND"

   185

   186

    try:

   187

        colidx = col_headers[colname]

   188

    except KeyError:

   189

        return "#COL_NOT_FOUND"

   190

   191

    try:

   192

        return data[rowidx][colidx]

   193

    except IndexError:

   194

        return "#COL_NOT_FOUND"

   195

   196

   197

def CSVWrInt(fname, rowidx, colidx, content):

   198

    """

   199

    Update value at row/column pointed by integer indexes

   200

    Assumes data starts at first row and first column, no headers.

   201

    """

   202

   203

    try:

   204

        entry = _CSV_int_Load(fname)

   205

    except IOError:

   206

        return "#FILE_NOT_FOUND"

   207

    except csv.Error as e:

   208

        return "#CSV_ERROR"

   209

   210

    fname, dialect, data = entry()

   211

    try:

   212

        if rowidx == len(data):

   213

            row = []

   214

            data.append(row)

   215

        else:

   216

            row = data[rowidx]

   217

    except IndexError:

   218

        return "#ROW_NOT_FOUND"

   219

   220

    try:

   221

        if rowidx > 0 and colidx >= len(data[0]):

   222

            raise IndexError

   223

        if colidx >= len(row):

   224

            row.extend([""] * (colidx - len(row)) + [content])

   225

        else:

   226

            row[colidx] = content

   227

    except IndexError:

   228

        return "#COL_NOT_FOUND"

   229

   230

    _CSV_int_modified(entry)

   231

   232

    return "OK"

   233

   234

   235

def CSVWrStr(fname, rowname, colname, content):

   236

    """

   237

    Update value at row/column pointed by a pair of names as string.

   238

    Assumes first row is column headers and first column is row name.

   239

    """

   240

   241

    if not rowname:

   242

        return "#INVALID_ROW"

   243

    if not colname:

   244

        return "#INVALID_COLUMN"

   245

   246

    try:

   247

        entry = _CSV_str_Load(fname)

   248

    except IOError:

   249

        entry = _CSV_str_Create(fname)

   250

    except csv.Error:

   251

        return "#CSV_ERROR"

   252

   253

    fname, dialect, col_headers, row_headers, max_row_len, data = entry()

   254

    try:

   255

        rowidx = row_headers[rowname]

   256

        row = data[rowidx]

   257

    except KeyError:

   258

        # create a new row with appropriate header

   259

        row = [rowname]

   260

        # put it at the end

   261

        rowidx = len(data)

   262

        data.append(row)

   263

        row_headers[rowname] = rowidx

   264

   265

    try:

   266

        colidx = col_headers[colname]

   267

    except KeyError:

   268

        # adjust col headers content

   269

        first_row = data[0] 

   270

        first_row += [""]*(max_row_len - len(first_row)) + [colname]

   271

        # create a new column

   272

        colidx = col_headers[colname] = max_row_len

   273

        max_row_len = max_row_len + 1

   274

   275

    try:

   276

        row[colidx] = content

   277

    except IndexError:

   278

        # create a new cell

   279

        row += [""]*(colidx - len(row)) + [content]

   280

   281

    _CSV_str_modified(entry)

   282

   283

    return "OK"

   284

   285

   286

def CSVReload():

   287

    global csv_int_files, csv_str_files, cvs_int_changed, cvs_str_changed

   288

   289

    # Force saving modified CSV files

   290

    _CSV_OnIdle_callback()

   291

   292

    # Wipe data model

   293

    csv_int_files.clear()

   294

    csv_str_files.clear()

   295

    cvs_int_changed.clear()

   296

    cvs_str_changed.clear()

   297