Lichen

lib/__builtins__/stream.py

932:c07b0dd14f85
10 months ago Paul Boddie Moved integer instantiation support to library functions.
     1 #!/usr/bin/env python     2      3 """     4 Stream objects.     5      6 Copyright (C) 2015, 2016, 2017 Paul Boddie <paul@boddie.org.uk>     7      8 This program is free software; you can redistribute it and/or modify it under     9 the terms of the GNU General Public License as published by the Free Software    10 Foundation; either version 3 of the License, or (at your option) any later    11 version.    12     13 This program is distributed in the hope that it will be useful, but WITHOUT    14 ANY WARRANTY; without even the implied warranty of MERCHANTABILITY or FITNESS    15 FOR A PARTICULAR PURPOSE.  See the GNU General Public License for more    16 details.    17     18 You should have received a copy of the GNU General Public License along with    19 this program.  If not, see <http://www.gnu.org/licenses/>.    20 """    21     22 from __builtins__.types import check_int, check_string    23 from native import isinstance as _isinstance, fclose, fflush, fread, fwrite    24     25 class filestream:    26     27     "Generic file-oriented stream functionality."    28     29     def __init__(self, encoding=None, bufsize=1024):    30     31         "Initialise the stream with the given 'encoding' and 'bufsize'."    32     33         self.encoding = encoding    34         self.bufsize = bufsize    35     36         # Internal stream details.    37     38         self.__data__ = None    39     40     def _convert(self, bytes):    41     42         "Convert 'bytes' to text if necessary."    43     44         if self.encoding:    45             return unicode(bytes, self.encoding)    46         else:    47             return bytes    48     49     def flush(self):    50     51         "Flush the stream."    52     53         fflush(self.__data__)    54     55     def read(self, n=0):    56     57         "Read 'n' bytes from the stream."    58     59         check_int(n)    60     61         # Read any indicated number of bytes.    62     63         if n > 0:    64             s = fread(self.__data__, n)    65     66         # Read all remaining bytes.    67     68         else:    69             l = []    70     71             # Read until end-of-file.    72     73             try:    74                 while True:    75                     self._read_data(l)    76     77             # Handle end-of-file reads.    78     79             except EOFError:    80                 pass    81     82             s = "".join(l)    83     84         return self._convert(s)    85     86     def readline(self, n=0):    87     88         """    89         Read until an end-of-line indicator is encountered or at most 'n' bytes,    90         if indicated.    91         """    92     93         check_int(n)    94     95         # Read any indicated number of bytes.    96     97         if n > 0:    98             s = fread(self.__data__, n)    99    100         # Read until an end-of-line indicator.   101    102         else:   103             l = []   104    105             # Read until end-of-line or end-of-file.   106    107             try:   108                 while not self._read_until_newline(l):   109                     pass   110    111             # Handle end-of-file reads.   112    113             except EOFError:   114                 pass   115    116             s = "".join(l)   117    118         return self._convert(s)   119    120     def _read_data(self, l):   121    122         "Read data into 'l'."   123    124         l.append(fread(self.__data__, self.bufsize))   125    126     def _read_until_newline(self, l):   127    128         "Read data into 'l', returning whether a newline has been read."   129    130         # NOTE: Only POSIX newlines are supported currently.   131    132         s = fread(self.__data__, 1)   133         l.append(s)   134         return s == "\n"   135    136     def readlines(self, n=None): pass   137    138     def write(self, s):   139    140         "Write string 's' to the stream."   141    142         check_string(s)   143    144         # Encode text as bytes if necessary. When the encoding is not set, any   145         # original encoding of the text will be applied.   146    147         if _isinstance(s, utf8string):   148             s = s.encode(self.encoding)   149    150         fwrite(self.__data__, s)   151    152     def close(self):   153    154         "Close the stream."   155    156         fclose(self.__data__)   157    158 # vim: tabstop=4 expandtab shiftwidth=4