summaryrefslogtreecommitdiff
path: root/src/ipa/raspberrypi/cam_helper_imx477.cpp
blob: 338fdc0c416a35b0a946708afda0e6abde70a9ff (plain) 
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

/* SPDX-License-Identifier: BSD-2-Clause */
/*
 * Copyright (C) 2020, Raspberry Pi (Trading) Limited
 *
 * cam_helper_imx477.cpp - camera helper for imx477 sensor
 */

#include <assert.h>
#include <cmath>
#include <stddef.h>
#include <stdio.h>
#include <stdlib.h>

#include <libcamera/base/log.h>

#include "cam_helper.hpp"
#include "md_parser.hpp"

using namespace RPiController;
using namespace libcamera;
using libcamera::utils::Duration;

namespace libcamera {
LOG_DECLARE_CATEGORY(IPARPI)
}

/*
 * We care about two gain registers and a pair of exposure registers. Their
 * I2C addresses from the Sony IMX477 datasheet:
 */
constexpr uint32_t expHiReg = 0x0202;
constexpr uint32_t expLoReg = 0x0203;
constexpr uint32_t gainHiReg = 0x0204;
constexpr uint32_t gainLoReg = 0x0205;
constexpr uint32_t frameLengthHiReg = 0x0340;
constexpr uint32_t frameLengthLoReg = 0x0341;
constexpr std::initializer_list<uint32_t> registerList =
	{ expHiReg, expLoReg, gainHiReg, gainLoReg, frameLengthHiReg, frameLengthLoReg  };

class CamHelperImx477 : public CamHelper
{
public:
	CamHelperImx477();
	uint32_t GainCode(double gain) const override;
	double Gain(uint32_t gain_code) const override;
	void Prepare(libcamera::Span<const uint8_t> buffer, Metadata &metadata) override;
	uint32_t GetVBlanking(Duration &exposure, Duration minFrameDuration,
			      Duration maxFrameDuration) const override;
	void GetDelays(int &exposure_delay, int &gain_delay,
		       int &vblank_delay) const override;
	bool SensorEmbeddedDataPresent() const override;

private:
	/*
	 * Smallest difference between the frame length and integration time,
	 * in units of lines.
	 */
	static constexpr int frameIntegrationDiff = 22;
	/* Maximum frame length allowable for long exposure calculations. */
	static constexpr int frameLengthMax = 0xffdc;
	/* Largest long exposure scale factor given as a left shift on the frame length. */
	static constexpr int longExposureShiftMax = 7;

	void PopulateMetadata(const MdParser::RegisterMap &registers,
			      Metadata &metadata) const override;
};

CamHelperImx477::CamHelperImx477()
	: CamHelper(std::make_unique<MdParserSmia>(registerList), frameIntegrationDiff)
{
}

uint32_t CamHelperImx477::GainCode(double gain) const
{
	return static_cast<uint32_t>(1024 - 1024 / gain);
}

double CamHelperImx477::Gain(uint32_t gain_code) const
{
	return 1024.0 / (1024 - gain_code);
}

void CamHelperImx477::Prepare(libcamera::Span<const uint8_t> buffer, Metadata &metadata)
{
	MdParser::RegisterMap registers;
	DeviceStatus deviceStatus;

	if (metadata.Get("device.status", deviceStatus)) {
		LOG(IPARPI, Error) << "DeviceStatus not found from DelayedControls";
		return;
	}

	parseEmbeddedData(buffer, metadata);

	/*
	 * The DeviceStatus struct is first populated with values obtained from
	 * DelayedControls. If this reports frame length is > frameLengthMax,
	 * it means we are using a long exposure mode. Since the long exposure
	 * scale factor is not returned back through embedded data, we must rely
	 * on the existing exposure lines and frame length values returned by
	 * DelayedControls.
	 *
	 * Otherwise, all values are updated with what is reported in the
	 * embedded data.
	 */
	if (deviceStatus.frame_length > frameLengthMax) {
		DeviceStatus parsedDeviceStatus;

		metadata.Get("device.status", parsedDeviceStatus);
		parsedDeviceStatus.shutter_speed = deviceStatus.shutter_speed;
		parsedDeviceStatus.frame_length = deviceStatus.frame_length;
		metadata.Set("device.status", parsedDeviceStatus);

		LOG(IPARPI, Debug) << "Metadata updated for long exposure: "
				   << parsedDeviceStatus;
	}
}

uint32_t CamHelperImx477::GetVBlanking(Duration &exposure,
				       Duration minFrameDuration,
				       Duration maxFrameDuration) const
{
	uint32_t frameLength, exposureLines;
	unsigned int shift = 0;

	frameLength = mode_.height + CamHelper::GetVBlanking(exposure, minFrameDuration,
							     maxFrameDuration);
	/*
	 * Check if the frame length calculated needs to be setup for long
	 * exposure mode. This will require us to use a long exposure scale
	 * factor provided by a shift operation in the sensor.
	 */
	while (frameLength > frameLengthMax) {
		if (++shift > longExposureShiftMax) {
			shift = longExposureShiftMax;
			frameLength = frameLengthMax;
			break;
		}
		frameLength >>= 1;
	}

	if (shift) {
		/* Account for any rounding in the scaled frame length value. */
		frameLength <<= shift;
		exposureLines = ExposureLines(exposure);
		exposureLines = std::min(exposureLines, frameLength - frameIntegrationDiff);
		exposure = Exposure(exposureLines);
	}

	return frameLength - mode_.height;
}

void CamHelperImx477::GetDelays(int &exposure_delay, int &gain_delay,
				int &vblank_delay) const
{
	exposure_delay = 2;
	gain_delay = 2;
	vblank_delay = 3;
}

bool CamHelperImx477::SensorEmbeddedDataPresent() const
{
	return true;
}

void CamHelperImx477::PopulateMetadata(const MdParser::RegisterMap &registers,
				       Metadata &metadata) const
{
	DeviceStatus deviceStatus;

	deviceStatus.shutter_speed = Exposure(registers.at(expHiReg) * 256 + registers.at(expLoReg));
	deviceStatus.analogue_gain = Gain(registers.at(gainHiReg) * 256 + registers.at(gainLoReg));
	deviceStatus.frame_length = registers.at(frameLengthHiReg) * 256 + registers.at(frameLengthLoReg);

	metadata.Set("device.status", deviceStatus);
}

static CamHelper *Create()
{
	return new CamHelperImx477();
}

static RegisterCamHelper reg("imx477", &Create);
generated by cgit v1.2.1 (git 2.18.0) at 2025-03-11 02:23:28 +0000
class="hl opt">.diff_header_regex.match(line) if not match: raise RuntimeError("Malformed diff hunk header '%s'" % line) self.__from_line = int(match.group(1)) self.__to_line = int(match.group(3)) self.__from = DiffHunkSide(self.__from_line) self.__to = DiffHunkSide(self.__to_line) self.lines = [] def __repr__(self): s = '%s@@ -%u,%u +%u,%u @@\n' % \ (Colours.fg(Colours.Cyan), self.__from.start, len(self.__from), self.__to.start, len(self.__to)) for line in self.lines: if line[0] == '-': s += Colours.fg(Colours.Red) elif line[0] == '+': s += Colours.fg(Colours.Green) if line[0] == '-': spaces = 0 for i in range(len(line)): if line[-i-1].isspace(): spaces += 1 else: break spaces = len(line) - spaces line = line[0:spaces] + Colours.bg(Colours.Red) + line[spaces:] s += line s += Colours.reset() s += '\n' return s[:-1] def append(self, line): if line[0] == ' ': self.__from.untouched.append(self.__from_line) self.__from_line += 1 self.__to.untouched.append(self.__to_line) self.__to_line += 1 elif line[0] == '-': self.__from.touched.append(self.__from_line) self.__from_line += 1 elif line[0] == '+': self.__to.touched.append(self.__to_line) self.__to_line += 1 self.lines.append(line.rstrip('\n')) def intersects(self, lines): for line in lines: if line in self.__from.touched: return True return False def side(self, side): if side == 'from': return self.__from else: return self.__to def parse_diff(diff): hunks = [] hunk = None for line in diff: if line.startswith('@@'): if hunk: hunks.append(hunk) hunk = DiffHunk(line) elif hunk is not None: hunk.append(line) if hunk: hunks.append(hunk) return hunks # ------------------------------------------------------------------------------ # Style Checkers # _style_checkers = [] class StyleCheckerRegistry(type): def __new__(cls, clsname, bases, attrs): newclass = super(StyleCheckerRegistry, cls).__new__(cls, clsname, bases, attrs) if clsname != 'StyleChecker': _style_checkers.append(newclass) return newclass class StyleChecker(metaclass=StyleCheckerRegistry): def __init__(self): pass # # Class methods # @classmethod def checkers(cls, filename): for checker in _style_checkers: if checker.supports(filename): yield checker @classmethod def supports(cls, filename): for pattern in cls.patterns: if fnmatch.fnmatch(os.path.basename(filename), pattern): return True return False @classmethod def all_patterns(cls): patterns = set() for checker in _style_checkers: patterns.update(checker.patterns) return patterns class StyleIssue(object): def __init__(self, line_number, line, msg): self.line_number = line_number self.line = line self.msg = msg class LogCategoryChecker(StyleChecker): log_regex = re.compile('\\bLOG\((Debug|Info|Warning|Error|Fatal)\)') patterns = ('*.cpp',) def __init__(self, content): super().__init__() self.__content = content def check(self, line_numbers): issues = [] for line_number in line_numbers: line = self.__content[line_number-1] if not LogCategoryChecker.log_regex.search(line): continue issues.append(StyleIssue(line_number, line, 'LOG() should use categories')) return issues class MesonChecker(StyleChecker): patterns = ('meson.build',) def __init__(self, content): super().__init__() self.__content = content def check(self, line_numbers): issues = [] for line_number in line_numbers: line = self.__content[line_number-1] if line.find('\t') != -1: issues.append(StyleIssue(line_number, line, 'meson.build should use spaces for indentation')) return issues class Pep8Checker(StyleChecker): patterns = ('*.py',) results_regex = re.compile('stdin:([0-9]+):([0-9]+)(.*)') def __init__(self, content): super().__init__() self.__content = content def check(self, line_numbers): issues = [] data = ''.join(self.__content).encode('utf-8') try: ret = subprocess.run(['pep8', '--ignore=E501', '-'], input=data, stdout=subprocess.PIPE) except FileNotFoundError: issues.append(StyleIssue(0, None, "Please install pep8 to validate python additions")) return issues results = ret.stdout.decode('utf-8').splitlines() for item in results: search = re.search(Pep8Checker.results_regex, item) line_number = int(search.group(1)) position = int(search.group(2)) msg = search.group(3) if line_number in line_numbers: line = self.__content[line_number - 1] issues.append(StyleIssue(line_number, line, msg)) return issues # ------------------------------------------------------------------------------ # Formatters # _formatters = [] class FormatterRegistry(type): def __new__(cls, clsname, bases, attrs): newclass = super(FormatterRegistry, cls).__new__(cls, clsname, bases, attrs) if clsname != 'Formatter': _formatters.append(newclass) return newclass class Formatter(metaclass=FormatterRegistry): enabled = True def __init__(self): pass # # Class methods # @classmethod def formatters(cls, filename): for formatter in _formatters: if not cls.enabled: continue if formatter.supports(filename): yield formatter @classmethod def supports(cls, filename): if not cls.enabled: return False for pattern in cls.patterns: if fnmatch.fnmatch(os.path.basename(filename), pattern): return True return False @classmethod def all_patterns(cls): patterns = set() for formatter in _formatters: if not cls.enabled: continue patterns.update(formatter.patterns) return patterns class AStyleFormatter(Formatter): enabled = False patterns = ('*.c', '*.cpp', '*.h') @classmethod def format(cls, filename, data): ret = subprocess.run(['astyle', *astyle_options], input=data.encode('utf-8'), stdout=subprocess.PIPE) return ret.stdout.decode('utf-8') class CLangFormatter(Formatter): enabled = False patterns = ('*.c', '*.cpp', '*.h') @classmethod def format(cls, filename, data): ret = subprocess.run(['clang-format', '-style=file', '-assume-filename=' + filename], input=data.encode('utf-8'), stdout=subprocess.PIPE) return ret.stdout.decode('utf-8') class DoxygenFormatter(Formatter): patterns = ('*.c', '*.cpp') return_regex = re.compile(' +\\* +\\\\return +[a-z]') @classmethod def format(cls, filename, data): lines = [] in_doxygen = False for line in data.split('\n'): if line.find('/**') != -1: in_doxygen = True if not in_doxygen: lines.append(line) continue line = cls.return_regex.sub(lambda m: m.group(0)[:-1] + m.group(0)[-1].upper(), line) if line.find('*/') != -1: in_doxygen = False lines.append(line) return '\n'.join(lines) class StripTrailingSpaceFormatter(Formatter): patterns = ('*.c', '*.cpp', '*.h', '*.py', 'meson.build') @classmethod def format(cls, filename, data): lines = data.split('\n') for i in range(len(lines)): lines[i] = lines[i].rstrip() + '\n' return ''.join(lines) # ------------------------------------------------------------------------------ # Style checking # def check_file(top_level, commit, filename): # Extract the line numbers touched by the commit. diff = subprocess.run(['git', 'diff', '%s~..%s' % (commit, commit), '--', '%s/%s' % (top_level, filename)], stdout=subprocess.PIPE).stdout diff = diff.decode('utf-8').splitlines(True) commit_diff = parse_diff(diff) lines = [] for hunk in commit_diff: lines.extend(hunk.side('to').touched) # Skip commits that don't add any line. if len(lines) == 0: return 0 # Format the file after the commit with all formatters and compute the diff # between the unformatted and formatted contents. after = subprocess.run(['git', 'show', '%s:%s' % (commit, filename)], stdout=subprocess.PIPE).stdout after = after.decode('utf-8') formatted = after for formatter in Formatter.formatters(filename): formatted = formatter.format(filename, formatted) after = after.splitlines(True) formatted = formatted.splitlines(True) diff = difflib.unified_diff(after, formatted) # Split the diff in hunks, recording line number ranges for each hunk, and # filter out hunks that are not touched by the commit. formatted_diff = parse_diff(diff) formatted_diff = [hunk for hunk in formatted_diff if hunk.intersects(lines)] # Check for code issues not related to formatting. issues = [] for checker in StyleChecker.checkers(filename): checker = checker(after) for hunk in commit_diff: issues += checker.check(hunk.side('to').touched) # Print the detected issues. if len(issues) == 0 and len(formatted_diff) == 0: return 0 print('%s---' % Colours.fg(Colours.Red), filename) print('%s+++' % Colours.fg(Colours.Green), filename) if len(formatted_diff): for hunk in formatted_diff: print(hunk) if len(issues): issues = sorted(issues, key=lambda i: i.line_number) for issue in issues: print('%s#%u: %s' % (Colours.fg(Colours.Yellow), issue.line_number, issue.msg)) if issue.line is not None: print('+%s%s' % (issue.line.rstrip(), Colours.reset())) return len(formatted_diff) + len(issues) def check_style(top_level, commit): # Get the commit title and list of files. ret = subprocess.run(['git', 'show', '--pretty=oneline','--name-only', commit], stdout=subprocess.PIPE) files = ret.stdout.decode('utf-8').splitlines() title = files[0] files = files[1:] separator = '-' * len(title) print(separator) print(title) print(separator) # Filter out files we have no checker for. patterns = set() patterns.update(StyleChecker.all_patterns()) patterns.update(Formatter.all_patterns()) files = [f for f in files if len([p for p in patterns if fnmatch.fnmatch(os.path.basename(f), p)])] if len(files) == 0: print("Commit doesn't touch source files, skipping") return issues = 0 for f in files: issues += check_file(top_level, commit, f) if issues == 0: print("No style issue detected") else: print('---') print("%u potential style %s detected, please review" % \ (issues, 'issue' if issues == 1 else 'issues')) def extract_revlist(revs): """Extract a list of commits on which to operate from a revision or revision range. """ ret = subprocess.run(['git', 'rev-parse', revs], stdout=subprocess.PIPE, stderr=subprocess.PIPE) if ret.returncode != 0: print(ret.stderr.decode('utf-8').splitlines()[0]) return [] revlist = ret.stdout.decode('utf-8').splitlines() # If the revlist contains more than one item, pass it to git rev-list to list # each commit individually. if len(revlist) > 1: ret = subprocess.run(['git', 'rev-list', *revlist], stdout=subprocess.PIPE) revlist = ret.stdout.decode('utf-8').splitlines() revlist.reverse() return revlist def git_top_level(): """Get the absolute path of the git top-level directory.""" ret = subprocess.run(['git', 'rev-parse', '--show-toplevel'], stdout=subprocess.PIPE, stderr=subprocess.PIPE) if ret.returncode != 0: print(ret.stderr.decode('utf-8').splitlines()[0]) return None return ret.stdout.decode('utf-8').strip() def main(argv): # Parse command line arguments parser = argparse.ArgumentParser() parser.add_argument('--formatter', '-f', type=str, choices=['astyle', 'clang-format'], help='Code formatter. Default to clang-format if not specified.') parser.add_argument('revision_range', type=str, default='HEAD', nargs='?', help='Revision range (as defined by git rev-parse). Defaults to HEAD if not specified.') args = parser.parse_args(argv[1:]) # Check for required dependencies. for command, mandatory in dependencies.items(): found = shutil.which(command) if mandatory and not found: print("Executable %s not found" % command) return 1 dependencies[command] = found if args.formatter: if not args.formatter in dependencies or \ not dependencies[args.formatter]: print("Formatter %s not available" % args.formatter) return 1 formatter = args.formatter else: if dependencies['clang-format']: CLangFormatter.enabled = True elif dependencies['astyle']: AStyleFormatter.enabled = True else: print("No formatter found, please install clang-format or astyle") return 1 # Get the top level directory to pass absolute file names to git diff # commands, in order to support execution from subdirectories of the git # tree. top_level = git_top_level() if top_level is None: return 1 revlist = extract_revlist(args.revision_range) for commit in revlist: check_style(top_level, commit) print('') return 0 if __name__ == '__main__': sys.exit(main(sys.argv))
generated by cgit v1.2.1 (git 2.18.0) at 2025-03-11 02:23:26 +0000