This file is indexed.

/usr/share/pyshared/nxt/brick.py is in python-nxt 2.2.2-1.

This file is owned by root:root, with mode 0o644.

The actual contents of the file can be viewed below.

  1
  2
  3
  4
  5
  6
  7
  8
  9
 10
 11
 12
 13
 14
 15
 16
 17
 18
 19
 20
 21
 22
 23
 24
 25
 26
 27
 28
 29
 30
 31
 32
 33
 34
 35
 36
 37
 38
 39
 40
 41
 42
 43
 44
 45
 46
 47
 48
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
# nxt.brick module -- Classes to represent LEGO Mindstorms NXT bricks
# Copyright (C) 2006  Douglas P Lau
# Copyright (C) 2009  Marcus Wanner, rhn
# Copyright (C) 2010  rhn, Marcus Wanner, zonedabone
#
# This program is free software: you can redistribute it and/or modify
# it under the terms of the GNU General Public License as published by
# the Free Software Foundation, either version 3 of the License, or
# (at your option) any later version.
#
# This program is distributed in the hope that it will be useful,
# but WITHOUT ANY WARRANTY; without even the implied warranty of
# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
# GNU General Public License for more details.

from time import sleep
from threading import Lock
from nxt.error import FileNotFound, ModuleNotFound
from nxt.telegram import OPCODES, Telegram
from nxt.sensor import get_sensor
from nxt.motcont import MotCont

def _make_poller(opcode, poll_func, parse_func):
    def poll(self, *args, **kwargs):
        ogram = poll_func(opcode, *args, **kwargs)
        with self.lock:
            self.sock.send(str(ogram))
            if ogram.reply:
                igram = Telegram(opcode=opcode, pkt=self.sock.recv())
        if ogram.reply:
            return parse_func(igram)
        else:
            return None
    return poll

class _Meta(type):
    'Metaclass which adds one method for each telegram opcode'

    def __init__(cls, name, bases, dict):
        super(_Meta, cls).__init__(name, bases, dict)
        for opcode in OPCODES:
            poll_func, parse_func = OPCODES[opcode][0:2]
            m = _make_poller(opcode, poll_func, parse_func)
            try:
                m.__doc__ = OPCODES[opcode][2]
            except:
                pass
            setattr(cls, poll_func.__name__, m)


class FileFinder(object):
    'A generator to find files on a NXT brick.'

    def __init__(self, brick, pattern):
        self.brick = brick
        self.pattern = pattern
        self.handle = None

    def _close(self):
        if self.handle is not None:
            self.brick.close(self.handle)
            self.handle = None

    def __del__(self):
        self._close()

    def __iter__(self):
        results = []
        self.handle, fname, size = self.brick.find_first(self.pattern)
        results.append((fname, size))
        while True:
            try:
                handle, fname, size = self.brick.find_next(self.handle)
                results.append((fname, size))
            except FileNotFound:
                self._close()
                break
        for result in results:
            yield result


def File(brick, name, mode='r', size=None):
    """Opens a file for reading/writing. Mode is 'r' or 'w'. If mode is 'w',
    size must be provided.
    """
    if mode == 'w':
        if size is not None:
            return FileWriter(brick, name, size)
        else:
            return ValueError('Size not specified')
    elif mode == 'r':
        return FileReader(brick, name)
    else:
        return ValueError('Mode ' + str(mode) + ' not supported')


class FileReader(object):
    """Context manager to read a file on a NXT brick. Do use the iterator or
    the read() method, but not both at the same time!
    The iterator returns strings of an arbitrary (short) length.
    """

    def __init__(self, brick, fname):
        self.brick = brick
        self.handle, self.size = brick.open_read(fname)

    def read(self, bytes=None):
        if bytes is not None:
            remaining = bytes
        else:
            remaining = self.size
        bsize = self.brick.sock.bsize
        data = []
        while remaining > 0:
            handle, bsize, buffer_ = self.brick.read(self.handle,
                min(bsize, remaining))
            remaining -= len(buffer_)
            data.append(buffer_)
        return ''.join(data)

    def close(self):
        if self.handle is not None:
            self.brick.close(self.handle)
            self.handle = None

    def __del__(self):
        self.close()

    def __enter__(self):
        return self

    def __exit__(self, etp, value, tb):
        self.close()

    def __iter__(self):
        rem = self.size
        bsize = self.brick.sock.bsize
        while rem > 0:
            handle, bsize, data = self.brick.read(self.handle,
                min(bsize, rem))
            yield data
            rem -= len(data)


class FileWriter(object):
    "Object to write to a file on a NXT brick"

    def __init__(self, brick, fname, size):
        self.brick = brick
        self.handle = self.brick.open_write(fname, size)
        self._position = 0
        self.size = size

    def __del__(self):
        self.close()

    def close(self):
        if self.handle is not None:
            self.brick.close(self.handle)
            self.handle = None

    def tell(self):
        return self._position

    def write(self, data):
        remaining = len(data)
        if remaining > self.size - self._position:
            raise ValueError('Data will not fit into remaining space')
        bsize = self.brick.sock.bsize
        data_position = 0

        while remaining > 0:
            batch_size = min(bsize, remaining)
            next_data_position = data_position + batch_size
            buffer_ = data[data_position:next_data_position]

            handle, size = self.brick.write(self.handle, buffer_)

            self._position += batch_size
            data_position = next_data_position
            remaining -= batch_size


class ModuleFinder(object):
    'Iterator to lookup modules on a NXT brick'

    def __init__(self, brick, pattern):
        self.brick = brick
        self.pattern = pattern
        self.handle = None

    def _close(self):
        if self.handle:
            self.brick.close(self.handle)
            self.handle = None

    def __del__(self):
        self._close()

    def __iter__(self):
        self.handle, mname, mid, msize, miomap_size = \
            self.brick.request_first_module(self.pattern)
        yield (mname, mid, msize, miomap_size)
        while True:
            try:
                handle, mname, mid, msize, miomap_size = \
                    self.brick.request_next_module(
                    self.handle)
                yield (mname, mid, msize, miomap_size)
            except ModuleNotFound:
                self._close()
                break


class Brick(object): #TODO: this begs to have explicit methods
    'Main object for NXT Control'

    __metaclass__ = _Meta

    def __init__(self, sock):
        self.sock = sock
        self.lock = Lock()
        self.mc = MotCont(self)

    def play_tone_and_wait(self, frequency, duration):
        self.play_tone(frequency, duration)
        sleep(duration / 1000.0)

    def __del__(self):
        self.sock.close()

    find_files = FileFinder
    find_modules = ModuleFinder
    open_file = File
    get_sensor = get_sensor