13079399260

Committed 31 Jan 2025 07:22PM UTC coverage: 55.653% (+5.2%) from 50.45%

Build # 13079399260

Build Type

push

github

Committed by

mhasself

Commit Message

Remove: hk "Reframer" class; other references to hkagg_version 0

Run Details

1334 of 2397 relevant lines covered (55.65%)

0.56 hits per line

Source File
Press 'n' to go to next uncovered line, 'b' for previous

94.87

/python/hk/session.py

import so3g
from spt3g import core
import time
import os
import binascii


class HKSessionHelper:
    def __init__(self, session_id=None, start_time=None, hkagg_version=None,
                 description='No description provided.'):
        """Helper class to produce G3Frame templates for creating streams of
        generic HK data.

        Arguments:
          session_id: an integer session ID for the HK session.  If
            not provided (recommended) then it will be generated based
            on the PID, the start_time, and the description string.
          start_time (float): a timestamp to use for the HK session.
          hkagg_version (int): schema version code, which will be
            written into every frame.  Defaults to zero, for backwards
            compatibility.  Make sure this is set correctly.
          description (str): a description of the agent generating the
            stream.

        """
        if start_time is None:
            start_time = time.time()
        self.start_time = start_time
        self.description = description
        self.provs = {}
        self.next_prov_id = 0
        if session_id is None:
            session_id = self._generate_session_id(start_time, description)
        self.session_id = session_id
        if hkagg_version is None:
            hkagg_version = 0
        self.hkagg_version = hkagg_version

    @staticmethod
    def _generate_session_id(timestamp=None, description=''):
        if timestamp is None:
            timestamp = time.time()
        if description is None:
            description = '?'
        # Bit-combine some unique stuff.  It's useful if this sorts in
        # time, so lead off with the timestamp.
        elements = [(int(timestamp), 32),
                    (os.getpid(), 14),
                    (binascii.crc32(bytes(description, 'utf8')), 14)]
        session_id = 0
        for i, b in elements:
            session_id = (session_id << b) | (i % (1 << b))
        return session_id

    def add_provider(self, description=None):
        """Register a provider and return the unique prov_id.  (Remember to
        write a status frame before starting to write data for this
        provider.)

        Args:
          description (str): The name to use for the provider.  When
            later retrieving the data, this will act as a prefix for
            all the data fields.

        Returns:
          prov_id (int).
        """
        assert(description is not None) # Need a provider name!
        prov_id = self.next_prov_id
        self.next_prov_id += 1
        self.provs[prov_id] = {'description': description}
        return prov_id

    def remove_provider(self, prov_id):
        """Drops a provider from the active list, so it will not be listed in
        the status frame.

        Args:
          prov_id (int): The provider ID returned by add_provider.
        """
        del self.provs[prov_id]

    """
    Frame generators.
    """

    def session_frame(self):
        """
        Return the Session frame.  No additional information needs to be
        added to this frame.  This frame initializes the HK stream (so
        it should be the first frame of each HK file; it should
        precede all other HK frames in a network source).
        """
        f = core.G3Frame()
        f.type = core.G3FrameType.Housekeeping
        f['hkagg_type'] = so3g.HKFrameType.session
        f['hkagg_version'] = self.hkagg_version
        f['session_id'] = self.session_id
        f['start_time'] = self.start_time
        f['description'] = self.description
        return f

    def status_frame(self, timestamp=None):
        """
        Return a Status frame template.  Before processing, the session
        manager should update with information about what providers
        are currently connected.
        """
        if timestamp is None:
            timestamp = time.time()
        f = core.G3Frame()
        f.type = core.G3FrameType.Housekeeping
        f['hkagg_type'] = so3g.HKFrameType.status
        f['hkagg_version'] = self.hkagg_version
        f['session_id'] = self.session_id
        f['timestamp'] = timestamp
        provs = core.G3VectorFrameObject()
        for prov_id in sorted(self.provs.keys()):
            prov = core.G3MapFrameObject()
            prov['prov_id'] = core.G3Int(prov_id)
            prov['description'] = core.G3String(
                self.provs[prov_id]['description'])
            provs.append(prov)
        f['providers'] = provs
        return f

    def data_frame(self, prov_id, timestamp=None):
        """
        Return a Data frame template.  The prov_id must match the prov_id
        in one of the Provider blocks in the preceding status frame.
        The session manager should create and add IrregBlockDouble items
        to the 'blocks' list.
        """
        if timestamp is None:
            timestamp = time.time()
        f = core.G3Frame()
        f.type = core.G3FrameType.Housekeeping
        f['hkagg_version'] = self.hkagg_version
        f['hkagg_type'] = so3g.HKFrameType.data
        f['session_id'] = self.session_id
        f['prov_id'] = prov_id
        f['timestamp'] = timestamp
        f['blocks'] = core.G3VectorFrameObject()
        if self.hkagg_version >= 2:
            f['block_names'] = core.G3VectorString()
        return f

1	import so3g	1✔
2	from spt3g import core	1✔
3	import time	1✔
4	import os	1✔
5	import binascii	1✔
6
7
8	class HKSessionHelper:	1✔
9	def __init__(self, session_id=None, start_time=None, hkagg_version=None,	1✔
10	description='No description provided.'):
11	"""Helper class to produce G3Frame templates for creating streams of
12	generic HK data.
13
14	Arguments:
15	session_id: an integer session ID for the HK session. If
16	not provided (recommended) then it will be generated based
17	on the PID, the start_time, and the description string.
18	start_time (float): a timestamp to use for the HK session.
19	hkagg_version (int): schema version code, which will be
20	written into every frame. Defaults to zero, for backwards
21	compatibility. Make sure this is set correctly.
22	description (str): a description of the agent generating the
23	stream.
24
25	"""
26	if start_time is None:	1✔
27	start_time = time.time()	1✔
28	self.start_time = start_time	1✔
29	self.description = description	1✔
30	self.provs = {}	1✔
31	self.next_prov_id = 0	1✔
32	if session_id is None:	1✔
33	session_id = self._generate_session_id(start_time, description)	1✔
34	self.session_id = session_id	1✔
35	if hkagg_version is None:	1✔
36	hkagg_version = 0	×
37	self.hkagg_version = hkagg_version	1✔
38
39	@staticmethod	1✔
40	def _generate_session_id(timestamp=None, description=''):	1✔
41	if timestamp is None:	1✔
42	timestamp = time.time()	×
43	if description is None:	1✔
44	description = '?'	×
45	# Bit-combine some unique stuff. It's useful if this sorts in
46	# time, so lead off with the timestamp.
47	elements = [(int(timestamp), 32),	1✔
48	(os.getpid(), 14),
49	(binascii.crc32(bytes(description, 'utf8')), 14)]
50	session_id = 0	1✔
51	for i, b in elements:	1✔
52	session_id = (session_id << b) \| (i % (1 << b))	1✔
53	return session_id	1✔
54
55	def add_provider(self, description=None):	1✔
56	"""Register a provider and return the unique prov_id. (Remember to
57	write a status frame before starting to write data for this
58	provider.)
59
60	Args:
61	description (str): The name to use for the provider. When
62	later retrieving the data, this will act as a prefix for
63	all the data fields.
64
65	Returns:
66	prov_id (int).
67	"""
68	assert(description is not None) # Need a provider name!	1✔
69	prov_id = self.next_prov_id	1✔
70	self.next_prov_id += 1	1✔
71	self.provs[prov_id] = {'description': description}	1✔
72	return prov_id	1✔
73
74	def remove_provider(self, prov_id):	1✔
75	"""Drops a provider from the active list, so it will not be listed in
76	the status frame.
77
78	Args:
79	prov_id (int): The provider ID returned by add_provider.
80	"""
81	del self.provs[prov_id]	×
82
83	"""
84	Frame generators.
85	"""
86
87	def session_frame(self):	1✔
88	"""
89	Return the Session frame. No additional information needs to be
90	added to this frame. This frame initializes the HK stream (so
91	it should be the first frame of each HK file; it should
92	precede all other HK frames in a network source).
93	"""
94	f = core.G3Frame()	1✔
95	f.type = core.G3FrameType.Housekeeping	1✔
96	f['hkagg_type'] = so3g.HKFrameType.session	1✔
97	f['hkagg_version'] = self.hkagg_version	1✔
98	f['session_id'] = self.session_id	1✔
99	f['start_time'] = self.start_time	1✔
100	f['description'] = self.description	1✔
101	return f	1✔
102
103	def status_frame(self, timestamp=None):	1✔
104	"""
105	Return a Status frame template. Before processing, the session
106	manager should update with information about what providers
107	are currently connected.
108	"""
109	if timestamp is None:	1✔
110	timestamp = time.time()	1✔
111	f = core.G3Frame()	1✔
112	f.type = core.G3FrameType.Housekeeping	1✔
113	f['hkagg_type'] = so3g.HKFrameType.status	1✔
114	f['hkagg_version'] = self.hkagg_version	1✔
115	f['session_id'] = self.session_id	1✔
116	f['timestamp'] = timestamp	1✔
117	provs = core.G3VectorFrameObject()	1✔
118	for prov_id in sorted(self.provs.keys()):	1✔
119	prov = core.G3MapFrameObject()	1✔
120	prov['prov_id'] = core.G3Int(prov_id)	1✔
121	prov['description'] = core.G3String(	1✔
122	self.provs[prov_id]['description'])
123	provs.append(prov)	1✔
124	f['providers'] = provs	1✔
125	return f	1✔
126
127	def data_frame(self, prov_id, timestamp=None):	1✔
128	"""
129	Return a Data frame template. The prov_id must match the prov_id
130	in one of the Provider blocks in the preceding status frame.
131	The session manager should create and add IrregBlockDouble items
132	to the 'blocks' list.
133	"""
134	if timestamp is None:	1✔
135	timestamp = time.time()	1✔
136	f = core.G3Frame()	1✔
137	f.type = core.G3FrameType.Housekeeping	1✔
138	f['hkagg_version'] = self.hkagg_version	1✔
139	f['hkagg_type'] = so3g.HKFrameType.data	1✔
140	f['session_id'] = self.session_id	1✔
141	f['prov_id'] = prov_id	1✔
142	f['timestamp'] = timestamp	1✔
143	f['blocks'] = core.G3VectorFrameObject()	1✔
144	if self.hkagg_version >= 2:	1✔
145	f['block_names'] = core.G3VectorString()	1✔
146	return f	1✔

simonsobs / so3g / 13079399260

Source File Press 'n' to go to next uncovered line, 'b' for previous

Source File
Press 'n' to go to next uncovered line, 'b' for previous