<commit_before>#!/usr/bin/python

"""
ReadWriteFileServer.py

This sample application is a BACnet device that has one record access file at
('file', 1) and one stream access file at ('file', 2).
"""

import random
import string

from bacpypes.debugging import bacpypes_debugging, ModuleLogger
from bacpypes.consolelogging import ConfigArgumentParser

from bacpypes.core import run

from bacpypes.app import LocalDeviceObject, BIPSimpleApplication
from bacpypes.object import FileObject, register_object_type

from bacpypes.basetypes import ServicesSupported

# some debugging
_debug = 0
_log = ModuleLogger(globals())

# configuration
RECORD_LEN = 128
RECORD_COUNT = 100
OCTET_COUNT = 4096

#
#   Local Record Access File Object Type
#

@bacpypes_debugging
class LocalRecordAccessFileObject(FileObject):

    def __init__(self, **kwargs):
        """ Initialize a record accessed file object. """
        if _debug:
            LocalRecordAccessFileObject._debug("__init__ %r",
                kwargs,
                )
        FileObject.__init__(self,
            fileAccessMethod='recordAccess',
             **kwargs
             )

        self._record_data = [
            ''.join(random.choice(string.ascii_letters)
            for i in range(RECORD_LEN))
            for j in range(RECORD_COUNT)
            ]
        if _debug: LocalRecordAccessFileObject._debug("    - %d records",
                len(self._record_data),
                )

    def __len__(self):
        """ Return the number of records. """
        if _debug: LocalRecordAccessFileObject._debug("__len__")

        return len(self._record_data)

    def ReadFile(self, start_record, record_count):
        """ Read a number of records starting at a specific record. """
        if _debug: LocalRecordAccessFileObject._debug("ReadFile %r %r",
                start_record, record_count,
                )

        # end of file is true if last record is returned
        end_of_file = (start_record+record_count) >= len(self._record_data)

        return end_of_file, \
            self._record_data[start_record:start_record + record_count]

    def WriteFile(self, start_record, record_count, record_data):
        """ Write a number of records, starting at a specific record. """
        # check for append
        if (start_record < 0):
            start_record = len(self._record_data)
            self._record_data.extend(record_data)

        # check to extend the file out to start_record records
        elif (start_record > len(self._record_data)):
            self._record_data.extend(['' for i in range(start_record - len(self._record_data))])
            start_record = len(self._record_data)
            self._record_data.extend(record_data)

        # slice operation works for other cases
        else:
            self._record_data[start_record:start_record + record_count] = record_data

        # return where the 'writing' actually started
        return start_record

register_object_type(LocalRecordAccessFileObject)

#
#   Local Stream Access File Object Type
#

@bacpypes_debugging
class LocalStreamAccessFileObject(FileObject):

    def __init__(self, **kwargs):
        """ Initialize a stream accessed file object. """
        if _debug:
            LocalStreamAccessFileObject._debug("__init__ %r",
                kwargs,
                )
        FileObject.__init__(self,
            fileAccessMethod='streamAccess',
             **kwargs
             )

        self._file_data = ''.join(random.choice(string.ascii_letters)
            for i in range(OCTET_COUNT))
        if _debug: LocalRecordAccessFileObject._debug("    - %d octets",
                len(self._file_data),
                )

    def __len__(self):
        """ Return the number of octets in the file. """
        if _debug: LocalStreamAccessFileObject._debug("__len__")

        return len(self._file_data)

    def ReadFile(self, start_position, octet_count):
        """ Read a chunk of data out of the file. """
        if _debug: LocalStreamAccessFileObject._debug("ReadFile %r %r",
                start_position, octet_count,
                )

        # end of file is true if last record is returned
        end_of_file = (start_position+octet_count) >= len(self._file_data)

        return end_of_file, \
            self._file_data[start_position:start_position + octet_count]

    def WriteFile(self, start_position, data):
        """ Write a number of octets, starting at a specific offset. """
        # check for append
        if (start_position < 0):
            start_position = len(self._file_data)
            self._file_data += data

        # check to extend the file out to start_record records
        elif (start_position > len(self._file_data)):
            self._file_data += '\0' * (start_position - len(self._file_data))
            start_position = len(self._file_data)
            self._file_data += data

        # no slice assignment, strings are immutable 
        else:
            data_len = len(data)
            prechunk = self._file_data[:start_position]
            postchunk = self._file_data[start_position + data_len:]
            self._file_data = prechunk + data + postchunk<commit_msg>optionally get configuration values from the environment, add some debugging<commit_after>#!/usr/bin/python

"""
This sample application is a BACnet device that has one record access file
('file', 1) and one stream access file ('file', 2).
"""

import os
import random
import string

from bacpypes.debugging import bacpypes_debugging, ModuleLogger
from bacpypes.consolelogging import ConfigArgumentParser

from bacpypes.core import run

from bacpypes.app import LocalDeviceObject, BIPSimpleApplication
from bacpypes.object import FileObject, register_object_type

from bacpypes.basetypes import ServicesSupported

# some debugging
_debug = 0
_log = ModuleLogger(globals())

# configuration
RECORD_LEN = int(os.getenv('RECORD_LEN', 128))
RECORD_COUNT = int(os.getenv('RECORD_COUNT', 100))
OCTET_COUNT = int(os.getenv('OCTET_COUNT', 4096))

#
#   Local Record Access File Object Type
#

@bacpypes_debugging
class LocalRecordAccessFileObject(FileObject):

    def __init__(self, **kwargs):
        """ Initialize a record accessed file object. """
        if _debug:
            LocalRecordAccessFileObject._debug("__init__ %r",
                kwargs,
                )
        FileObject.__init__(self,
            fileAccessMethod='recordAccess',
             **kwargs
             )

        self._record_data = [
            ''.join(random.choice(string.ascii_letters)
            for i in range(RECORD_LEN))
            for j in range(RECORD_COUNT)
            ]
        if _debug: LocalRecordAccessFileObject._debug("    - %d records",
                len(self._record_data),
                )

    def __len__(self):
        """ Return the number of records. """
        if _debug: LocalRecordAccessFileObject._debug("__len__")

        return len(self._record_data)

    def ReadFile(self, start_record, record_count):
        """ Read a number of records starting at a specific record. """
        if _debug: LocalRecordAccessFileObject._debug("ReadFile %r %r",
                start_record, record_count,
                )

        # end of file is true if last record is returned
        end_of_file = (start_record+record_count) >= len(self._record_data)

        return end_of_file, \
            self._record_data[start_record:start_record + record_count]

    def WriteFile(self, start_record, record_count, record_data):
        """ Write a number of records, starting at a specific record. """
        if _debug: LocalRecordAccessFileObject._debug("WriteFile %r %r %r",
                start_record, record_count, record_data,
                )

        # check for append
        if (start_record < 0):
            start_record = len(self._record_data)
            self._record_data.extend(record_data)

        # check to extend the file out to start_record records
        elif (start_record > len(self._record_data)):
            self._record_data.extend(['' for i in range(start_record - len(self._record_data))])
            start_record = len(self._record_data)
            self._record_data.extend(record_data)

        # slice operation works for other cases
        else:
            self._record_data[start_record:start_record + record_count] = record_data

        # return where the 'writing' actually started
        return start_record

register_object_type(LocalRecordAccessFileObject)

#
#   Local Stream Access File Object Type
#

@bacpypes_debugging
class LocalStreamAccessFileObject(FileObject):

    def __init__(self, **kwargs):
        """ Initialize a stream accessed file object. """
        if _debug:
            LocalStreamAccessFileObject._debug("__init__ %r",
                kwargs,
                )
        FileObject.__init__(self,
            fileAccessMethod='streamAccess',
             **kwargs
             )

        self._file_data = ''.join(random.choice(string.ascii_letters)
            for i in range(OCTET_COUNT))
        if _debug: LocalRecordAccessFileObject._debug("    - %d octets",
                len(self._file_data),
                )

    def __len__(self):
        """ Return the number of octets in the file. """
        if _debug: LocalStreamAccessFileObject._debug("__len__")

        return len(self._file_data)

    def ReadFile(self, start_position, octet_count):
        """ Read a chunk of data out of the file. """
        if _debug: LocalStreamAccessFileObject._debug("ReadFile %r %r",
                start_position, octet_count,
                )

        # end of file is true if last record is returned
        end_of_file = (start_position+octet_count) >= len(self._file_data)

        return end_of_file, \
            self._file_data[start_position:start_position + octet_count]

    def WriteFile(self, start_position, data):
        """ Write a number of octets, starting at a specific offset. """
        if _debug: LocalStreamAccessFileObject._debug("WriteFile %r %r",
                start_position, data,
                )

        # check for append
        if (start_position < 0):
            start_position = len(self._file_data)
            self._file_data += data

        # check to extend the file out to start_record records
        elif (start_position > len(self._file_data)):
            self._file_data += '\0' * (start_position - len(self._file_data))
            start_position = len(self._file_data)
            self._file_data += data

        # no slice assignment, strings are immutable 
        else:
            data_len = len(data)
            prechunk = self._file_data[:start_position]
            postchunk = self._file_data[start_position + data_len:]
            self._file_data = prechunk + data + postchunk
{'fraction_non_alphanumeric': 0.08597785977859779, 'fraction_numerical': 0.002952029520295203, 'mean_word_length': 2.46912, 'pattern_counts': {'":': 0, '<': 7, '<?xml version=': 0, '>': 11, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 0, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '4308117', 'n_tokens_mistral': 3058, 'n_tokens_neox': 2939, 'n_words': 782}