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