Skip to content

state

StateType = TypeVar('StateType') module-attribute

all_states = [field.name for field in dataclasses.fields(StateList)] module-attribute

states = StateList() module-attribute

AvailableMemoryState

Bases: State[int]

Source code in roc/jupyter/state.py 
67
68
69
70
71
72
73
74
75
76
77
78
class AvailableMemoryState(State[int]):
    def __init__(self) -> None:
        super().__init__("sysmem")
        self.val = self.get()

    def __str__(self) -> str:
        vm = self.get()
        return f"Available System Memory: {bytes2human(vm)}"

    def get(self) -> int:
        vm = psutil.virtual_memory()
        return cast(int, vm.available)

val = self.get() instance-attribute

__init__()

Source code in roc/jupyter/state.py 
68
69
70
def __init__(self) -> None:
    super().__init__("sysmem")
    self.val = self.get()

__str__()

Source code in roc/jupyter/state.py 
72
73
74
def __str__(self) -> str:
    vm = self.get()
    return f"Available System Memory: {bytes2human(vm)}"

get()

Source code in roc/jupyter/state.py 
76
77
78
def get(self) -> int:
    vm = psutil.virtual_memory()
    return cast(int, vm.available)

ComponentsState

Bases: State[list[str]]

Source code in roc/jupyter/state.py 
233
234
235
236
237
238
239
240
241
242
243
244
class ComponentsState(State[list[str]]):
    def __init__(self) -> None:
        super().__init__("components", display_name="Components")
        self.val = []

    def get(self) -> list[str]:
        self.val = Component.get_loaded_components()
        return self.val

    def __str__(self) -> str:
        component_str = "\t" + "\n\t".join(self.get())
        return f"{Component.get_component_count()} components loaded:\n{component_str}"

val = [] instance-attribute

__init__()

Source code in roc/jupyter/state.py 
234
235
236
def __init__(self) -> None:
    super().__init__("components", display_name="Components")
    self.val = []

__str__()

Source code in roc/jupyter/state.py 
242
243
244
def __str__(self) -> str:
    component_str = "\t" + "\n\t".join(self.get())
    return f"{Component.get_component_count()} components loaded:\n{component_str}"

get()

Source code in roc/jupyter/state.py 
238
239
240
def get(self) -> list[str]:
    self.val = Component.get_loaded_components()
    return self.val

CpuLoadState

Bases: State[list[float]]

Source code in roc/jupyter/state.py 
81
82
83
84
85
86
87
88
89
90
91
class CpuLoadState(State[list[float]]):
    def __init__(self) -> None:
        super().__init__("cpuload")
        self.val = self.get()

    def __str__(self) -> str:
        load = self.get()
        return f"CPU Load: {load[0]:1.1f}% / {load[1]:1.1f}% / {load[2]:1.1f}% (1m / 5m / 15m)"

    def get(self) -> list[float]:
        return [x / psutil.cpu_count() * 100 for x in psutil.getloadavg()]

val = self.get() instance-attribute

__init__()

Source code in roc/jupyter/state.py 
82
83
84
def __init__(self) -> None:
    super().__init__("cpuload")
    self.val = self.get()

__str__()

Source code in roc/jupyter/state.py 
86
87
88
def __str__(self) -> str:
    load = self.get()
    return f"CPU Load: {load[0]:1.1f}% / {load[1]:1.1f}% / {load[2]:1.1f}% (1m / 5m / 15m)"

get()

Source code in roc/jupyter/state.py 
90
91
def get(self) -> list[float]:
    return [x / psutil.cpu_count() * 100 for x in psutil.getloadavg()]

CurrentAttentionState

Bases: State[VisionAttentionData]

Source code in roc/jupyter/state.py 
204
205
206
207
208
209
210
211
212
213
214
215
216
class CurrentAttentionState(State[VisionAttentionData]):
    def __init__(self) -> None:
        super().__init__("curr-saliency", display_name="Current Saliency Map")

    def set(self, att: VisionAttentionData) -> None:
        self.val = att

    def __str__(self) -> str:
        if self.val is not None:
            s = f"Current Attention:\n{str(self.val)}\n"
            return s
        else:
            return "Current Attention: None"

__init__()

Source code in roc/jupyter/state.py 
205
206
def __init__(self) -> None:
    super().__init__("curr-saliency", display_name="Current Saliency Map")

__str__()

Source code in roc/jupyter/state.py 
211
212
213
214
215
216
def __str__(self) -> str:
    if self.val is not None:
        s = f"Current Attention:\n{str(self.val)}\n"
        return s
    else:
        return "Current Attention: None"

set(att)

Source code in roc/jupyter/state.py 
208
209
def set(self, att: VisionAttentionData) -> None:
    self.val = att

CurrentObjectState

Bases: State[Object]

Source code in roc/jupyter/state.py 
219
220
221
222
223
224
225
226
227
228
229
230
class CurrentObjectState(State[Object]):
    def __init__(self) -> None:
        super().__init__("curr-object", display_name="Current Object")

    def set(self, obj: Object) -> None:
        self.val = obj

    def __str__(self) -> str:
        if self.val is not None:
            return f"Current Object:\n{str(self.val)}\n"
        else:
            return "Current Object: None"

__init__()

Source code in roc/jupyter/state.py 
220
221
def __init__(self) -> None:
    super().__init__("curr-object", display_name="Current Object")

__str__()

Source code in roc/jupyter/state.py 
226
227
228
229
230
def __str__(self) -> str:
    if self.val is not None:
        return f"Current Object:\n{str(self.val)}\n"
    else:
        return "Current Object: None"

set(obj)

Source code in roc/jupyter/state.py 
223
224
def set(self, obj: Object) -> None:
    self.val = obj

CurrentSaliencyMapState

Bases: State[SaliencyMap]

Source code in roc/jupyter/state.py 
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
class CurrentSaliencyMapState(State[SaliencyMap]):
    def __init__(self) -> None:
        super().__init__("curr-saliency", display_name="Current Saliency Map")

    def set(self, sal: SaliencyMap) -> None:
        self.val = sal

    def __str__(self) -> str:
        if self.val is not None:
            s = f"Current Saliency Map:\n{str(self.val)}\n"
            s += "\tFeatures:\n"
            features = self.val.feature_report()
            for feat_name in features:
                s += f"\t\t{feat_name}: {features[feat_name]}\n"
            return s
        else:
            return "Current Saliency Map: None"

__init__()

Source code in roc/jupyter/state.py 
186
187
def __init__(self) -> None:
    super().__init__("curr-saliency", display_name="Current Saliency Map")

__str__()

Source code in roc/jupyter/state.py 
192
193
194
195
196
197
198
199
200
201
def __str__(self) -> str:
    if self.val is not None:
        s = f"Current Saliency Map:\n{str(self.val)}\n"
        s += "\tFeatures:\n"
        features = self.val.feature_report()
        for feat_name in features:
            s += f"\t\t{feat_name}: {features[feat_name]}\n"
        return s
    else:
        return "Current Saliency Map: None"

set(sal)

Source code in roc/jupyter/state.py 
189
190
def set(self, sal: SaliencyMap) -> None:
    self.val = sal

CurrentScreenState

Bases: State[str]

Source code in roc/jupyter/state.py 
171
172
173
174
175
176
177
178
179
180
181
182
class CurrentScreenState(State[str]):
    def __init__(self) -> None:
        super().__init__("curr-screen", display_name="Current Screen")

    def set(self, screen: str) -> None:
        self.val = screen

    def __str__(self) -> str:
        if self.val is not None:
            return f"Current Screen:\n-------------\n{self.val}\n-------------"
        else:
            return "Current Screen: None"

__init__()

Source code in roc/jupyter/state.py 
172
173
def __init__(self) -> None:
    super().__init__("curr-screen", display_name="Current Screen")

__str__()

Source code in roc/jupyter/state.py 
178
179
180
181
182
def __str__(self) -> str:
    if self.val is not None:
        return f"Current Screen:\n-------------\n{self.val}\n-------------"
    else:
        return "Current Screen: None"

set(screen)

Source code in roc/jupyter/state.py 
175
176
def set(self, screen: str) -> None:
    self.val = screen

DiskIoState

Bases: State[dict[str, float]]

Source code in roc/jupyter/state.py 
 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
class DiskIoState(State[dict[str, float]]):
    def __init__(self) -> None:
        super().__init__("diskio")
        self.last_time = time.time_ns()
        ioc = psutil.disk_io_counters()
        self.last_read_io = ioc.read_count
        self.last_write_io = ioc.write_count
        self.last_read_bytes = ioc.read_bytes
        self.last_write_bytes = ioc.write_bytes

    def __str__(self) -> str:
        disk_io = self.get()
        read_io = disk_io["read_io"]
        write_io = disk_io["write_io"]
        read_bytes = int(disk_io["read_bytes"])
        write_bytes = int(disk_io["write_bytes"])
        return f"Disk Read I/O: {read_io:1.1f}/s ({bytes2human(read_bytes)}/s), Write I/O {write_io:1.1f}/s ({bytes2human(write_bytes)}/s)"

    def get(self) -> dict[str, float]:
        ioc = psutil.disk_io_counters()
        now = time.time_ns()
        delta_sec = (now - self.last_time) / 10e8
        read_io_per_sec = (ioc.read_count - self.last_read_io) / delta_sec
        write_io_per_sec = (ioc.write_count - self.last_write_io) / delta_sec
        read_bytes_per_sec = (ioc.read_bytes - self.last_read_bytes) / delta_sec
        write_bytes_per_sec = (ioc.write_bytes - self.last_write_bytes) / delta_sec
        self.last_read_io = ioc.read_count
        self.last_write_io = ioc.write_count
        self.last_read_bytes = ioc.read_bytes
        self.last_write_bytes = ioc.write_bytes
        self.last_time = now

        return {
            "read_io": read_io_per_sec,
            "write_io": write_io_per_sec,
            "read_bytes": read_bytes_per_sec,
            "write_bytes": write_bytes_per_sec,
        }

last_read_bytes = ioc.read_bytes instance-attribute

last_read_io = ioc.read_count instance-attribute

last_time = time.time_ns() instance-attribute

last_write_bytes = ioc.write_bytes instance-attribute

last_write_io = ioc.write_count instance-attribute

__init__()

Source code in roc/jupyter/state.py 
 95
 96
 97
 98
 99
100
101
102
def __init__(self) -> None:
    super().__init__("diskio")
    self.last_time = time.time_ns()
    ioc = psutil.disk_io_counters()
    self.last_read_io = ioc.read_count
    self.last_write_io = ioc.write_count
    self.last_read_bytes = ioc.read_bytes
    self.last_write_bytes = ioc.write_bytes

__str__()

Source code in roc/jupyter/state.py 
104
105
106
107
108
109
110
def __str__(self) -> str:
    disk_io = self.get()
    read_io = disk_io["read_io"]
    write_io = disk_io["write_io"]
    read_bytes = int(disk_io["read_bytes"])
    write_bytes = int(disk_io["write_bytes"])
    return f"Disk Read I/O: {read_io:1.1f}/s ({bytes2human(read_bytes)}/s), Write I/O {write_io:1.1f}/s ({bytes2human(write_bytes)}/s)"

get()

Source code in roc/jupyter/state.py 
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
def get(self) -> dict[str, float]:
    ioc = psutil.disk_io_counters()
    now = time.time_ns()
    delta_sec = (now - self.last_time) / 10e8
    read_io_per_sec = (ioc.read_count - self.last_read_io) / delta_sec
    write_io_per_sec = (ioc.write_count - self.last_write_io) / delta_sec
    read_bytes_per_sec = (ioc.read_bytes - self.last_read_bytes) / delta_sec
    write_bytes_per_sec = (ioc.write_bytes - self.last_write_bytes) / delta_sec
    self.last_read_io = ioc.read_count
    self.last_write_io = ioc.write_count
    self.last_read_bytes = ioc.read_bytes
    self.last_write_bytes = ioc.write_bytes
    self.last_time = now

    return {
        "read_io": read_io_per_sec,
        "write_io": write_io_per_sec,
        "read_bytes": read_bytes_per_sec,
        "write_bytes": write_bytes_per_sec,
    }

EdgeCacheState

Bases: State[float]

Source code in roc/jupyter/state.py 
157
158
159
160
161
162
163
164
165
166
167
168
class EdgeCacheState(State[float]):
    def __init__(self) -> None:
        super().__init__("edge-cache", display_name="Edge Cache")
        self.val = 0

    def get(self) -> float:
        c = Edge.get_cache()
        return c.currsize / c.maxsize

    def __str__(self) -> str:
        c = Edge.get_cache()
        return f"Edge Cache: {c.currsize} / {c.maxsize} ({self.get():1.1f}%)"

val = 0 instance-attribute

__init__()

Source code in roc/jupyter/state.py 
158
159
160
def __init__(self) -> None:
    super().__init__("edge-cache", display_name="Edge Cache")
    self.val = 0

__str__()

Source code in roc/jupyter/state.py 
166
167
168
def __str__(self) -> str:
    c = Edge.get_cache()
    return f"Edge Cache: {c.currsize} / {c.maxsize} ({self.get():1.1f}%)"

get()

Source code in roc/jupyter/state.py 
162
163
164
def get(self) -> float:
    c = Edge.get_cache()
    return c.currsize / c.maxsize

LoopState

Bases: State[int]

Source code in roc/jupyter/state.py 
134
135
136
137
138
139
140
class LoopState(State[int]):
    def __init__(self) -> None:
        super().__init__("loop", display_name="Loop Number")
        self.val = 0

    def incr(self) -> None:
        self.val = self.get() + 1

val = 0 instance-attribute

__init__()

Source code in roc/jupyter/state.py 
135
136
137
def __init__(self) -> None:
    super().__init__("loop", display_name="Loop Number")
    self.val = 0

incr()

Source code in roc/jupyter/state.py 
139
140
def incr(self) -> None:
    self.val = self.get() + 1

NodeCacheState

Bases: State[float]

Source code in roc/jupyter/state.py 
143
144
145
146
147
148
149
150
151
152
153
154
class NodeCacheState(State[float]):
    def __init__(self) -> None:
        super().__init__("node-cache", display_name="Node Cache")
        self.val = 0

    def get(self) -> float:
        c = Node.get_cache()
        return c.currsize / c.maxsize

    def __str__(self) -> str:
        c = Node.get_cache()
        return f"Node Cache: {c.currsize} / {c.maxsize} ({self.get():1.1f}%)"

val = 0 instance-attribute

__init__()

Source code in roc/jupyter/state.py 
144
145
146
def __init__(self) -> None:
    super().__init__("node-cache", display_name="Node Cache")
    self.val = 0

__str__()

Source code in roc/jupyter/state.py 
152
153
154
def __str__(self) -> str:
    c = Node.get_cache()
    return f"Node Cache: {c.currsize} / {c.maxsize} ({self.get():1.1f}%)"

get()

Source code in roc/jupyter/state.py 
148
149
150
def get(self) -> float:
    c = Node.get_cache()
    return c.currsize / c.maxsize

ProcessMemoryState

Bases: State[int]

Source code in roc/jupyter/state.py 
52
53
54
55
56
57
58
59
60
61
62
63
64
class ProcessMemoryState(State[int]):
    def __init__(self) -> None:
        super().__init__("memory")
        self.val = self.get()

    def __str__(self) -> str:
        mem = self.get()
        return f"Process Memory Usage: {bytes2human(mem)}"

    def get(self) -> int:
        process = psutil.Process(os.getpid())
        mem_info = process.memory_info()
        return cast(int, mem_info.rss)

val = self.get() instance-attribute

__init__()

Source code in roc/jupyter/state.py 
53
54
55
def __init__(self) -> None:
    super().__init__("memory")
    self.val = self.get()

__str__()

Source code in roc/jupyter/state.py 
57
58
59
def __str__(self) -> str:
    mem = self.get()
    return f"Process Memory Usage: {bytes2human(mem)}"

get()

Source code in roc/jupyter/state.py 
61
62
63
64
def get(self) -> int:
    process = psutil.Process(os.getpid())
    mem_info = process.memory_info()
    return cast(int, mem_info.rss)

State

Bases: ABC, Generic[StateType]

Source code in roc/jupyter/state.py 
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
class State(ABC, Generic[StateType]):
    def __init__(self, name: str, display_name: str | None = None) -> None:
        self.name = name
        self.display_name = display_name or name
        self.val: StateType | None = None

    def __str__(self) -> str:
        return f"{self.display_name}: {self.val}"

    def get(self) -> StateType:
        if self.val is None:
            raise Exception("Trying to get state value before it is set")

        return self.val

    def set(self, v: StateType) -> None:
        self.val = v

display_name = display_name or name instance-attribute

name = name instance-attribute

val = None instance-attribute

__init__(name, display_name=None)

Source code in roc/jupyter/state.py 
24
25
26
27
def __init__(self, name: str, display_name: str | None = None) -> None:
    self.name = name
    self.display_name = display_name or name
    self.val: StateType | None = None

__str__()

Source code in roc/jupyter/state.py 
29
30
def __str__(self) -> str:
    return f"{self.display_name}: {self.val}"

get()

Source code in roc/jupyter/state.py 
32
33
34
35
36
def get(self) -> StateType:
    if self.val is None:
        raise Exception("Trying to get state value before it is set")

    return self.val

set(v)

Source code in roc/jupyter/state.py 
38
39
def set(self, v: StateType) -> None:
    self.val = v

StateComponent

Bases: Component

Source code in roc/jupyter/state.py 
267
268
class StateComponent(Component):
    pass

StateList dataclass

Source code in roc/jupyter/state.py 
247
248
249
250
251
252
253
254
255
256
257
258
259
260
@dataclass
class StateList:
    memory: ProcessMemoryState = ProcessMemoryState()
    sysmem: AvailableMemoryState = AvailableMemoryState()
    loop: LoopState = LoopState()
    cpuload: CpuLoadState = CpuLoadState()
    diskio: DiskIoState = DiskIoState()
    node_cache: NodeCacheState = NodeCacheState()
    edge_cache: EdgeCacheState = EdgeCacheState()
    screen: CurrentScreenState = CurrentScreenState()
    salency: CurrentSaliencyMapState = CurrentSaliencyMapState()
    attention: CurrentAttentionState = CurrentAttentionState()
    object: CurrentObjectState = CurrentObjectState()
    components: ComponentsState = ComponentsState()

attention = CurrentAttentionState() class-attribute instance-attribute

components = ComponentsState() class-attribute instance-attribute

cpuload = CpuLoadState() class-attribute instance-attribute

diskio = DiskIoState() class-attribute instance-attribute

edge_cache = EdgeCacheState() class-attribute instance-attribute

loop = LoopState() class-attribute instance-attribute

memory = ProcessMemoryState() class-attribute instance-attribute

node_cache = NodeCacheState() class-attribute instance-attribute

object = CurrentObjectState() class-attribute instance-attribute

salency = CurrentSaliencyMapState() class-attribute instance-attribute

screen = CurrentScreenState() class-attribute instance-attribute

sysmem = AvailableMemoryState() class-attribute instance-attribute

__init__(memory=ProcessMemoryState(), sysmem=AvailableMemoryState(), loop=LoopState(), cpuload=CpuLoadState(), diskio=DiskIoState(), node_cache=NodeCacheState(), edge_cache=EdgeCacheState(), screen=CurrentScreenState(), salency=CurrentSaliencyMapState(), attention=CurrentAttentionState(), object=CurrentObjectState(), components=ComponentsState())

SystemCpuState

Bases: State[int]

Source code in roc/jupyter/state.py 
42
43
44
45
46
47
48
49
class SystemCpuState(State[int]):
    def __init__(self) -> None:
        super().__init__("cpu", display_name="CPU Usage")
        self.val = self.get()

    def get(self) -> int:
        psutil.cpu_times()
        return 1

val = self.get() instance-attribute

__init__()

Source code in roc/jupyter/state.py 
43
44
45
def __init__(self) -> None:
    super().__init__("cpu", display_name="CPU Usage")
    self.val = self.get()

get()

Source code in roc/jupyter/state.py 
47
48
49
def get(self) -> int:
    psutil.cpu_times()
    return 1

init_state()

Source code in roc/jupyter/state.py 
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
def init_state() -> None:
    global _state_init_done
    if _state_init_done:
        return

    # attention
    att_conn = Attention.bus.connect(StateComponent())

    def att_evt_handler(e: Event[VisionAttentionData]) -> None:
        assert isinstance(e.data, VisionAttentionData)
        states.salency.set(deepcopy(e.data.saliency_map))
        states.attention.set(deepcopy(e.data))

    att_conn.listen(att_evt_handler, filter=lambda e: isinstance(e.data, VisionAttentionData))

    # object
    obj_conn = ObjectResolver.bus.connect(StateComponent())

    def obj_evt_handler(e: Event[Object]) -> None:
        states.object.set(e.data)

    obj_conn.listen(obj_evt_handler, filter=lambda e: isinstance(e.data, Object))

    _state_init_done = True

print_state()

Source code in roc/jupyter/state.py 
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317
318
319
320
321
322
323
324
325
def print_state() -> None:
    init_state()

    def header(s: str) -> None:
        print(f"\n=== {s.upper()} ===")  # noqa: T201

    header("System Health")
    print(states.cpuload)  # noqa: T201
    print(states.diskio)  # noqa: T201
    print(states.memory)  # noqa: T201
    print(states.sysmem)  # noqa: T201

    header("Environment")
    print(states.loop)  # noqa: T201
    print(states.screen)  # noqa: T201
    # TODO: blstats

    header("Graph DB")
    print(states.node_cache)  # noqa: T201
    print(states.edge_cache)  # noqa: T201

    header("Agent")
    print(states.components)  # noqa: T201
    print(states.salency)  # noqa: T201
    print(states.attention)  # noqa: T201
    print(states.object)  # noqa: T201

state_cli(var)

Source code in roc/jupyter/state.py 
328
329
330
331
332
333
334
335
336
337
338
339
340
341
342
@click.command()
@click.argument(
    "var",
    nargs=-1,
    type=click.Choice(all_states, case_sensitive=False),
)
def state_cli(var: list[str]) -> None:
    if var is None or len(var) < 1:
        # if no state is specified, print a selection of the most interesting states
        print_state()
        return

    for v in var:
        s = getattr(states, v)
        print(str(s))  # noqa: T201